Building a GraphQL Video Discovery API with Strawberry and FastAPI

One video, twelve clients, twelve shapes

At TopVideoHub we aggregate trending video across Japan, Korea, Taiwan, Thailand, Vietnam and Hong Kong. A single canonical video row feeds a mobile app that wants a thumbnail and a title, a smart-TV client that wants duration and localized captions, and an internal editorial dashboard that wants everything including per-region rank history. For a while we served all of them from REST, and the endpoint list metastasized: /api/trending, /api/trending/compact, /api/search, /api/video/{id}, /api/video/{id}/related, /api/channel/{id}/videos. Every new client shape meant either a new endpoint or a ?fields= query-string parser nobody wanted to maintain.

The breaking point was localization. A video has titles and descriptions in up to eight languages, and CJK strings are large. The TV client in Seoul only needs ko. The mobile client in Bangkok only needs th with an en fallback. Shipping the full language map on every payload was wasting bandwidth on exactly the market we care about most, where mobile data is metered and latency-sensitive. This is textbook over-fetching, and it is exactly the class of problem GraphQL was designed to eliminate: the client declares the shape, the server resolves precisely that shape, nothing more.

This article is the writeup of how we replaced that endpoint sprawl with a single GraphQL API built on Strawberry and FastAPI, sitting in front of the same SQLite FTS5 search index and PHP 8.4 origin we already ran. I will show the schema, the resolver layer, the N+1 problem you will absolutely hit, and how we kept the whole thing safe under public traffic behind Cloudflare and LiteSpeed.

Why Strawberry over the alternatives

Strawberry is a code-first, dataclass-and-type-hints GraphQL library. You describe types with @strawberry.type and Python type annotations, and the SDL (schema definition language) is generated from them. That matters to us for one reason above all: our video model already exists as typed Python, and we did not want to hand-maintain a .graphql file in parallel with the code. Code-first means the type checker is the schema linter.

The other realistic option was Ariadne (schema-first). Schema-first is fine, but it inverts the source of truth — you write SDL, then bind resolvers to it by string name, and drift between the two only surfaces at runtime. With Strawberry, if a resolver returns the wrong type, mypy complains before the request ever lands. For a small backend team shipping fast, that feedback loop is worth more than the theoretical purity of a language-agnostic schema file.

FastAPI is the host because we already speak ASGI, and Strawberry ships a first-class GraphQLRouter that mounts as a normal FastAPI route. That means GraphQL lives next to our existing health checks, auth middleware, and Prometheus metrics without a second server process.

The schema

Start with the types. Note how the language map is expressed as a resolver argument, not a fat field — the client asks for the language it wants and pays only for that.

# schema/types.py
from __future__ import annotations
import strawberry
from enum import Enum
from typing import Optional


@strawberry.enum
class Region(Enum):
    JP = "JP"
    KR = "KR"
    TW = "TW"
    TH = "TH"
    VN = "VN"
    HK = "HK"


@strawberry.type
class Channel:
    id: strawberry.ID
    name: str
    subscriber_count: int


@strawberry.type
class Video:
    id: strawberry.ID
    duration_seconds: int
    published_at: str
    view_count: int
    channel_id: strawberry.Private[str]  # not exposed; used by resolvers

    @strawberry.field
    def title(self, lang: str = "en") -> str:
        # localized_titles is a dict[str, str] loaded onto the instance
        titles = getattr(self, "_titles", {})
        return titles.get(lang) or titles.get("en") or next(iter(titles.values()), "")

    @strawberry.field
    async def channel(self, info: strawberry.Info) -> Optional[Channel]:
        loader = info.context["channel_loader"]
        return await loader.load(self.channel_id)

The title field takes a lang argument with an en fallback chain — this is the localization win encoded directly in the schema. A Seoul client sends title(lang: "ko") and gets one string; the editorial dashboard can request several aliased fields (ko: title(lang:"ko") ja: title(lang:"ja")) in a single round trip.

The channel field is where the danger lives, and I will come back to it. strawberry.Private marks channel_id as internal — it exists on the Python object so resolvers can use it, but it never appears in the public schema.

The query root and search

Our search backend is SQLite with an FTS5 virtual table using a custom CJK-aware tokenizer. That is not incidental — the default unicode61 tokenizer treats a run of Han or Hiragana characters as a single token, which makes Japanese and Chinese search effectively useless. We run a bigram/ICU-style tokenizer so that 東京 matches inside 東京タワー. The GraphQL resolver is a thin async wrapper over that same query the PHP origin uses.

# schema/query.py
import strawberry
from typing import Optional
from .types import Video, Region


@strawberry.type
class Query:
    @strawberry.field
    async def search(
        self,
        info: strawberry.Info,
        query: str,
        region: Optional[Region] = None,
        first: int = 20,
    ) -> list[Video]:
        first = max(1, min(first, 50))  # hard cap page size
        db = info.context["db"]
        # FTS5 MATCH against the CJK-tokenized index; bm25() ranks results
        sql = (
            "SELECT v.id, v.duration_seconds, v.published_at, "
            "       v.view_count, v.channel_id "
            "FROM video_fts f "
            "JOIN videos v ON v.id = f.rowid "
            "WHERE video_fts MATCH ? "
            + ("AND v.region = ? " if region else "")
            + "ORDER BY bm25(video_fts) LIMIT ?"
        )
        params = [query] + ([region.value] if region else []) + [first]
        rows = await db.fetch_all(sql, params)
        return [_hydrate(r, db) for r in rows]


def _hydrate(row, db) -> Video:
    v = Video(
        id=row["id"],
        duration_seconds=row["duration_seconds"],
        published_at=row["published_at"],
        view_count=row["view_count"],
        channel_id=row["channel_id"],
    )
    # attach localized titles lazily-loaded elsewhere; kept off the hot path
    v._titles = {}
    return v

Two details worth calling out. First, first is clamped server-side to a maximum of 50 regardless of what the client sends — never trust a client-supplied limit, GraphQL or not. Second, we pass the raw MATCH string as a bound parameter. FTS5 has its own query syntax (AND, OR, NEAR, prefix *), so we sanitize user input before it becomes a MATCH expression in production; a bare user string can produce syntax errors or unintended operators. For the aggregator, we wrap terms in double quotes to force phrase matching against the tokenizer.

The N+1 problem is not optional in GraphQL

Here is the query that will take your database down:

{
  search(query: "東京", region: JP, first: 20) {
    id
    title(lang: "ja")
    channel { name subscriberCount }
  }
}

Twenty videos come back from one FTS5 query. Then, for each of the twenty, the channel resolver fires and issues its own SELECT * FROM channels WHERE id = ?. That is 1 + 20 = 21 queries for one request. Scale that to a first: 50 page with nested channel { relatedVideos { channel { ... } } } and you are into hundreds of queries per HTTP request. This is the single most common way a naive GraphQL server falls over, and it is invisible in development where your dataset is small.

The fix is the DataLoader pattern: batch and cache within a single request. Strawberry ships strawberry.dataloader.DataLoader. You give it a batch function that takes a list of keys and returns a list of results in the same order; the loader collects all the .load() calls made during one tick of the event loop and dispatches them as a single batched query.

# loaders.py
from strawberry.dataloader import DataLoader
from .types import Channel


def make_channel_loader(db) -> DataLoader:
    async def batch_load(keys: list[str]) -> list[Channel | None]:
        # one query for all requested channel ids
        placeholders = ",".join("?" for _ in keys)
        rows = await db.fetch_all(
            f"SELECT id, name, subscriber_count "
            f"FROM channels WHERE id IN ({placeholders})",
            keys,
        )
        by_id = {
            r["id"]: Channel(
                id=r["id"], name=r["name"], subscriber_count=r["subscriber_count"]
            )
            for r in rows
        }
        # MUST return results in the same order as keys, None for misses
        return [by_id.get(k) for k in keys]

    return DataLoader(load_fn=batch_load)

With the loader wired in, those twenty channel resolvers each call loader.load(channel_id), Strawberry coalesces them, and the batch function runs one WHERE id IN (...) query. 21 queries collapse to 2. The ordering contract is strict and easy to get wrong: the returned list must line up positionally with the input keys, with None for anything not found — that is why we build a dict and re-index rather than returning rows in database order.

Critically, the loader must be created per request, not once at module load. Its cache is meant to live and die with a single operation; a process-lifetime loader would serve stale channel data forever. That constraint drives how we build the context.

Wiring it into FastAPI

The GraphQLRouter takes a context_getter that runs per request. This is where the per-request DataLoader and the database handle get injected into info.context.

# app.py
import strawberry
from fastapi import FastAPI, Request
from strawberry.fastapi import GraphQLRouter
from strawberry.extensions import QueryDepthLimiter, MaxTokensLimiter

from schema.query import Query
from loaders import make_channel_loader
from db import Database

app = FastAPI()
db = Database("file:videos.db?mode=ro", uri=True)  # read-only replica


async def get_context(request: Request) -> dict:
    return {
        "request": request,
        "db": db,
        "channel_loader": make_channel_loader(db),  # fresh per request
    }


schema = strawberry.Schema(
    query=Query,
    extensions=[
        QueryDepthLimiter(max_depth=6),
        MaxTokensLimiter(max_token_count=800),
    ],
)

graphql_app = GraphQLRouter(
    schema,
    context_getter=get_context,
    graphql_ide=None,  # disable the in-browser IDE in production
)
app.include_router(graphql_app, prefix="/graphql")


@app.get("/healthz")
async def healthz() -> dict:
    return {"ok": True}

The two extensions are not decoration. QueryDepthLimiter(max_depth=6) rejects deeply nested queries before execution — without it, a malicious client can write channel { videos { channel { videos { ... }}}} recursion that fans out into an exponential number of resolver calls. MaxTokensLimiter caps the raw size of the incoming query document. A public GraphQL endpoint without depth and complexity limits is a denial-of-service vector, full stop; the flexibility that makes GraphQL nice for clients is the same flexibility an attacker uses to ask for the entire graph in one request.

We also disable the GraphQL IDE (graphql_ide=None) and, in a real deployment, disable introspection in production. Introspection is wonderful in staging and a free schema map for anyone probing your API in production.

Living alongside the PHP origin

We did not rewrite the backend. The GraphQL service is a read-only ASGI process that shares the same SQLite database file (opened mode=ro) that PHP 8.4 writes to via its cron ingestion. The FTS5 query the Python resolver runs is the same one our PHP search controller has run for two years. Here is the origin-side equivalent, kept in sync deliberately so both paths rank results identically:

<?php
// SearchModel.php (PHP 8.4) — the canonical FTS5 query the GraphQL layer mirrors
declare(strict_types=1);

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

    /** @return list<array{id:string,view_count:int}> */
    public function search(string $query, ?string $region, int $first = 20): array
    {
        $first = max(1, min($first, 50));
        $sql = 'SELECT v.id, v.view_count '
             . 'FROM video_fts f '
             . 'JOIN videos v ON v.id = f.rowid '
             . 'WHERE video_fts MATCH :q '
             . ($region ? 'AND v.region = :region ' : '')
             . 'ORDER BY bm25(video_fts) LIMIT :first';

        $stmt = $this->db->prepare($sql);
        // phrase-quote the term so the CJK bigram tokenizer matches cleanly
        $stmt->bindValue(':q', '"' . str_replace('"', '', $query) . '"');
        if ($region) {
            $stmt->bindValue(':region', $region);
        }
        $stmt->bindValue(':first', $first, \PDO::PARAM_INT);
        $stmt->execute();

        return $stmt->fetchAll(\PDO::FETCH_ASSOC);
    }
}

Keeping the two queries byte-for-byte comparable is a discipline, not an accident. When search ranking looks wrong, having one canonical SQL statement means we debug ranking in one place. The GraphQL layer adds shaping and batching; it does not add a second, divergent notion of what "relevant" means.

Caching in front of a POST endpoint

GraphQL's usual weakness is caching: everything is a POST /graphql, and CDNs do not cache POST bodies. Behind Cloudflare and LiteSpeed this matters, because our read traffic is heavily skewed toward a handful of trending queries per region that repeat thousands of times an hour.

We solved it with two moves. First, persisted queries: clients register their query documents ahead of time and send only a hash plus variables at runtime. That converts an opaque POST body into something with a stable, cacheable identity. Second, for the highest-traffic operations we accept the same registered query over GET with the hash and variables in the query string, which Cloudflare and LiteSpeed's page cache will happily cache with a short TTL and stale-while-revalidate. Trending changes on the order of hours, not seconds, so a 10-minute edge TTL on the trending operation removes the overwhelming majority of requests from the origin entirely — the same caching posture we already run for our REST trending endpoints.

Everything else — search with a long-tail of unique queries, authenticated editorial requests — stays uncached POST, which is correct. The point is not to cache all of GraphQL; it is to recognize the small set of hot, public, cacheable operations and give them an identity the edge can key on.

What we measured

After cutting the mobile clients over, the median trending payload for a Korean client dropped from 34 KB to 9 KB, almost entirely from no longer shipping seven unused language variants of every title and description. The endpoint count in our router went from fourteen REST routes to one GraphQL route plus a health check. And the N+1 fix took the p99 database time on a nested search-with-channels query from roughly 180 ms down to under 20 ms, because 40-odd sequential lookups became two batched ones.

The part I did not expect to matter as much as it did: schema evolution got calmer. Adding a captions(lang:) field to Video broke nothing, because existing clients simply do not ask for it. In REST, that same change is a version bump or a new endpoint and a coordination dance across three client teams.

Conclusion

GraphQL earns its keep when you have one canonical resource and many clients that each want a different slice of it — which is exactly the shape of a multi-region, multi-language video aggregator. Strawberry's code-first model kept the schema honest against our existing typed Python, FastAPI gave us a home next to the rest of our ASGI stack, and DataLoader turned the N+1 trap into two queries instead of forty.

If you take three things from this: create your DataLoaders per request or you will serve stale data; put depth and complexity limits on before you expose the endpoint or you have built a DoS amplifier; and treat your search SQL as one canonical statement shared with your origin rather than reimplementing ranking in the resolver. Do those, and a GraphQL layer becomes a thin, safe shaping tier over infrastructure you already trust — not a rewrite, just a better front door.