ligo.at
decentralized and customizable links page on top of atproto
ligo.at
atproto
link-in-bio
python
uv
shut up ty linter
nauta.one
+4
-5
ingestor/__main__.py
+4
-5
ingestor/__main__.py
···
1
-
import sqlite3
2
1
import asyncio
3
-
import dotenv
4
2
import json
5
3
import logging
4
+
import sqlite3
5
+
6
+
import dotenv
6
7
from atproto_jetstream import Jetstream, JetstreamCommitEvent, JetstreamOptions
7
8
8
9
logger = logging.getLogger(__name__)
···
22
23
commit: JetstreamCommitEvent.Commit,
23
24
config: dict[str, str | None],
24
25
):
25
-
is_delete: bool = commit.operation == "delete"
26
-
27
26
if commit.rkey != "self":
28
27
return
29
28
···
46
45
if prefix is None:
47
46
return
48
47
49
-
if is_delete:
48
+
if isinstance(commit, JetstreamCommitEvent.DeleteCommit):
50
49
logger.debug(f"deleting {prefix} for {did}")
51
50
_ = cursor.execute(
52
51
"delete from keyval where prefix = ? and key = ?",