Total Complexity | 6 |
Total Lines | 41 |
Duplicated Lines | 0 % |
Changes | 1 | ||
Bugs | 0 | Features | 0 |
1 | from .model import * |
||
5 | class BadScan(Base): |
||
6 | __tablename__ = 'badscans' |
||
7 | |||
8 | id = Column(Integer, primary_key=True, nullable=False) |
||
9 | timestamp = Column(ArrowType, nullable=False, default=datetime.datetime.utcnow) |
||
10 | badscan = Column(String(255), nullable=False) |
||
11 | |||
12 | def __init__(self, scan): |
||
13 | self.badscan = scan[0:255] |
||
14 | |||
15 | @classmethod |
||
16 | def from_id(cls, id): |
||
17 | return DBSession.query(cls).filter(cls.id == id).one() |
||
18 | |||
19 | @classmethod |
||
20 | def all(cls): |
||
21 | return DBSession.query(cls)\ |
||
22 | .order_by(cls.timestamp).all() |
||
23 | |||
24 | @classmethod |
||
25 | def get_scans_with_counts(cls): |
||
26 | out = [] |
||
27 | badscans = DBSession.query(cls.badscan, func.count(cls.badscan).label('count'))\ |
||
28 | .group_by(cls.badscan)\ |
||
29 | .all() |
||
30 | for badscan in badscans: |
||
31 | most_recent = DBSession.query(cls)\ |
||
32 | .filter(cls.badscan==badscan.badscan)\ |
||
33 | .order_by(cls.timestamp.desc())\ |
||
34 | .first() |
||
35 | out.append({ |
||
36 | 'badscan': badscan.badscan, |
||
37 | 'count': badscan.count, |
||
38 | 'timestamp': most_recent.timestamp |
||
39 | }) |
||
40 | |||
41 | return out |
||
42 | |||
43 | @classmethod |
||
44 | def delete_scans(cls, badscan): |
||
45 | DBSession.query(cls).filter(cls.badscan==badscan).delete() |
||
46 |