Building a GraphQL Video Discovery API With Strawberry and FastAPI

Every video discovery product hits the same wall. We serve eight regions from TrendVidStream, and the moment you have more than one client — a web grid, a mobile app, an internal ranking dashboard, a sitemap generator — your REST surface fractures. The web grid wants thumbnails, titles, and view counts. The recommendation job wants region availability, crawl timestamps, and channel metadata. The mobile app wants a trimmed payload because it is on cellular in São Paulo. With REST you either ship a dozen bespoke endpoints or you serve one fat /videos object and let every caller over-fetch. Both roads end in the same place: a maintenance tax you pay forever, plus an N+1 problem the day someone asks for the channel name next to each video.

We moved video discovery to GraphQL to kill that tax, and after a year of running it in production I would make the same call again. This post walks through the exact stack we use — Strawberry for the schema, FastAPI for transport, and SQLite FTS5 behind the resolvers — with runnable code you can drop into your own project. It sits in front of the same PHP 8.4 and multi-region cron pipeline that powers discovery at TrendVidStream, so I will show how the two halves talk to each other.

Why Strawberry instead of a schema-first library

Most GraphQL tutorials for Python reach for Ariadne or Graphene. We went with Strawberry for three concrete reasons that matter once you are past the toy stage.

• It is code-first and type-hint native. Your schema is your Python dataclasses. There is no .graphql file to keep in sync with your resolvers, which is exactly the kind of drift that bit us with our old REST OpenAPI spec.
• It plays cleanly with async. Every resolver can be a coroutine, which is non-negotiable when a single query fans out to multiple regions and a full-text index.
• DataLoader support is first-class. The N+1 killer is built in, not bolted on.

The cost is that your schema lives in Python and you need to actually run the server to introspect it. For a backend team that already writes typed Python, that is a trade worth making.

Modeling the domain

Video discovery has a small, stable core: a video, the channel that published it, and the regions where it is trending. The trick is that "trending" is region-scoped — a clip can be #3 in Germany and unranked in Japan — so region cannot be a scalar field on the video. It is an edge with its own data (rank, view velocity, first-seen date).

Here is the Strawberry schema. Note that every type maps to a row shape our SQLite pipeline already produces, so the GraphQL layer is a thin, honest projection over storage rather than a second source of truth.

# schema.py
import strawberry
from enum import Enum
from typing import Optional


@strawberry.enum
class Region(Enum):
    US = "US"
    GB = "GB"
    DE = "DE"
    FR = "FR"
    IN = "IN"
    BR = "BR"
    JP = "JP"
    KR = "KR"


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


@strawberry.type
class RegionRank:
    region: Region
    rank: int
    view_velocity: float  # views per hour at last crawl
    first_seen: str       # ISO 8601


@strawberry.type
class Video:
    id: str
    title: str
    thumbnail_url: str
    view_count: int
    published_at: str
    channel_id: strawberry.Private[str]  # not exposed, used by resolvers

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

    @strawberry.field
    async def regions(self, info: strawberry.Info) -> list[RegionRank]:
        return await info.context["region_loader"].load(self.id)

Two details earn their keep. strawberry.Private[str] on channel_id keeps the foreign key out of the public schema while still letting the channel resolver use it. And both channel and regions resolve through a loader pulled off the request context — that is where the N+1 protection lives, which we wire up next.

Killing N+1 with DataLoader

Without batching, a query for 50 trending videos that also asks for each channel name fires 50 separate channel lookups. Multiply that by the regions edge and you are issuing hundreds of round trips for one screen. DataLoader collects every .load(id) call inside a single tick of the event loop and hands you the full list of keys once, so you issue exactly one query per relationship.

# loaders.py
import sqlite3
from strawberry.dataloader import DataLoader

DB_PATH = "discovery.db"


def _connect() -> sqlite3.Connection:
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    return conn


async def batch_load_channels(keys: list[str]) -> list[object]:
    placeholders = ",".join("?" for _ in keys)
    with _connect() as conn:
        rows = conn.execute(
            f"SELECT id, name, subscriber_count FROM channels "
            f"WHERE id IN ({placeholders})",
            keys,
        ).fetchall()
    by_id = {r["id"]: r for r in rows}
    # DataLoader REQUIRES results in the same order as keys
    from schema import Channel
    return [
        Channel(id=r["id"], name=r["name"], subscriber_count=r["subscriber_count"])
        if (r := by_id.get(k)) else None
        for k in keys
    ]


async def batch_load_regions(keys: list[str]) -> list[list]:
    placeholders = ",".join("?" for _ in keys)
    with _connect() as conn:
        rows = conn.execute(
            f"SELECT video_id, region, rank, view_velocity, first_seen "
            f"FROM region_ranks WHERE video_id IN ({placeholders}) "
            f"ORDER BY rank ASC",
            keys,
        ).fetchall()
    from schema import RegionRank, Region
    grouped: dict[str, list] = {k: [] for k in keys}
    for r in rows:
        grouped[r["video_id"]].append(
            RegionRank(
                region=Region(r["region"]),
                rank=r["rank"],
                view_velocity=r["view_velocity"],
                first_seen=r["first_seen"],
            )
        )
    return [grouped[k] for k in keys]


def make_loaders() -> dict:
    return {
        "channel_loader": DataLoader(load_fn=batch_load_channels),
        "region_loader": DataLoader(load_fn=batch_load_regions),
    }

The two rules that trip everyone up: the loader must return a list the same length and order as the incoming keys, and you must build a fresh loader per request. A loader caches within its lifetime, so a process-wide loader would happily serve yesterday's channel name after a crawl updated it. We create them in the FastAPI context factory below, which scopes them to a single HTTP request.

Wiring Strawberry into FastAPI

FastAPI gives us async transport, dependency injection, and a place to hang the region routing middleware we already run for the PHP site. Strawberry ships a GraphQLRouter that mounts as a normal APIRouter, so the GraphQL endpoint lives next to our health checks and the IndexNow webhook.

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

from schema import Video, Region
from loaders import make_loaders, _connect
from search import search_videos  # defined in the next section


@strawberry.type
class Query:
    @strawberry.field
    async def trending(
        self, region: Region, limit: int = 25
    ) -> list[Video]:
        with _connect() as conn:
            rows = conn.execute(
                """
                SELECT v.id, v.title, v.thumbnail_url, v.view_count,
                       v.published_at, v.channel_id
                FROM videos v
                JOIN region_ranks r ON r.video_id = v.id
                WHERE r.region = ?
                ORDER BY r.rank ASC
                LIMIT ?
                """,
                (region.value, min(limit, 100)),
            ).fetchall()
        return [
            Video(
                id=r["id"], title=r["title"], thumbnail_url=r["thumbnail_url"],
                view_count=r["view_count"], published_at=r["published_at"],
                channel_id=r["channel_id"],
            )
            for r in rows
        ]

    @strawberry.field
    async def search(self, query: str, limit: int = 25) -> list[Video]:
        return await search_videos(query, limit)


async def get_context(request: Request) -> dict:
    return make_loaders()


schema = strawberry.Schema(query=Query)
graphql_app = GraphQLRouter(schema, context_getter=get_context)

app = FastAPI(title="Discovery GraphQL")
app.include_router(graphql_app, prefix="/graphql")


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

The get_context function is the important seam. It runs once per request and returns the dict that every resolver reads via info.context. That is where the per-request loaders get born, and it is also where you would attach the caller's resolved region, auth claims, or a request-scoped DB connection. Cap limit server-side (min(limit, 100)) so a client cannot ask for ten thousand rows and DoS your SQLite file — GraphQL makes it easy to forget that the client no longer chooses the endpoint shape, only the field selection.

Full-text search over SQLite FTS5

This is where the reference stack pays off. Our crawler already maintains an FTS5 virtual table so the PHP site can do instant title search without Elasticsearch. GraphQL reuses that index verbatim — the search field is a thin resolver over an FTS5 MATCH query, ranked by BM25.

# search.py
import sqlite3
from loaders import _connect
from schema import Video


def _sanitize(term: str) -> str:
    # FTS5 treats many chars as operators; quote each token to be safe
    tokens = [t for t in term.replace('"', " ").split() if t]
    return " ".join(f'"{t}"' for t in tokens)


async def search_videos(query: str, limit: int = 25) -> list[Video]:
    match = _sanitize(query)
    if not match:
        return []
    with _connect() as conn:
        rows = conn.execute(
            """
            SELECT v.id, v.title, v.thumbnail_url, v.view_count,
                   v.published_at, v.channel_id
            FROM videos_fts f
            JOIN videos v ON v.id = f.rowid
            WHERE f.videos_fts MATCH ?
            ORDER BY bm25(videos_fts) ASC, v.view_count DESC
            LIMIT ?
            """,
            (match, min(limit, 100)),
        ).fetchall()
    return [
        Video(
            id=r["id"], title=r["title"], thumbnail_url=r["thumbnail_url"],
            view_count=r["view_count"], published_at=r["published_at"],
            channel_id=r["channel_id"],
        )
        for r in rows
    ]

The FTS5 table is created once by the crawler with CREATE VIRTUAL TABLE videos_fts USING fts5(title, content='videos', content_rowid='rowid') and kept current with triggers on the videos table. Two production notes: always sanitize the user's string before it reaches MATCH, because raw input containing AND, OR, NEAR, or an unbalanced quote will raise an FTS5 syntax error and leak a 500. And bm25() returns lower is better, which is why the ORDER BY is ascending — a mistake I have watched more than one engineer debug for an afternoon. Breaking ties on view_count keeps the ranking stable for the head queries that dominate discovery traffic.

Talking to it from the PHP site

Our front end is PHP 8.4, deployed over FTP to LiteSpeed hosts on a multi-region cron. It does not need a GraphQL client library — a single POST with a query string is enough, and the whole point of GraphQL is that the PHP grid asks for exactly the four fields it renders and nothing else. Here is the fetch helper we run, using PHP 8.4 features like the #[\SensitiveParameter] attribute and first-class readonly DTOs.

<?php
// GraphQLClient.php  (PHP 8.4)
declare(strict_types=1);

final readonly class GraphQLClient
{
    public function __construct(
        private string $endpoint,
        #[\SensitiveParameter] private string $token = '',
    ) {}

    public function query(string $query, array $variables = []): array
    {
        $payload = json_encode(
            ['query' => $query, 'variables' => $variables],
            JSON_THROW_ON_ERROR,
        );

        $ch = curl_init($this->endpoint);
        curl_setopt_array($ch, [
            CURLOPT_RETURNTRANSFER => true,
            CURLOPT_POST           => true,
            CURLOPT_POSTFIELDS     => $payload,
            CURLOPT_TIMEOUT        => 4,
            CURLOPT_HTTPHEADER     => array_filter([
                'Content-Type: application/json',
                $this->token ? "Authorization: Bearer {$this->token}" : null,
            ]),
        ]);

        $body = curl_exec($ch);
        if ($body === false) {
            throw new RuntimeException('GraphQL transport error: ' . curl_error($ch));
        }
        curl_close($ch);

        $decoded = json_decode($body, true, flags: JSON_THROW_ON_ERROR);
        if (isset($decoded['errors'])) {
            throw new RuntimeException($decoded['errors'][0]['message'] ?? 'GraphQL error');
        }
        return $decoded['data'];
    }
}

// Region-scoped trending grid — asks for exactly what the template renders.
$client = new GraphQLClient('http://127.0.0.1:8000/graphql');
$data = $client->query(<<<'GQL'
    query Trending($region: Region!) {
        trending(region: $region, limit: 24) {
            id
            title
            thumbnailUrl
            viewCount
            channel { name }
        }
    }
GQL, ['region' => 'DE']);

foreach ($data['trending'] as $video) {
    printf("%s — %s\n", $video['channel']['name'], $video['title']);
}

The grid query touches channel { name } for all 24 videos, and thanks to the DataLoader that resolves as a single WHERE id IN (...) query, not 24 lookups. The PHP side never learns that detail — it just gets its data in one round trip and caches the rendered HTML the way it always has. Keep a short CURLOPT_TIMEOUT; a discovery grid should degrade to a cached page, never hang on a slow resolver.

A lightweight Go edge probe

We also run a tiny Go binary at the edge that pings the API on a schedule to keep the SQLite page cache warm and to alert if the p95 crosses a threshold before real users notice. GraphQL's single endpoint makes this trivial — one POST, one JSON decode, no per-route client to maintain.

package main

import (
    "bytes"
    "encoding/json"
    "fmt"
    "net/http"
    "time"
)

const q = `{"query":"{ trending(region: US, limit: 1) { id } }"}`

func main() {
    client := &http.Client{Timeout: 3 * time.Second}
    start := time.Now()

    resp, err := client.Post(
        "http://127.0.0.1:8000/graphql",
        "application/json",
        bytes.NewBufferString(q),
    )
    if err != nil {
        fmt.Printf("UNHEALTHY transport: %v\n", err)
        return
    }
    defer resp.Body.Close()

    var out struct {
        Data   map[string]any `json:"data"`
        Errors []any          `json:"errors"`
    }
    _ = json.NewDecoder(resp.Body).Decode(&out)

    elapsed := time.Since(start).Milliseconds()
    if len(out.Errors) > 0 || out.Data == nil {
        fmt.Printf("UNHEALTHY schema: %+v (%dms)\n", out.Errors, elapsed)
        return
    }
    fmt.Printf("OK %dms\n", elapsed)
}

Performance and operational lessons

A few things I wish someone had told me before we shipped:

• Depth-limit and complexity-limit the schema. GraphQL lets a client write a query that recurses video → channel → videos → channel. Strawberry supports query-depth validation extensions — turn them on before you expose the endpoint publicly, or keep the API internal behind the PHP layer as we do.
• Persisted queries beat ad-hoc queries in production. Once the PHP grid queries stabilize, register them by hash so the server rejects anything it has not seen. You get the flexibility during development and a locked-down surface in production.
• The DataLoader cache is per-request, not per-process. This is a feature. It means correctness after a crawl updates a row, at the cost of no cross-request memoization — put that layer in front, in LiteSpeed or Cloudflare, keyed on the query hash and region.
• SQLite is enough far longer than you think. With FTS5 handling search and a WAL-mode connection, our single-file database serves the entire eight-region discovery surface. The GraphQL layer did not change that; it just gave every client a precise way to ask for slices of it.
• Watch your LIMIT and MATCH sanitization. These are the two places raw client input reaches the database. Cap the first, quote the second, and you have closed the obvious foot-guns.

Conclusion

GraphQL did not make our video discovery faster — SQLite and BM25 already handled that. What it did was collapse a sprawling REST surface into one typed schema that every client, from the PHP grid to the Go probe, queries for exactly the fields it needs, with N+1 handled once at the loader instead of forever at each endpoint. Strawberry keeps the schema honest by making it plain Python types, FastAPI gives it async transport next to our existing services, and FTS5 means search is a resolver, not a second system to operate. If you are drowning in bespoke /videos-for-mobile endpoints, this stack is a weekend of work that pays itself back the first time a new client shows up asking for a slightly different shape.