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

Our Android client painted one search screen with four HTTP requests: GET /api/search?q=, then GET /api/channels?ids=, then GET /api/categories, then a per-video GET /api/video/{id} as soon as the user tapped a card. Every one of those endpoints returned the whole row out of SQLite — 41 columns for a video — and the client rendered six of them. On a throttled 3G profile the screen took 2.4 s at p95 to first meaningful paint, and about 180 KB of JSON crossed the wire to draw something that genuinely needs 20 KB. That is the problem I set out to fix at DailyWatch, and the fix was a read-only GraphQL sidecar built with Strawberry and FastAPI, pointed at the same SQLite file the PHP site already reads.

This is the write-up of what that actually took: schema design against an FTS5 index, keyset pagination over bm25 scores, killing the N+1 with DataLoader, cost limits, and the part nobody blogs about — making a POST-shaped protocol cacheable at the CDN edge.

Four round trips, six fields used

The REST surface grew the way REST surfaces grow. /api/search was written first and returned video rows. Then the client needed channel names next to each result, so we added /api/channels?ids=a,b,c — which is a hand-rolled DataLoader with no schema, no batching contract, and a URL length limit. Then a category filter needed the category list. The endpoints were individually fine; the composition was the problem.

The concrete costs, measured rather than guessed:

• Over-fetching. The videos table carries description text, raw region codes, fetch timestamps, and moderation flags. The card UI needs id, title, duration, view count, thumbnail, and channel title.
• Latency chaining. Request 2 cannot start until request 1 returns the channel IDs. Two serial RTTs on a 250 ms mobile connection is half a second before any pixel moves.
• Cache fragmentation. Four responses, four TTLs, four invalidation paths. When the cron job refreshed trending videos, one of the four consistently went stale in a way that made the UI look broken.
• No contract. Adding a field to the search response broke an older client that did strict JSON decoding. There was no way to know which fields were actually in use.

The site itself is a PHP 8.4 monolith served through LiteSpeed with a page cache in front, and it is genuinely fast — server-rendered HTML, no hydration, SQLite reads in single-digit milliseconds. Nothing about that needed to change. The apps were the ones paying for a data model shaped around HTML pages.

Why a Python sidecar and not GraphQL inside the PHP app

I considered bolting a GraphQL layer onto the monolith. There are competent PHP implementations. I chose a separate service anyway, and the reasoning was mostly operational rather than aesthetic:

• The read model is already a file. SQLite means a second process can open the same database read-only, in WAL mode, with zero coordination and zero additional infrastructure. No replica, no connection pool, no network hop to the data.
• Blast radius. If the GraphQL service falls over, the website keeps serving HTML. Only the apps degrade, and they degrade to a cached view. A GraphQL layer inside the monolith shares a fatal error handler with the thing that pays the bills.
• Strawberry is code-first with real type hints. The schema is generated from Python types, so pyright catches a resolver returning the wrong shape at build time. An SDL file in a repo drifts from its resolvers within about two sprints; a dataclass cannot.
• Async fits the access pattern. A GraphQL request here is many small indexed reads, not one big query. That maps cleanly onto batching.

The honest cost of the sidecar: one more deploy target, one more origin behind Cloudflare, one more thing to monitor, and a hard invariant that this service never writes. I enforce that last one with PRAGMA query_only = 1 on every connection rather than trusting myself.

The first thing that had to be shared, not forked, was full-text query construction. Here is the PHP side that already existed — the FTS5 tokenizer sanitizer, plus the query it feeds:

<?php
declare(strict_types=1);

final class FtsQuery
{
    private const int MAX_TOKENS = 8;

    /** Turn arbitrary user input into a safe FTS5 MATCH expression. */
    public static function fromUserInput(string $raw): ?string
    {
        // FTS5 reads " * : ^ - AND OR NOT as syntax. Users type them as text.
        $clean = preg_replace('/[^\p{L}\p{N}\s]+/u', ' ', mb_strtolower(trim($raw))) ?? '';
        $tokens = array_slice(
            array_values(array_filter(explode(' ', $clean), strlen(...))),
            0,
            self::MAX_TOKENS
        );
        if ($tokens === []) {
            return null;
        }

        $last = array_key_last($tokens);
        foreach ($tokens as $i => $t) {
            // Prefix-match only the trailing token: 'coffee gr' -> "coffee" AND "gr"*
            $tokens[$i] = ($i === $last && mb_strlen($t) >= 2)
                ? '"' . $t . '"*'
                : '"' . $t . '"';
        }
        return implode(' AND ', $tokens);
    }
}

$pdo = new PDO('sqlite:' . __DIR__ . '/../data/app.db', options: [
    PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION,
    PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC,
]);
$pdo->exec('PRAGMA query_only = 1');

$match = FtsQuery::fromUserInput($_GET['q'] ?? '');
if ($match === null) {
    echo json_encode(['items' => []]);
    exit;
}

// Title weighted 10x, channel title 3x, description 1x. bm25 is negative: lower = better.
$stmt = $pdo->prepare(<<<SQL
    SELECT v.id, v.title, v.duration_seconds, v.view_count,
           bm25(videos_fts, 10.0, 3.0, 1.0) AS score
    FROM videos_fts
    JOIN videos v ON v.rowid = videos_fts.rowid
    WHERE videos_fts MATCH :m
    ORDER BY score, v.rowid
    LIMIT 20
SQL);
$stmt->execute([':m' => $match]);
echo json_encode(['items' => $stmt->fetchAll()], JSON_UNESCAPED_UNICODE);

That sanitizer had to be reimplemented in Python byte-for-byte, and getting it wrong means the app and the website return different results for the same query. I will come back to how I pinned that down.

The schema is the contract, not the table

The temptation with code-first GraphQL is to mirror your tables. Resist it. The schema should describe what clients need; the repository layer translates. Two concrete decisions here: thumbnail is a computed field with a width argument rather than four stored URL columns, and channel is a real object resolved lazily so a query that does not ask for it costs nothing.

# schema.py
from __future__ import annotations

from datetime import datetime

import strawberry
from strawberry.types import Info


@strawberry.type
class Channel:
    id: str
    title: str
    subscriber_count: int | None
    thumbnail_url: str | None


@strawberry.type
class Video:
    id: str
    title: str
    published_at: datetime
    duration_seconds: int
    view_count: int
    region: str
    # Never leaves the process; used only to feed the loader.
    channel_id: strawberry.Private[str]

    @strawberry.field(description='Uploading channel. Batched across the whole request.')
    async def channel(self, info: Info) -> Channel | None:
        return await info.context['loaders']['channel'].load(self.channel_id)

    @strawberry.field(description='Thumbnail URL sized for the requested render width.')
    def thumbnail(self, width: int = 320) -> str:
        size = 'maxresdefault' if width > 480 else 'hqdefault'
        return f'https://i.ytimg.com/vi/{self.id}/{size}.jpg'


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


@strawberry.type
class VideoConnection:
    edges: list[VideoEdge]
    has_next_page: bool


@strawberry.type
class Query:
    @strawberry.field
    async def search(
        self,
        info: Info,
        q: str,
        first: int = 20,
        after: str | None = None,
        region: str | None = None,
    ) -> VideoConnection:
        first = max(1, min(first, 50))
        return await info.context['repo'].search(q=q, first=first, after=after, region=region)

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

strawberry.Private[str] is doing real work: it keeps channel_id on the Python object for the resolver while excluding it from the generated schema. Clients cannot query it, so I am free to change how channels are keyed later.

Capping first in the resolver rather than trusting the client is not paranoia. It is the difference between a request that reads 20 rows and one that reads 10,000 because someone typo'd a zero.

Making FTS5 behave under async

SQLite is synchronous. aiosqlite hides that behind a dedicated thread per connection, which means one connection serializes its reads. That is fine — the reads are 0.2–4 ms — but it shapes the deployment: four uvicorn workers, one connection each, four concurrent readers against a WAL file. Do not build a 50-connection pool and expect 50× throughput from one disk.

The second decision is pagination. LIMIT ? OFFSET ? re-scans everything before the offset, so page 40 costs forty times page 1. Keyset pagination on (score, id) costs the same on every page. SQLite's row-value comparison (3.15+) makes the predicate readable.

# repo.py
from __future__ import annotations

import base64
import re

import aiosqlite

from schema import Channel, Video, VideoConnection, VideoEdge

MAX_TOKENS = 8
SEARCH_CEILING = 500  # deepest reachable result; documented, not silently truncated
_NOISE = re.compile(r'[\W_]+', re.UNICODE)


def fts_query(raw: str) -> str | None:
    """Port of FtsQuery::fromUserInput. Must stay byte-identical to the PHP version."""
    tokens = [t for t in _NOISE.sub(' ', raw.lower()).split() if t][:MAX_TOKENS]
    if not tokens:
        return None
    last = len(tokens) - 1
    parts = [
        f'"{t}"*' if (i == last and len(t) >= 2) else f'"{t}"'
        for i, t in enumerate(tokens)
    ]
    return ' AND '.join(parts)


def _encode(score: float, vid: str) -> str:
    return base64.urlsafe_b64encode(f'{score:.6f}|{vid}'.encode()).decode()


def _decode(cursor: str) -> tuple[float, str]:
    score, vid = base64.urlsafe_b64decode(cursor.encode()).decode().split('|', 1)
    return float(score), vid


SEARCH_SQL = '''
WITH ranked AS (
    SELECT v.id, v.title, v.published_at, v.duration_seconds,
           v.view_count, v.region, v.channel_id,
           bm25(videos_fts, 10.0, 3.0, 1.0) AS score
    FROM videos_fts
    JOIN videos v ON v.rowid = videos_fts.rowid
    WHERE videos_fts MATCH :m
      AND (:region IS NULL OR v.region = :region)
    ORDER BY score, v.id
    LIMIT :ceiling
)
SELECT * FROM ranked
WHERE (:has_cursor = 0) OR ((score, id) > (:c_score, :c_id))
ORDER BY score, id
LIMIT :lim
'''


class Repo:
    def __init__(self, uri: str) -> None:
        self._uri = uri
        self._conn: aiosqlite.Connection | None = None

    async def connect(self) -> None:
        self._conn = await aiosqlite.connect(self._uri, uri=True)
        self._conn.row_factory = aiosqlite.Row
        await self._conn.execute('PRAGMA query_only = 1')
        await self._conn.execute('PRAGMA cache_size = -32000')      # 32 MB page cache
        await self._conn.execute('PRAGMA mmap_size = 268435456')    # 256 MB mmap

    async def close(self) -> None:
        if self._conn is not None:
            await self._conn.close()

    async def search(self, q: str, first: int, after: str | None, region: str | None) -> VideoConnection:
        match = fts_query(q)
        if match is None or self._conn is None:
            return VideoConnection(edges=[], has_next_page=False)

        c_score, c_id = _decode(after) if after else (0.0, '')
        rows = await self._conn.execute_fetchall(SEARCH_SQL, {
            'm': match,
            'region': region,
            'ceiling': SEARCH_CEILING,
            'has_cursor': 1 if after else 0,
            'c_score': c_score,
            'c_id': c_id,
            'lim': first + 1,   # one extra row tells us has_next_page for free
        })

        has_next = len(rows) > first
        edges = [
            VideoEdge(
                cursor=_encode(r['score'], r['id']),
                node=Video(
                    id=r['id'],
                    title=r['title'],
                    published_at=r['published_at'],
                    duration_seconds=r['duration_seconds'],
                    view_count=r['view_count'],
                    region=r['region'],
                    channel_id=r['channel_id'],
                ),
            )
            for r in rows[:first]
        ]
        return VideoConnection(edges=edges, has_next_page=has_next)

    async def channels_by_ids(self, ids: list[str]) -> list[Channel | None]:
        """DataLoader batch function. MUST return one slot per key, in key order."""
        if self._conn is None:
            return [None] * len(ids)
        placeholders = ','.join('?' * len(ids))
        rows = await self._conn.execute_fetchall(
            f'SELECT id, title, subscriber_count, thumbnail_url '
            f'FROM channels WHERE id IN ({placeholders})',
            ids,
        )
        found = {r['id']: Channel(**dict(r)) for r in rows}
        return [found.get(i) for i in ids]

A few details worth calling out, because each of them cost me an afternoon:

• bm25() is negative, and more negative means a better match. Sorting ascending is correct. Sorting descending gives you the worst results first and looks superficially plausible in a demo.
• Ties need a tiebreaker. Two videos with identical scores and no secondary sort key will swap order between queries, and keyset pagination will then skip or duplicate rows. ORDER BY score, id fixes it.
• The SEARCH_CEILING is a deliberate limit. The CTE materializes matches before the keyset filter, so unbounded deep paging would degrade. Five hundred results is more than any human pages through; I document it rather than pretending it does not exist.
• Fetch first + 1 rows to compute has_next_page without a second COUNT(*) over the FTS index.

Killing the N+1 with a per-request DataLoader

Twenty search results reference up to twenty distinct channels. Naively, Video.channel fires twenty separate SELECT ... WHERE id = ? queries. Each one is fast; twenty of them behind a thread-serialized connection is not.

strawberry.dataloader.DataLoader collects the keys requested within a single tick of the event loop and calls your batch function once. The two rules that matter: the batch function must return exactly one slot per key in the same order, and the loader must be created per request, never at module scope. A module-level loader caches results across users forever, which is a stale-data bug and, if your schema ever grows an authenticated field, a data-leak bug.

# app.py
import hashlib
import json
from contextlib import asynccontextmanager
from pathlib import Path

import strawberry
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from graphql.validation import NoSchemaIntrospectionCustomRule
from strawberry.dataloader import DataLoader
from strawberry.extensions import (
    AddValidationRules,
    MaxTokensLimiter,
    ParserCache,
    QueryDepthLimiter,
    ValidationCache,
)
from strawberry.fastapi import GraphQLRouter

from repo import Repo
from schema import Query

DB_URI = 'file:/var/www/data/app.db?mode=ro'
repo = Repo(DB_URI)

schema = strawberry.Schema(
    query=Query,
    extensions=[
        ParserCache(maxsize=256),        # skip re-parsing the same document
        ValidationCache(maxsize=256),    # skip re-validating it too
        QueryDepthLimiter(max_depth=8),
        MaxTokensLimiter(max_token_count=1200),
        AddValidationRules([NoSchemaIntrospectionCustomRule]),  # prod only
    ],
)


async def build_context(request: Request) -> dict:
    return {
        'request': request,
        'repo': repo,
        'loaders': {
            # Fresh per request. max_batch_size stays under SQLite's parameter limit.
            'channel': DataLoader(load_fn=repo.channels_by_ids, max_batch_size=200),
        },
    }


@asynccontextmanager
async def lifespan(app: FastAPI):
    await repo.connect()
    yield
    await repo.close()


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

# --- Persisted queries over GET, so the CDN can cache them ---------------
_raw = json.loads(Path('persisted-queries.json').read_text())
PERSISTED: dict[str, str] = {}
for doc_id, document in _raw.items():
    digest = hashlib.sha256(document.encode()).hexdigest()
    if digest != doc_id:
        raise RuntimeError(f'persisted query {doc_id} does not match its own hash')
    PERSISTED[doc_id] = document


@app.get('/gql')
async def persisted_get(request: Request) -> JSONResponse:
    document = PERSISTED.get(request.query_params.get('id', ''))
    if document is None:
        return JSONResponse({'errors': [{'message': 'PersistedQueryNotFound'}]}, status_code=400)

    raw_vars = request.query_params.get('v')
    try:
        variables = json.loads(raw_vars) if raw_vars else {}
    except json.JSONDecodeError:
        return JSONResponse({'errors': [{'message': 'BadVariables'}]}, status_code=400)

    result = await schema.execute(
        document,
        variable_values=variables,
        context_value=await build_context(request),
    )
    if result.errors:
        return JSONResponse(
            {'data': result.data, 'errors': [e.formatted for e in result.errors]},
            status_code=200,
            headers={'Cache-Control': 'no-store'},
        )
    return JSONResponse(
        {'data': result.data},
        headers={
            'Cache-Control': 'public, s-maxage=300, stale-while-revalidate=600',
            'Vary': 'Accept-Encoding',
        },
    )

Cost limits before you need them

A public GraphQL endpoint is an open invitation to write an expensive query. Four cheap defences, in the order I would add them:

• Depth limiting. QueryDepthLimiter(max_depth=8) stops the classic channel { videos { channel { videos { ... } } } } recursion. Our schema's legitimate deepest path is 5.
• Token limiting. MaxTokensLimiter rejects absurdly large documents before the parser does real work. Cheaper than any post-parse analysis.
• Argument caps in resolvers. Every list field clamps first. This is the one that actually bounds database work, and no generic extension can do it for you.
• Introspection off in production. NoSchemaIntrospectionCustomRule as a validation rule. Leave it enabled in staging, where your tooling needs it.

Parser and validation caches are not security features, but they matter here: with persisted queries the document set is small and fixed, so both caches hit essentially 100% of the time and take parsing off the hot path entirely.

Caching a POST protocol at the edge

This is the part that made the whole project worth it. GraphQL over POST is uncacheable by any CDN, which is a real regression when your site already gets a fat cache hit ratio on HTML through LiteSpeed and Cloudflare.

Persisted queries solve it. At build time, a script extracts every GraphQL document from the client, hashes each one with SHA-256, and writes persisted-queries.json. The client ships hashes, not documents. At runtime the app issues GET /gql?id=<sha256>&v=<url-encoded-json> — a plain, idempotent, fully-qualified GET that Cloudflare will cache on the URL. Three things fall out of this:

• The request is smaller. A 900-byte query document becomes a 64-character hash.
• The endpoint is an allowlist. An unknown hash is a 400. Arbitrary queries are impossible in production, which makes the cost limits above a second line of defence rather than the only one.
• The edge does the work. Search results with s-maxage=300 plus stale-while-revalidate=600 mean repeated queries never touch Python. Our persisted GET hit ratio settled around 70%, and the trending query — identical for every user in a region — sits far higher.

I keep /graphql (POST, ad-hoc, introspection enabled) available in staging only. Production exposes /gql alone. Errors always get no-store; caching a transient failure for five minutes is a self-inflicted outage.

Pinning the tokenizer with a parity test

Back to the shared logic. Two implementations of the same sanitizer will diverge, and search that disagrees between the website and the app is a bug users report as "the app is broken" with no further detail. So the Python version is tested against the PHP version directly, by running it.

# tests/test_tokenizer_parity.py
import json
import subprocess

import pytest

from repo import fts_query

CASES = [
    'coffee grinder',
    'C++ tutorial',
    "l'ete a Paris",
    'naive_bayes',
    '   ',
    'a',
    'one two three four five six seven eight nine ten',
    '"drop" OR 1=1',
]


def php_fts(raw: str) -> str | None:
    proc = subprocess.run(
        ['php', 'tools/fts_query.php', raw],
        capture_output=True, text=True, check=True,
    )
    return json.loads(proc.stdout)


@pytest.mark.parametrize('raw', CASES)
def test_python_matches_php(raw: str) -> None:
    assert fts_query(raw) == php_fts(raw)

The first version of the Python tokenizer used re.sub(r'[^\w\s]', ' ', ...), which keeps underscores because \w includes them, while the PHP [^\p{L}\p{N}\s] class strips them. So naive_bayes became one token in Python and two in PHP — different result sets for the same input. The parity test caught it in the first run. Ten lines of test for a class of bug that would otherwise have been found by a user.

What actually changed

Numbers from our hardware, which is one modest box, so treat them as directional:

• Search screen: 4 requests to 1.
• Payload for 20 results with channel data: ~180 KB down to ~24 KB uncompressed.
• p95 time to first meaningful paint on a throttled 3G profile: 2.4 s to ~1.1 s.
• Server time for the search query: 6–12 ms warm, dominated by FTS5, not by GraphQL execution.
• Channel lookups per search request: 20 down to 1.

And the costs, stated plainly, because they are real:

• HTTP status codes stop meaning much. A GraphQL response with errors is still a 200 with an errors array. Every dashboard and alert built on status codes needs rewriting against the response body.
• Partial data is now a client concern. data.search.edges[3].channel can be null because one channel row is missing. The client must handle it; REST hid this behind a whole-response failure.
• Schema evolution is forever. Removing a field means proving no deployed client asks for it. Persisted queries help enormously — the allowlist is the usage inventory.
• Logs get worse. POST /graphql 200 tells you nothing. Log the operation name and the persisted-query id, or you are debugging blind.

I would not do this for a server-rendered site with no API clients. The PHP monolith stays exactly as it was, because for HTML it is the right tool and the numbers say so. GraphQL earned its place here for precisely one reason: multiple clients with different data shapes, hitting the same read model, over slow connections.

Conclusion

The pieces that mattered, in order of impact: a per-request DataLoader, keyset pagination on (score, id) instead of OFFSET, argument caps in resolvers, and persisted queries over GET so the CDN can absorb the repeat traffic. Strawberry contributed the part I care about most — the schema is Python types, so the type checker enforces the contract instead of a hand-maintained SDL file drifting away from reality.

If you are starting this on top of SQLite, do the parity test first. Everything else you can refactor later; a search tokenizer that quietly disagrees with itself will burn a week before you find it.