Building a GraphQL Video Discovery API with Strawberry and FastAPI on SQLite

Eleven variants of /trending and none of them fit

We ingest viral video signals from 14 European markets and republish ranked lists — per country, per category, per 6-hour velocity window. The public site is PHP 8.4 on LiteSpeed with SQLite in WAL mode behind Cloudflare Workers, and that stack is genuinely fine for serving HTML. What broke was everything that was not HTML.

Over eighteen months we accumulated /api/trending, /api/trending/compact, /api/trending/embed, /api/market/{cc}/trending, plus a v2 of three of those. Each one existed because some consumer needed three fields fewer or one join more than the last. The embed widget pulled 38 KB of JSON to render six thumbnails and a title. The internal trend dashboard fired nine sequential requests to build one view.

The thing that actually forced a decision was neither of those. Every new column on the video record meant auditing eleven serializers to confirm we had not just leaked a region code derived from a request IP into a public payload. Since ViralVidVault sells itself on GDPR-compliant analytics, an accidental field in an accidental endpoint is not a papercut, it is the whole product's credibility. Eleven hand-written serializers is eleven places to get that wrong.

So we put a read-only GraphQL API in front of the same SQLite file, written in Python with Strawberry and FastAPI. One schema, one place where a field becomes public, one audit surface. Below is the design that survived six months in production, plus the three things that bit us.

Why a second runtime instead of a PHP GraphQL server

webonyx/graphql-php is a solid library and staying single-runtime would have been the boring, correct-sounding choice. We went Python anyway, for reasons that were specific rather than ideological:

• The ranking code was already Python. Velocity scoring, decay curves, and the market normalisation live in a Python batch job. Reusing those dataclasses as the schema source of truth removed a translation layer.
• DataLoader wants an event loop. Batching nested resolvers is the single biggest performance lever in GraphQL, and doing it well in synchronous PHP-FPM means either fibers or a lot of manual pre-fetching.
• Code-first typing. Strawberry builds the schema from annotated Python classes. mypy catches a resolver returning the wrong shape before the SDL is ever generated.
• Process isolation as a safety property. The Python service opens the database mode=ro and sets PRAGMA query_only. It is structurally incapable of corrupting the site's data, which made the deploy risk close to zero.

The cost is real: two runtimes, two deploy pipelines, two dependency audits. We made that acceptable by scoping hard. The GraphQL service is read-only, stateless, and serves no HTML. If it falls over, the site does not notice. That is the only reason a second runtime is worth it.

Model the queries, not the tables

The most common GraphQL failure I see is a schema that is a 1:1 mirror of the database, which gives you an ORM over HTTP and all of the coupling that implies. We started from the four things consumers actually ask for: a ranked page of videos for a market, one video by id, the channel behind a video, and a velocity time series.

Two deliberate omissions shaped everything after. There is no viewer type and no user type — the API is anonymous, so there is no session to model. And Channel.subscriberCount is exposed as a bucket string, not an integer, because exact counts on small channels combined with a market filter get uncomfortably close to identifying a specific creator's performance data.

# schema.py
from __future__ import annotations

from datetime import datetime
from enum import Enum

import strawberry


@strawberry.enum
class Market(Enum):
    DE = 'DE'
    FR = 'FR'
    ES = 'ES'
    IT = 'IT'
    NL = 'NL'
    PL = 'PL'
    SE = 'SE'
    GB = 'GB'


@strawberry.enum
class Window(Enum):
    H6 = 'h6'
    H24 = 'h24'
    D7 = 'd7'


@strawberry.type
class VelocityPoint:
    at: datetime
    views: int
    delta_per_hour: float


@strawberry.type
class Channel:
    id: strawberry.ID
    name: str
    market: Market
    subscriber_bucket: str  # '10k-50k', never an exact figure


@strawberry.type
class Video:
    id: strawberry.ID
    title: str
    published_at: datetime
    duration_s: int
    thumb: str
    score: float
    # Private fields exist on the Python object but never reach the SDL.
    channel_id: strawberry.Private[str]

    @strawberry.field
    async def channel(self, info: strawberry.Info) -> Channel | None:
        return await info.context['loaders'].channel.load(self.channel_id)

    @strawberry.field
    async def velocity(
        self, info: strawberry.Info, window: Window = Window.H24
    ) -> list[VelocityPoint]:
        return await info.context['loaders'].velocity.load((str(self.id), window.value))


@strawberry.type
class VideoEdge:
    cursor: str
    node: Video


@strawberry.type
class PageInfo:
    has_next_page: bool
    end_cursor: str | None


@strawberry.type
class VideoConnection:
    edges: list[VideoEdge]
    page_info: PageInfo
    rank_epoch: int  # which ranking snapshot this page was cut from


@strawberry.type
class Query:
    @strawberry.field(description='Ranked viral videos for one market.')
    async def trending(
        self,
        info: strawberry.Info,
        market: Market,
        window: Window = Window.H24,
        first: int = 25,
        after: str | None = None,
    ) -> VideoConnection:
        return await trending_page(
            info.context['db'], market.value, window.value, min(first, 100), after
        )

    @strawberry.field
    async def video(self, info: strawberry.Info, id: strawberry.ID) -> Video | None:
        return await info.context['loaders'].video.load(str(id))

strawberry.Private is doing quiet but important work here. channel_id is needed by the resolver and must never appear in the public schema. Declaring it private means the leak is impossible rather than merely unlikely, which is the whole reason we moved off hand-written serializers.

Wiring Strawberry into FastAPI over a read-only SQLite pool

SQLite in WAL mode allows one writer and many concurrent readers without the readers blocking. That is exactly the shape of this workload: a PHP cron writes ranking snapshots every few hours, and the API only ever reads.

A few things are not obvious the first time:

• journal_mode = WAL is a persistent property of the database file, set once by the writer. A read-only connection cannot set it and will error if it tries.
• Opening a connection and running pragmas costs roughly 0.4 ms. Trivial once, wasteful at a few hundred requests per second, so we keep a small fixed pool.
• Long-lived read connections pin the WAL. SQLite starts an implicit read transaction on the first query and holds that snapshot until the transaction ends. A pooled connection that never commits or rolls back keeps an old snapshot alive, which blocks WAL checkpointing and lets the -wal file grow without bound. We learned this when a 40 MB database grew a 900 MB WAL over a weekend. The fix is one line in the pool's release path.

# db.py
import asyncio
from contextlib import asynccontextmanager

import aiosqlite

DB_URI = 'file:/var/www/vvv/data/vvv.sqlite?mode=ro'


class ReadPool:
    def __init__(self, uri: str, size: int = 8) -> None:
        self._uri = uri
        self._size = size
        self._q: asyncio.Queue = asyncio.Queue(maxsize=size)

    async def open(self) -> None:
        for _ in range(self._size):
            conn = await aiosqlite.connect(self._uri, uri=True)
            conn.row_factory = aiosqlite.Row
            await conn.execute('PRAGMA query_only = ON')
            await conn.execute('PRAGMA busy_timeout = 2000')
            await conn.execute('PRAGMA cache_size = -32000')      # 32 MB page cache
            await conn.execute('PRAGMA mmap_size = 268435456')    # 256 MB mmap
            self._q.put_nowait(conn)

    @asynccontextmanager
    async def acquire(self):
        conn = await self._q.get()
        try:
            yield conn
        finally:
            # Ends the implicit read transaction. Without this the connection
            # pins an old WAL snapshot and checkpointing stalls forever.
            await conn.rollback()
            self._q.put_nowait(conn)

    async def close(self) -> None:
        while not self._q.empty():
            await self._q.get_nowait().close()


# main.py
from fastapi import FastAPI
from strawberry.fastapi import GraphQLRouter

from .loaders import Loaders
from .schema import schema

pool = ReadPool(DB_URI)


@asynccontextmanager
async def lifespan(app: FastAPI):
    await pool.open()
    yield
    await pool.close()


async def get_context():
    # One connection and one fresh loader set per HTTP request.
    async with pool.acquire() as conn:
        yield {'db': conn, 'loaders': Loaders(conn)}


app = FastAPI(lifespan=lifespan)
app.include_router(
    GraphQLRouter(schema, context_getter=get_context, graphql_ide=None),
    prefix='/graphql',
)

context_getter is resolved as a normal FastAPI dependency, so an async generator works and the connection is returned to the pool when the response finishes. Building Loaders per request rather than per process matters: DataLoader keeps a per-instance memo cache, and a process-lifetime cache would happily serve last Tuesday's ranking.

The N+1 that WAL hides from you

SQLite is fast enough locally to disguise bad query patterns. A page of 200 videos issuing 200 single-row channel lookups at ~30 µs each is 6 ms, which nobody notices in a flame graph. Then the dashboard asks for a 24-point velocity series per video and the same shape costs 200 range scans, and now you are at 340 ms and blaming Python.

DataLoader collapses both into one query per field per request. Two rules matter: the loader must return results in the same order as the requested keys, and the batch must be capped below SQLite's bound-parameter limit (32766 in modern builds, but 200 is a saner batch anyway).

This file also holds the pagination logic, because the cursor and the batch loader share the same snapshot concept.

# loaders.py
import base64
import json

from strawberry.dataloader import DataLoader

from .schema import Channel, Market, Video, VideoConnection, VideoEdge, PageInfo


def encode_cursor(score: float, video_id: str, epoch: int) -> str:
    raw = json.dumps([round(score, 6), video_id, epoch], separators=(',', ':'))
    return base64.urlsafe_b64encode(raw.encode()).decode().rstrip('=')


def decode_cursor(cur: str) -> tuple[float, str, int]:
    pad = '=' * (-len(cur) % 4)
    score, video_id, epoch = json.loads(base64.urlsafe_b64decode(cur + pad))
    return float(score), str(video_id), int(epoch)


async def trending_page(conn, market: str, window: str, first: int, after: str | None):
    keyset, params = '', [market, window]

    if after:
        score, vid, epoch = decode_cursor(after)
        # Pin the caller to the snapshot they started paging on.
        keyset = 'AND (r.score < ? OR (r.score = ? AND r.video_id > ?))'
    else:
        row = await (await conn.execute(
            'SELECT MAX(rank_epoch) AS e FROM rank_snapshot WHERE market = ? AND window = ?',
            (market, window),
        )).fetchone()
        epoch = row['e']

    params.append(epoch)
    if after:
        params += [score, score, vid]
    params.append(first + 1)

    sql = f'''
        SELECT v.id, v.title, v.published_at, v.duration_s, v.thumb, v.channel_id, r.score
          FROM rank_snapshot r
          JOIN video v ON v.id = r.video_id
         WHERE r.market = ? AND r.window = ? AND r.rank_epoch = ? {keyset}
         ORDER BY r.score DESC, r.video_id ASC
         LIMIT ?
    '''
    rows = await (await conn.execute(sql, params)).fetchall()

    has_next = len(rows) > first
    rows = rows[:first]
    edges = [
        VideoEdge(
            cursor=encode_cursor(r['score'], r['id'], epoch),
            node=Video(
                id=r['id'], title=r['title'], published_at=r['published_at'],
                duration_s=r['duration_s'], thumb=r['thumb'],
                score=r['score'], channel_id=r['channel_id'],
            ),
        )
        for r in rows
    ]
    return VideoConnection(
        edges=edges,
        page_info=PageInfo(
            has_next_page=has_next,
            end_cursor=edges[-1].cursor if edges else None,
        ),
        rank_epoch=epoch,
    )


async def load_channels(conn, ids: list[str]) -> list[Channel | None]:
    placeholders = ','.join('?' * len(ids))
    rows = await (await conn.execute(
        f'SELECT id, name, market, subscriber_bucket FROM channel WHERE id IN ({placeholders})',
        ids,
    )).fetchall()
    by_id = {
        r['id']: Channel(
            id=r['id'], name=r['name'],
            market=Market(r['market']), subscriber_bucket=r['subscriber_bucket'],
        )
        for r in rows
    }
    # Order must mirror the requested keys, gaps become None.
    return [by_id.get(i) for i in ids]


class Loaders:
    def __init__(self, conn) -> None:
        self.channel = DataLoader(
            load_fn=lambda keys: load_channels(conn, keys), max_batch_size=200
        )
        self.velocity = DataLoader(
            load_fn=lambda keys: load_velocity(conn, keys), max_batch_size=200
        )
        self.video = DataLoader(load_fn=lambda keys: load_videos(conn, keys))

Why the cursor carries the epoch

OFFSET pagination against a table that re-ranks on a cron is quietly broken. A reader on page 3 when the 06:00 job lands will see items they already saw and miss items entirely, because the offsets now point at a different ordering. Keyset pagination on (score DESC, video_id ASC) fixes duplicates within a stable ordering, but not the re-rank itself.

So the cursor encodes the rank_epoch it was cut from, and subsequent pages are served from that same snapshot. A paging session stays internally consistent even if the ranking changes underneath it, and VideoConnection.rank_epoch tells the client which snapshot it is looking at so it can decide whether to restart from page one.

The index that makes all of this a single scan:

• CREATE INDEX idx_rank_page ON rank_snapshot(market, window, rank_epoch, score DESC, video_id);
• Check it with EXPLAIN QUERY PLAN; you want SEARCH r USING INDEX idx_rank_page and no USE TEMP B-TREE FOR ORDER BY.

The write side stays in PHP

SQLite tolerates exactly one writer, so we kept the writer where it already lived. The cron computes scores in PHP 8.4, inserts a whole new epoch in one transaction, and prunes old ones — keeping two epochs alive so readers mid-page do not fall off a cliff.

<?php
declare(strict_types=1);

final class RankSnapshotWriter
{
    public function __construct(private readonly PDO $db) {}

    /** @param array<string,float> $scored video_id => score */
    public function publish(string $market, string $window, array $scored): int
    {
        $epoch = time();
        $this->db->beginTransaction();

        try {
            $ins = $this->db->prepare(
                'INSERT INTO rank_snapshot (market, window, rank_epoch, video_id, score)
                 VALUES (?, ?, ?, ?, ?)'
            );
            foreach ($scored as $videoId => $score) {
                $ins->execute([$market, $window, $epoch, $videoId, $score]);
            }

            // Keep the live epoch plus the previous one, so in-flight cursors resolve.
            $prune = $this->db->prepare(<<<'SQL'
                DELETE FROM rank_snapshot
                 WHERE market = ? AND window = ?
                   AND rank_epoch NOT IN (
                       SELECT rank_epoch FROM (
                           SELECT DISTINCT rank_epoch FROM rank_snapshot
                            WHERE market = ? AND window = ?
                            ORDER BY rank_epoch DESC LIMIT 2
                       )
                   )
                SQL);
            $prune->execute([$market, $window, $market, $window]);

            $this->db->commit();
        } catch (Throwable $e) {
            $this->db->rollBack();
            throw $e;
        }

        return $epoch;
    }
}

$pdo = new PDO('sqlite:/var/www/vvv/data/vvv.sqlite', options: [
    PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION,
]);
$pdo->exec('PRAGMA journal_mode = WAL');
$pdo->exec('PRAGMA synchronous = NORMAL');
$pdo->exec('PRAGMA wal_autocheckpoint = 512');

// PHP 8.4: new expressions no longer need wrapping parentheses.
$epoch = new RankSnapshotWriter($pdo)->publish('DE', 'h24', $scores);
fprintf(STDERR, 'published epoch %d (%d rows)%s', $epoch, count($scores), PHP_EOL);

Because the insert is one transaction, readers either see the whole new epoch or none of it. There is no window where a page is half old and half new. That is the property that makes the epoch-pinned cursor honest.

GraphQL is a denial-of-service footgun by default

A public GraphQL endpoint lets a stranger write your queries. Three controls, in the order they earn their keep:

• Static cost analysis before execution. Depth limits alone are not enough — trending(first: 100) { velocity } is shallow and expensive.
• Persisted queries in production. Clients send a SHA-256 hash of a known document, not arbitrary text. This turns the whole class of hostile-query problems into a lookup miss, and as a bonus makes the request a cacheable GET.
• Introspection off in production. Not security, but it removes the free schema map.

# security.py
import hashlib
from pathlib import Path

from fastapi import HTTPException
from graphql import GraphQLError
from graphql.validation import ValidationRule
from strawberry.extensions import AddValidationRules

FIELD_COST = {'trending': 2, 'velocity': 5, 'channel': 1, 'video': 1}
MAX_COST = 1000


class CostLimit(ValidationRule):
    def __init__(self, context) -> None:
        super().__init__(context)
        self.cost = 0

    def enter_field(self, node, *_args) -> None:
        base = FIELD_COST.get(node.name.value, 0)
        if not base:
            return
        first = next((a for a in node.arguments if a.name.value == 'first'), None)
        n = int(getattr(first.value, 'value', 1)) if first is not None else 1
        self.cost += base * max(n, 1)
        if self.cost > MAX_COST:
            self.report_error(
                GraphQLError(f'query cost {self.cost} exceeds limit {MAX_COST}', node)
            )


# Built at boot from a directory of .graphql files shipped with the clients.
PERSISTED = {
    hashlib.sha256(p.read_bytes()).hexdigest(): p.read_text()
    for p in Path('persisted').glob('*.graphql')
}


def resolve_persisted(digest: str) -> str:
    try:
        return PERSISTED[digest]
    except KeyError:
        raise HTTPException(status_code=400, detail='unknown persisted query')


# schema.py
# schema = strawberry.Schema(
#     query=Query,
#     extensions=[AddValidationRules([CostLimit])],
#     config=StrawberryConfig(...),
# )

We expose GET /graphql/q/{digest}?vars=... alongside the POST endpoint. A Cloudflare Worker rewrites it, adds Cache-Control: public, max-age=300, stale-while-revalidate=900, and the edge absorbs the repeat traffic. Persisted GETs currently run around an 87% edge hit ratio for the embed widget, which is the single biggest win in this whole project and has nothing to do with GraphQL being GraphQL.

What is deliberately absent from the schema

The GDPR posture is mostly about what does not exist:

• No viewer, no user, no session. The API is anonymous and sets no cookies, so there is nothing to consent to.
• Request IPs are never resolved to a region inside this service. Market is an explicit argument supplied by the caller.
• Aggregate market statistics suppress any bucket with fewer than 50 events, so a thin slice cannot be narrowed to one creator or one viewer.
• Logs record the operation name and the persisted-query digest, never the variables. Variables can carry a video id, and a video id plus a timestamp plus an IP in an access log is personal data by any reasonable reading.

Having one schema instead of eleven serializers is what makes this reviewable. Adding a field is a diff in one file, and the reviewer's only question is whether that field should be public.

What it cost and what it bought

On a 2 vCPU box, a 50-video connection with channel and a 24-point velocity series per node runs p50 3.1 ms and p95 11 ms — dominated by JSON serialisation, not SQLite. The embed payload went from 38 KB to 2.8 KB because the widget now asks for four fields. Eleven REST endpoints are down to two routes.

Things I would skip if starting again: subscriptions (nobody asked, and polling a 5-minute-cached GET is cheaper), and federation (one schema, one team, no seams to stitch). Things I would do sooner: the cursor epoch, and the rollback() in the pool release. Both were incidents before they were design decisions.

The general lesson is unglamorous. GraphQL did not make the API fast — keyset pagination, a covering index, and DataLoader did. What GraphQL bought was a single typed boundary where a field becomes public, which for a product built on a privacy promise turned out to be worth more than the bytes we saved.