ligo.at
decentralized and customizable links page on top of atproto
ligo.at
atproto
link-in-bio
python
uv
very basic jetstream ingestor
nauta.one
+4
Makefile
+4
Makefile
+94
src/ingest.py
+94
src/ingest.py
···
1
+
import sqlite3
2
+
from typing import Any
3
+
import aiohttp
4
+
import asyncio
5
+
import dotenv
6
+
import json
7
+
import logging
8
+
9
+
logger = logging.getLogger(__name__)
10
+
11
+
12
+
async def ingest_jetstream(config: dict[str, str | None]):
13
+
socket = f"wss://{config['JETSTREAM_URL']}/subscribe"
14
+
socket += "?wantedCollections=at.ligo.*"
15
+
logger.info(f"connecting to {socket}")
16
+
async with aiohttp.ClientSession() as session:
17
+
async with session.ws_connect(socket) as ws:
18
+
async for message in ws:
19
+
if message.type == aiohttp.WSMsgType.TEXT:
20
+
json = message.json()
21
+
did = json["did"]
22
+
if json["kind"] == "commit":
23
+
handle_commit(did, json["commit"], config)
24
+
else:
25
+
continue
26
+
27
+
28
+
def handle_commit(did: str, commit: dict[str, Any], config: dict[str, str | None]):
29
+
is_delete: bool = commit["operation"] == "delete"
30
+
collection: str = commit["collection"]
31
+
rkey: str = commit["rkey"]
32
+
33
+
if rkey != "self":
34
+
return
35
+
36
+
db = get_database(config)
37
+
if not db:
38
+
return
39
+
cursor = db.cursor()
40
+
41
+
prefix: str | None = None
42
+
type: str | None = None
43
+
match collection:
44
+
case "at.ligo.actor.profile":
45
+
prefix = "profile_from_did"
46
+
type = "at.ligo.actor.profile"
47
+
case "at.ligo.actor.links":
48
+
prefix = "links_from_did"
49
+
type = "at.ligo.actor.links"
50
+
case _:
51
+
pass
52
+
if prefix is None:
53
+
return
54
+
55
+
if is_delete:
56
+
logger.debug(f"deleting {prefix} for {did}")
57
+
_ = cursor.execute(
58
+
"delete from keyval where prefix = ? and key = ?",
59
+
(prefix, did),
60
+
)
61
+
else:
62
+
logger.debug(f"creating or updating {prefix} for {did}")
63
+
record: dict[str, str] = commit["record"]
64
+
if record["$type"] != type:
65
+
return
66
+
content = json.dumps(record)
67
+
_ = cursor.execute(
68
+
"insert or replace into keyval values (?, ?, ?)",
69
+
(prefix, did, content),
70
+
)
71
+
72
+
db.commit()
73
+
cursor.close()
74
+
db.close()
75
+
76
+
77
+
def get_database(config: dict[str, str | None]) -> sqlite3.Connection | None:
78
+
database_name = config.get("FLASK_DATABASE_URL", "ligoat.db")
79
+
if not database_name:
80
+
return None
81
+
return sqlite3.connect(database_name)
82
+
83
+
84
+
async def main(config: dict[str, str | None]):
85
+
try:
86
+
await ingest_jetstream(config)
87
+
except asyncio.CancelledError:
88
+
pass
89
+
90
+
91
+
if __name__ == "__main__":
92
+
config = dotenv.dotenv_values()
93
+
asyncio.run(main(config))
94
+
print("see you next time!")