Building a Video Metadata Service with Litestar and Async Python

A trending-video request on our Japanese front page fans out to roughly 40 metadata lookups: thumbnails, durations, channel names, localized titles, and per-region availability flags. Each of those used to be a blocking PHP call against SQLite, and under the LiteSpeed worker model that meant a single request could pin a worker for 200-300ms while it waited on I/O that was almost entirely network-bound (upstream provider APIs), not CPU-bound. When evening traffic from Tokyo and Seoul overlapped, the worker pool starved and p99 latency doubled. I run the backend for TopVideoHub, an Asia-Pacific multi-language video aggregator, and the fix was to pull the metadata fan-out out of the synchronous PHP path and into a small async service. I picked Litestar (formerly Starlite) over FastAPI, and this post is the concrete why and how — including how it coexists with a PHP 8.4 + SQLite FTS5 + LiteSpeed + Cloudflare stack that we are not rewriting.

The shape of the problem

Before any framework talk, here is what the metadata layer actually has to do on every page render:

• Resolve N video IDs (10-40) to a canonical metadata record.
• Merge in a localized title for the viewer's language — we serve Japanese, Korean, Traditional and Simplified Chinese, Thai, Vietnamese, and English.
• Filter availability by region, because a video trending in Taiwan may be geo-blocked in Vietnam.
• Occasionally enrich a record from an upstream provider when our cached copy is stale.

The first three are cheap local reads. The fourth is the killer: it is high-latency network I/O with unpredictable tail behavior. In synchronous PHP, one slow upstream call inside a batch serializes the whole render. What we wanted was a service where the local reads stay fast and the network fan-out happens concurrently, without spinning up a thread per request. That is the textbook case for an async runtime, and Python's asyncio plus httpx handles it well. The question was which framework to wrap around it.

Why Litestar and not FastAPI

I like FastAPI. But for a metadata service that is essentially a serialization engine in front of SQLite, three Litestar properties mattered more:

• msgspec instead of Pydantic for the hot path. Litestar uses msgspec for its default DTO and serialization layer. msgspec.Struct validation and JSON encoding are consistently several times faster than Pydantic v2 in our benchmarks, and for a service whose entire job is turning rows into JSON, encoding speed is not a micro-optimization — it is the workload.
• A real dependency-injection system with layers. Litestar lets you declare dependencies at the app, router, controller, or handler level, and they compose. Our DB connection is app-scoped, the repository is request-scoped, and the auth context is controller-scoped. No global singletons, no Depends() sprawl.
• Controllers and layered config. FastAPI is function-first; Litestar is comfortable with class-based controllers, which map cleanly onto "everything under /videos shares these dependencies and this cache policy."

None of this makes Litestar a FastAPI killer. It made it the better fit for this service. Let's build it.

Modeling metadata with msgspec Structs

The metadata record is defined once as a msgspec.Struct. This doubles as the validation schema, the OpenAPI model, and the wire encoder — there is no separate serializer.

from __future__ import annotations

import msgspec


class VideoMeta(msgspec.Struct, forbid_unknown_fields=True, frozen=True):
    video_id: str
    title: str
    # per-language overrides, e.g. {"ja": "...", "ko": "...", "zh-Hant": "..."}
    title_localized: dict[str, str]
    channel: str
    duration_s: int
    thumbnail: str
    regions: list[str]  # ISO codes where the video is playable

    def localized(self, lang: str) -> str:
        return self.title_localized.get(lang, self.title)

    def playable_in(self, region: str) -> bool:
        return not self.regions or region in self.regions

forbid_unknown_fields=True means a schema drift in the upstream feed fails loudly at decode time instead of silently dropping data. frozen=True makes the record hashable and safe to stash in an in-process cache. The localized and playable_in helpers keep the CJK and geo logic on the model where it belongs, not scattered across handlers.

The endpoint, with dependency injection

SQLite is a synchronous library, so the honest way to use it under asyncio is aiosqlite, which runs the underlying calls on a thread and gives you an awaitable interface. The connection is provided as an app-level dependency with a generator so it is opened and closed cleanly, and the repository that wraps it is request-scoped.

from collections.abc import AsyncIterator

import aiosqlite
from litestar import Litestar, get
from litestar.controller import Controller
from litestar.di import Provide
from litestar.exceptions import NotFoundException

DB_PATH = "/var/data/topvideohub.sqlite"


class MetaRepo:
    def __init__(self, db: aiosqlite.Connection) -> None:
        self._db = db

    async def by_ids(self, ids: list[str]) -> list[VideoMeta]:
        if not ids:
            return []
        placeholders = ",".join("?" * len(ids))
        sql = (
            "SELECT video_id, title, title_localized, channel, "
            "duration_s, thumbnail, regions "
            f"FROM videos WHERE video_id IN ({placeholders})"
        )
        self._db.row_factory = aiosqlite.Row
        async with self._db.execute(sql, ids) as cur:
            rows = await cur.fetchall()
        return [
            VideoMeta(
                video_id=r["video_id"],
                title=r["title"],
                title_localized=msgspec.json.decode(r["title_localized"] or b"{}"),
                channel=r["channel"],
                duration_s=r["duration_s"],
                thumbnail=r["thumbnail"],
                regions=msgspec.json.decode(r["regions"] or b"[]"),
            )
            for r in rows
        ]


async def provide_db() -> AsyncIterator[aiosqlite.Connection]:
    db = await aiosqlite.connect(DB_PATH)
    await db.execute("PRAGMA journal_mode=WAL")
    await db.execute("PRAGMA busy_timeout=3000")
    try:
        yield db
    finally:
        await db.close()


async def provide_repo(db: aiosqlite.Connection) -> MetaRepo:
    return MetaRepo(db)


class VideoController(Controller):
    path = "/videos"
    dependencies = {"repo": Provide(provide_repo)}

    @get("/{video_id:str}")
    async def get_one(self, video_id: str, repo: MetaRepo) -> VideoMeta:
        rows = await repo.by_ids([video_id])
        if not rows:
            raise NotFoundException(detail=f"unknown video {video_id}")
        return rows[0]

    @get()
    async def get_batch(self, ids: str, repo: MetaRepo) -> list[VideoMeta]:
        # /videos?ids=a,b,c — the front-page fan-out endpoint
        return await repo.by_ids([x for x in ids.split(",") if x][:64])


app = Litestar(
    route_handlers=[VideoController],
    dependencies={"db": Provide(provide_db)},
)

A few deliberate choices worth calling out:

• db is declared at the app level and repo at the controller level. Litestar resolves the graph — provide_repo receives the app-scoped db without any wiring on my part.
• The return type annotation is the response schema. Returning a VideoMeta or list[VideoMeta] triggers msgspec encoding and populates OpenAPI automatically.
• The batch cap of 64 is a cheap guardrail against someone turning /videos?ids= into a table scan.

With one connection per request and WAL mode, concurrent readers do not block each other, which is exactly what a read-mostly metadata service wants.

Where the CJK search lives: PHP and SQLite FTS5

The async service does not own search. Search is CJK-tokenized full-text over titles, and that lives in our PHP layer against the same SQLite file, because FTS5 is a SQLite feature and PHP's PDO speaks to it natively. The important design decision was to let each layer touch SQLite for what it is good at — Python for keyed reads and fan-out, PHP for FTS5 — over one WAL-mode database.

CJK text has no spaces, so a standard word tokenizer is useless. FTS5's trigram tokenizer indexes overlapping 3-character sequences, which gives usable substring matching for Japanese, Korean, and Chinese without a language-specific segmenter. Here is the table and a query as we run them in PHP 8.4:

<?php
declare(strict_types=1);

$db = new PDO('sqlite:/var/data/topvideohub.sqlite');
$db->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);

// trigram handles CJK substring matching where a word tokenizer cannot
$db->exec(<<<SQL
CREATE VIRTUAL TABLE IF NOT EXISTS videos_fts USING fts5(
    video_id UNINDEXED,
    title,
    title_localized,
    channel,
    tokenize = "trigram"
);
SQL);

function search_titles(PDO $db, string $q, int $limit = 20): array
{
    // trigram MATCH needs at least 3 chars; guard short queries
    if (mb_strlen($q) < 3) {
        return [];
    }
    $stmt = $db->prepare(
        'SELECT video_id, title, rank
         FROM videos_fts
         WHERE videos_fts MATCH :q
         ORDER BY rank
         LIMIT :n'
    );
    $stmt->bindValue(':q', '"' . str_replace('"', '', $q) . '"');
    $stmt->bindValue(':n', $limit, PDO::PARAM_INT);
    $stmt->execute();
    return $stmt->fetchAll(PDO::FETCH_ASSOC);
}

Quoting the match term and stripping embedded quotes matters: an unquoted FTS5 query treats characters like - and * as operators, and CJK titles are full of punctuation that would otherwise blow up the parser. rank is FTS5's built-in BM25 ordering, so nearer matches float up for free. PHP returns the matched IDs; the front end then calls the Litestar /videos?ids=... endpoint to hydrate full metadata. Two languages, one database, clean seam.

Structured concurrency for upstream enrichment

Now the part that motivated the rewrite. When a batch contains stale records, we refresh them from an upstream provider. In the old PHP path these calls ran one after another. Under asyncio we run them concurrently with a TaskGroup, which gives structured concurrency: if one task raises, the group cancels the siblings and propagates cleanly, so a single hung upstream cannot leak tasks.

import asyncio

import httpx

UPSTREAM = "https://provider.example/v"


async def enrich(ids: list[str], client: httpx.AsyncClient) -> dict[str, dict]:
    results: dict[str, dict] = {}
    sem = asyncio.Semaphore(8)  # cap concurrent upstream sockets

    async def fetch(vid: str) -> None:
        async with sem:
            try:
                resp = await client.get(f"{UPSTREAM}/{vid}", timeout=2.0)
                resp.raise_for_status()
                results[vid] = resp.json()
            except (httpx.HTTPError, asyncio.TimeoutError):
                # a stale record is better than a failed page render
                results[vid] = {}

    async with asyncio.TaskGroup() as tg:
        for vid in ids:
            tg.create_task(fetch(vid))
    return results


async def main() -> None:
    async with httpx.AsyncClient(
        limits=httpx.Limits(max_connections=32, max_keepalive_connections=16)
    ) as client:
        data = await enrich(["abc123", "def456", "ghi789"], client)
        print(len(data), "records refreshed")


if __name__ == "__main__":
    asyncio.run(main())

The semaphore is the load-bearing detail. Without it, a 40-item batch opens 40 upstream sockets at once and we get rate-limited or worse. With a cap of 8, wall-clock time for a batch drops to roughly the slowest 5 sequential rounds instead of the sum of all 40 — while the process still only uses one thread. Critically, the except clause degrades to an empty dict rather than failing the whole batch: on a video site, showing a slightly stale duration beats showing an error page.

Verifying it under concurrency

Because the whole point was surviving overlapping regional peaks, I load-test with a tiny Go program rather than trusting a single-request benchmark. Go's goroutines make it trivial to hammer the endpoint with real concurrency from one binary:

package main

import (
    "fmt"
    "net/http"
    "sync"
    "sync/atomic"
    "time"
)

func main() {
    const n = 500
    var wg sync.WaitGroup
    var ok int64
    start := time.Now()
    for i := 0; i < n; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            resp, err := http.Get("http://127.0.0.1:8000/videos?ids=abc123,def456")
            if err != nil {
                return
            }
            resp.Body.Close()
            if resp.StatusCode == 200 {
                atomic.AddInt64(&ok, 1)
            }
        }()
    }
    wg.Wait()
    fmt.Printf("%d/%d ok in %s\n", ok, n, time.Since(start))
}

Running the Litestar service under uvicorn with a couple of workers, 500 concurrent batch requests complete in well under a second on a modest VPS, with no worker starvation — because each request yields the event loop while it waits on SQLite or upstream I/O instead of blocking a whole worker the way the PHP path did.

Fitting it into the existing edge

The async service does not replace anything user-facing. It sits behind the same edge we already run:

• Cloudflare fronts everything; the metadata endpoints send Cache-Control: public, max-age=600, stale-while-revalidate=3600, so the vast majority of batch requests never reach origin.
• LiteSpeed still serves the PHP pages and the FTS5 search; it reverse-proxies /api/videos to the Litestar process on localhost.
• SQLite in WAL mode is shared read-only by the Python service and read/write by PHP cron jobs. One writer, many readers — SQLite's happy path.
• The Litestar app runs as a plain uvicorn systemd unit. No container, no orchestration; it is one Python process next to PHP.

The migration was incremental: I moved the batch metadata endpoint first, measured, then moved enrichment. Search never moved because FTS5 belongs in the SQLite/PHP layer. That is the pragmatic version of "rewrite in async" — move the part that is I/O-bound and fan-out heavy, leave the rest.

Conclusion

Litestar earned its place here not by being trendy but by matching the workload: msgspec makes the serialization-heavy hot path cheap, the layered DI keeps the connection and repository wiring honest, and controllers let a whole route group share cache and dependency policy without repetition. Paired with aiosqlite for keyed reads and asyncio.TaskGroup for bounded upstream fan-out, a single Python process now absorbs the evening overlap of Tokyo and Seoul traffic that used to starve our LiteSpeed workers. The CJK search stayed in PHP and FTS5 where it is strongest, and the two layers meet cleanly over one WAL-mode SQLite file behind Cloudflare. If you have a synchronous service dying on network fan-out, you probably do not need a full rewrite — you need to lift the I/O-bound endpoint into an async runtime and leave everything else alone. Litestar makes that lift small.