this repo has no description
edavis.dev
1from datetime import datetime, timezone, timedelta
2
3class BaseFeed:
4 def process_commit(self, commit):
5 raise NotImplementedError
6
7 def serve_feed(self, limit, offset, langs):
8 raise NotImplementedError
9
10 def serve_wildcard_feed(self, feed_uri, limit, offset, langs):
11 raise NotImplementedError
12
13 def commit_changes(self):
14 raise NotImplementedError
15
16 def parse_timestamp(self, timestamp):
17 # https://atproto.com/specs/lexicon#datetime
18 formats = {
19 # preferred
20 '1985-04-12T23:20:50.123Z': '%Y-%m-%dT%H:%M:%S.%f%z',
21 # '1985-04-12T23:20:50.123456Z': '%Y-%m-%dT%H:%M:%S.%f%z',
22 # '1985-04-12T23:20:50.120Z': '%Y-%m-%dT%H:%M:%S.%f%z',
23 # '1985-04-12T23:20:50.120000Z': '%Y-%m-%dT%H:%M:%S.%f%z',
24
25 # supported
26 # '1985-04-12T23:20:50.12345678912345Z': '',
27 '1985-04-12T23:20:50Z': '%Y-%m-%dT%H:%M:%S%z',
28 # '1985-04-12T23:20:50.0Z': '%Y-%m-%dT%H:%M:%S.%f%z',
29 # '1985-04-12T23:20:50.123+00:00': '%Y-%m-%dT%H:%M:%S.%f%z',
30 # '1985-04-12T23:20:50.123-07:00': '%Y-%m-%dT%H:%M:%S.%f%z',
31 }
32
33 for format in formats.values():
34 try:
35 ts = datetime.strptime(timestamp, format)
36 except ValueError:
37 continue
38 else:
39 return ts
40
41 return datetime.now(timezone.utc)
42
43 def safe_timestamp(self, timestamp):
44 utc_now = datetime.now(timezone.utc)
45 if timestamp is None:
46 return utc_now
47
48 parsed = self.parse_timestamp(timestamp)
49 if parsed.timestamp() <= 0:
50 return utc_now
51 elif parsed - timedelta(minutes=2) < utc_now:
52 return parsed
53 elif parsed > utc_now:
54 return utc_now
55
56 def transaction_begin(self, db):
57 if not db.in_transaction:
58 db.execute('BEGIN')
59
60 def transaction_commit(self, db):
61 if db.in_transaction:
62 db.execute('COMMIT')
63
64 def wal_checkpoint(self, db, mode='PASSIVE'):
65 return db.execute(f'PRAGMA wal_checkpoint({mode})')