Building a Multi-Region Video API with Python FastAPI

FastAPI's combination of async support, automatic validation, and OpenAPI docs makes it a strong choice for a video platform API. Here's how I'd build the backend for TrendVidStream, covering 8 diverse global regions.

Domain Models

from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime
from enum import Enum


class Region(str, Enum):
    """TrendVidStream's supported regions."""
    AE = "AE"  # UAE
    FI = "FI"  # Finland
    DK = "DK"  # Denmark
    CZ = "CZ"  # Czech Republic
    BE = "BE"  # Belgium
    CH = "CH"  # Switzerland
    US = "US"  # United States
    GB = "GB"  # United Kingdom


# RTL_REGIONS: Regions where content may be right-to-left
RTL_REGIONS = {Region.AE}


class VideoResponse(BaseModel):
    video_id: str
    title: str  # May be Arabic (RTL) for AE region
    channel_title: str
    thumbnail_url: str
    view_count: int
    region: Region
    category_id: int
    language: Optional[str] = None  # ISO 639-1: 'ar', 'fi', 'da', 'cs', etc.
    published_at: datetime
    is_rtl: bool = False  # Convenience flag for frontend rendering

    @classmethod
    def from_row(cls, row: dict) -> 'VideoResponse':
        region = Region(row['region'])
        return cls(
            **{k: v for k, v in row.items() if k in cls.model_fields},
            is_rtl=region in RTL_REGIONS
        )


class TrendingResponse(BaseModel):
    region: Region
    videos: list[VideoResponse]
    total: int
    fetched_at: datetime
    has_rtl_content: bool = False  # True for AE region


class SearchResponse(BaseModel):
    query: str
    region: Optional[Region]
    results: list[VideoResponse]
    total: int
    page: int
    per_page: int

Database Dependency

import asyncpg
from fastapi import Depends
from typing import AsyncGenerator

_pool: asyncpg.Pool | None = None


async def init_pool(dsn: str) -> None:
    global _pool
    _pool = await asyncpg.create_pool(
        dsn, min_size=2, max_size=15, command_timeout=20
    )


async def get_conn() -> AsyncGenerator[asyncpg.Connection, None]:
    async with _pool.acquire() as conn:
        yield conn

Trending Endpoint with RTL Awareness

from fastapi import APIRouter, Depends, Query, HTTPException
import asyncpg

router = APIRouter(prefix="/trending", tags=["trending"])


@router.get("/{region}", response_model=TrendingResponse)
async def get_trending(
    region: Region,
    category_id: int | None = Query(default=None),
    limit: int = Query(default=50, ge=1, le=100),
    conn: asyncpg.Connection = Depends(get_conn),
):
    query = """
        SELECT v.video_id, v.title, v.channel_title, v.thumbnail_url,
               v.view_count, v.category_id, v.language, v.published_at,
               vr.region, vr.fetched_at
        FROM videos v
        JOIN video_regions vr ON vr.video_id = v.id
        WHERE vr.region = $1
    """
    params: list = [region.value]

    if category_id is not None:
        params.append(category_id)
        query += f" AND v.category_id = ${len(params)}"

    params.append(limit)
    query += f" ORDER BY vr.trending_rank ASC NULLS LAST, v.view_count DESC LIMIT ${len(params)}"

    rows = await conn.fetch(query, *params)
    if not rows:
        raise HTTPException(404, detail=f"No data for {region.value}")

    videos = [VideoResponse.from_row(dict(r)) for r in rows]
    return TrendingResponse(
        region=region,
        videos=videos,
        total=len(videos),
        fetched_at=rows[0]['fetched_at'],
        has_rtl_content=region in RTL_REGIONS,
    )

Multi-Language Search

@router.get("/search", response_model=SearchResponse)
async def search(
    q: str = Query(..., min_length=1, max_length=200),
    region: Region | None = Query(default=None),
    language: str | None = Query(default=None, description="ISO 639-1 language code: ar, fi, da, cs"),
    page: int = Query(default=1, ge=1),
    per_page: int = Query(default=20, ge=1, le=50),
    conn: asyncpg.Connection = Depends(get_conn),
):
    offset = (page - 1) * per_page

    base = """
        SELECT v.video_id, v.title, v.channel_title, v.thumbnail_url,
               v.view_count, v.category_id, v.language, v.region, v.published_at,
               ts_rank(v.search_vec,
                   to_tsquery('simple', plainto_tsquery('simple', $1)::text), 32) AS rank
        FROM videos v
        WHERE v.search_vec @@ to_tsquery('simple', plainto_tsquery('simple', $1)::text)
    """
    params: list = [q]

    if region:
        params.append(region.value)
        base += f" AND v.region = ${len(params)}"

    if language:
        params.append(language)
        base += f" AND v.language = ${len(params)}"

    params.extend([per_page, offset])
    base += f" ORDER BY rank DESC LIMIT ${len(params)-1} OFFSET ${len(params)}"

    rows = await conn.fetch(base, *params)
    videos = [VideoResponse.from_row(dict(r)) for r in rows]

    return SearchResponse(
        query=q, region=region,
        results=videos, total=len(videos),
        page=page, per_page=per_page,
    )

App Entry Point

from fastapi import FastAPI
from contextlib import asynccontextmanager
import os

@asynccontextmanager
async def lifespan(app: FastAPI):
    await init_pool(os.environ["DATABASE_URL"])
    yield
    if _pool: await _pool.close()

app = FastAPI(
    title="TrendVidStream API",
    description="Video trending data for UAE, Nordic, and Central European markets",
    version="1.0.0",
    lifespan=lifespan,
)
app.include_router(router)

The is_rtl and has_rtl_content flags in the response models give frontend clients a clear signal about Arabic content from UAE — they need to render titles with dir="rtl" in HTML. TrendVidStream's diverse region set means the API has to accommodate scripts, languages, and text directions that a single-market platform never encounters.

---

This article is part of the Building TrendVidStream series. Check out TrendVidStream to see these techniques in action.