Stream activity from Mastodon into a SQLite database

README.md

Stream Mastodon activity into a SQLite database

This script subscribes to the live HTTP feed of public activity on my Mastodon instance and writes the results into SQLite database tables.

It needs sqlite-utils and httpx:

pip install sqlite-utils httpx

Then run:

python stream.py stream.db

If you want to point it at a different Mastodon instance you'll need to edit the script!

Here's the schema after I ran it for a while:

% sqlite-utils schema stream.db
CREATE TABLE [update] (
   [id] TEXT PRIMARY KEY,
   [created_at] TEXT,
   [in_reply_to_id] TEXT,
   [in_reply_to_account_id] TEXT,
   [sensitive] INTEGER,
   [spoiler_text] TEXT,
   [visibility] TEXT,
   [language] TEXT,
   [uri] TEXT,
   [url] TEXT,
   [replies_count] INTEGER,
   [reblogs_count] INTEGER,
   [favourites_count] INTEGER,
   [edited_at] TEXT,
   [content] TEXT,
   [reblog] TEXT,
   [account] TEXT,
   [media_attachments] TEXT,
   [mentions] TEXT,
   [tags] TEXT,
   [emojis] TEXT,
   [card] TEXT,
   [poll] TEXT
);

stream.py

import httpx
import asyncio
import json
import sqlite_utils
import sys


async def run(db):
    client = httpx.AsyncClient()
    async with client.stream(
        "GET", "https://fedi.simonwillison.net/api/v1/streaming/public", timeout=None
    ) as response:
        event_type = None
        async for line in response.aiter_lines():
            line = line.strip()
            if not line:
                continue
            if line.startswith("event:"):
                event_type = line.split(":", 1)[1].strip()
                continue
            if line.startswith("data:"):
                data = line.split(":", 1)[1]
                decoded = json.loads(data)
                if not isinstance(decoded, dict):
                    print("event_type", event_type, "data", data, " (not a dict)")
                    continue
                print(event_type)
                print(json.dumps(decoded, indent=2))
                db[event_type].insert(decoded, pk="id", alter=True)
                continue


if __name__ == "__main__":
    db = sqlite_utils.Database(sys.argv[1])
    db.enable_wal()
    asyncio.run(run(db))

Unsure why conflicting ids are being inserted, could be edits.

Workaround to keep the script going:

import sqlite3

...

try:
    db[event_type].insert(decoded, pk="id", alter=True)
    continue
except sqlite3.IntegrityError as exc:
    continue

You can fix that by adding replace=True to the .insert() call - that will cause it to replace existing records with the same ID.