A Video API for European Content Discovery
ViralVidVault serves trending videos from 7 regions including Poland, Netherlands, Sweden, Norway, and Austria. The API needs to handle region filtering, cross-language search, and engagement-based ranking. FastAPI is an excellent fit — async-native, auto-documented, and Pydantic validation handles the complex query parameters cleanly.
Data Models
from pydantic import BaseModel, Field
from datetime import datetime
from enum import Enum
class EuropeanRegion(str, Enum):
US = "US"
GB = "GB"
PL = "PL"
NL = "NL"
SE = "SE"
NO = "NO"
AT = "AT"
class VideoResponse(BaseModel):
video_id: str
title: str
channel_title: str
views: int
likes: int = 0
thumbnail_url: str
regions: list[str]
engagement_rate: float = 0.0
content_language: str = "en"
class TrendingFeed(BaseModel):
region: str
region_name: str
videos: list[VideoResponse]
total: int
updated_at: datetime
class SearchResults(BaseModel):
query: str
videos: list[VideoResponse]
total: int
page: int
languages_found: list[str]
REGION_NAMES = {
"US": "United States", "GB": "United Kingdom",
"PL": "Poland", "NL": "Netherlands",
"SE": "Sweden", "NO": "Norway", "AT": "Austria",
}
The EuropeanRegion enum restricts API inputs to our valid regions. FastAPI automatically returns a 422 error if someone passes regionCode=XX.
Database Layer with aiosqlite
import aiosqlite
from contextlib import asynccontextmanager
DB_PATH = "data/videos.db"
@asynccontextmanager
async def get_db():
db = await aiosqlite.connect(DB_PATH)
db.row_factory = aiosqlite.Row
await db.execute("PRAGMA journal_mode=WAL")
await db.execute("PRAGMA cache_size=-8000")
try:
yield db
finally:
await db.close()
async def get_trending_by_region(region: str, limit: int = 25) -> list[dict]:
async with get_db() as db:
cursor = await db.execute("""
SELECT v.video_id, v.title, v.channel_title, v.views, v.likes,
v.thumbnail_url, v.duration,
GROUP_CONCAT(DISTINCT vr.region) as regions
FROM videos v
JOIN video_regions vr ON v.video_id = vr.video_id
WHERE vr.region = ?
GROUP BY v.video_id
ORDER BY v.views DESC
LIMIT ?
""", (region, limit))
rows = await cursor.fetchall()
return [
{**dict(r), "regions": r["regions"].split(",") if r["regions"] else []}
for r in rows
]
async def search_videos(query: str, region: str | None, page: int, per_page: int):
async with get_db() as db:
base_query = """
FROM videos v
JOIN videos_fts fts ON v.video_id = fts.video_id
LEFT JOIN video_regions vr ON v.video_id = vr.video_id
WHERE videos_fts MATCH ?
"""
params = [query]
if region:
base_query += " AND vr.region = ?"
params.append(region)
# Count
cursor = await db.execute(f"SELECT COUNT(DISTINCT v.video_id) {base_query}", params)
total = (await cursor.fetchone())[0]
# Fetch page
offset = (page - 1) * per_page
cursor = await db.execute(f"""
SELECT DISTINCT v.video_id, v.title, v.channel_title, v.views,
v.likes, v.thumbnail_url,
GROUP_CONCAT(DISTINCT vr2.region) as regions
{base_query.replace("LEFT JOIN video_regions vr", "LEFT JOIN video_regions vr")}
LEFT JOIN video_regions vr2 ON v.video_id = vr2.video_id
GROUP BY v.video_id
ORDER BY rank
LIMIT ? OFFSET ?
""", params + [per_page, offset])
rows = await cursor.fetchall()
return [
{**dict(r), "regions": r["regions"].split(",") if r["regions"] else []}
for r in rows
], total
API Endpoints
from fastapi import FastAPI, Query, HTTPException
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI(
title="ViralVidVault API",
description="European viral video discovery API",
version="1.0.0",
)
app.add_middleware(
CORSMiddleware,
allow_origins=["https://viralvidvault.com"],
allow_methods=["GET"],
)
@app.get("/api/trending/{region}", response_model=TrendingFeed)
async def trending(region: EuropeanRegion, limit: int = Query(25, ge=1, le=50)):
videos = await get_trending_by_region(region.value, limit)
return TrendingFeed(
region=region.value,
region_name=REGION_NAMES.get(region.value, region.value),
videos=videos,
total=len(videos),
updated_at=datetime.utcnow(),
)
@app.get("/api/search", response_model=SearchResults)
async def search(
q: str = Query(..., min_length=2, max_length=100),
region: EuropeanRegion | None = None,
page: int = Query(1, ge=1, le=100),
per_page: int = Query(20, ge=1, le=50),
):
videos, total = await search_videos(q, region.value if region else None, page, per_page)
languages = list(set(v.get("content_language", "en") for v in videos))
return SearchResults(
query=q, videos=videos, total=total,
page=page, languages_found=languages,
)
@app.get("/api/regions")
async def list_regions():
return [
{"code": code, "name": name}
for code, name in REGION_NAMES.items()
]
Caching with In-Memory TTL
from datetime import datetime, timedelta
_cache: dict[str, tuple[datetime, any]] = {}
async def cached_trending(region: str) -> list[dict]:
key = f"trending:{region}"
if key in _cache:
expires, data = _cache[key]
if datetime.utcnow() < expires:
return data
data = await get_trending_by_region(region, 25)
_cache[key] = (datetime.utcnow() + timedelta(hours=3), data)
return data
Trending feeds update every 7 hours at ViralVidVault, so a 3-hour cache TTL ensures fresh data while eliminating redundant queries.
Running It
pip install fastapi uvicorn aiosqlite
uvicorn main:app --host 0.0.0.0 --port 8000
Visit /docs for auto-generated interactive API documentation. Every endpoint, parameter, and response model is documented automatically from the Pydantic types.
This article is part of the Building ViralVidVault series. Check out ViralVidVault to see these techniques in action.
Top comments (0)