Building a Video API from Scratch
Every video platform needs an API — whether it serves a frontend SPA, a mobile app, or third-party integrations. At DailyWatch, our API powers video search, category browsing, and trending feeds across 8 regions. Here is how to build one with FastAPI.
Project Structure
video_api/
├── main.py
├── models.py
├── database.py
├── routers/
│ ├── videos.py
│ ├── categories.py
│ └── trending.py
Pydantic Models
Define your data shapes once, get validation and serialization for free:
from pydantic import BaseModel, Field
from datetime import datetime
from enum import Enum
class Region(str, Enum):
US = "US"
GB = "GB"
DE = "DE"
FR = "FR"
IN_ = "IN"
BR = "BR"
AU = "AU"
CA = "CA"
class VideoBase(BaseModel):
video_id: str
title: str
channel_title: str
category_id: int
thumbnail_url: str
views: int = 0
likes: int = 0
duration: int = 0
regions: list[str] = []
class VideoDetail(VideoBase):
description: str = ""
published_at: datetime | None = None
fetched_at: datetime
class VideoSearchResult(BaseModel):
videos: list[VideoBase]
total: int
page: int
per_page: int
query: str
class TrendingResponse(BaseModel):
region: str
videos: list[VideoBase]
updated_at: datetime
Pydantic validates incoming requests and serializes outgoing responses. The Region enum restricts region parameters to valid values automatically.
Async Database Layer
FastAPI is async-native, so our database layer should be too:
import aiosqlite
from contextlib import asynccontextmanager
DATABASE_PATH = "data/videos.db"
@asynccontextmanager
async def get_db():
db = await aiosqlite.connect(DATABASE_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 search_videos(query: str, page: int = 1, per_page: int = 20) -> tuple[list[dict], int]:
async with get_db() as db:
# Count total matches
cursor = await db.execute(
"SELECT COUNT(*) FROM videos_fts WHERE videos_fts MATCH ?",
(query,)
)
row = await cursor.fetchone()
total = row[0]
# Fetch page of results
offset = (page - 1) * per_page
cursor = await db.execute("""
SELECT v.video_id, v.title, v.channel_title, v.category_id,
v.thumbnail_url, v.views, v.likes, v.duration,
GROUP_CONCAT(vr.region) as regions
FROM videos v
JOIN videos_fts ON v.video_id = videos_fts.video_id
LEFT JOIN video_regions vr ON v.video_id = vr.video_id
WHERE videos_fts MATCH ?
GROUP BY v.video_id
ORDER BY rank
LIMIT ? OFFSET ?
""", (query, per_page, offset))
rows = await cursor.fetchall()
videos = [dict(row) for row in rows]
for v in videos:
v["regions"] = v["regions"].split(",") if v["regions"] else []
return videos, total
Using aiosqlite wraps SQLite in an async interface so database queries do not block the event loop.
API Endpoints
from fastapi import FastAPI, Query, HTTPException, Depends
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI(title="DailyWatch API", version="1.0.0")
app.add_middleware(
CORSMiddleware,
allow_origins=["https://dailywatch.video"],
allow_methods=["GET"],
allow_headers=["*"],
)
@app.get("/api/search", response_model=VideoSearchResult)
async def search(
q: str = Query(..., min_length=2, max_length=100, description="Search query"),
page: int = Query(1, ge=1, le=100),
per_page: int = Query(20, ge=1, le=50),
):
videos, total = await search_videos(q, page, per_page)
return VideoSearchResult(
videos=videos,
total=total,
page=page,
per_page=per_page,
query=q,
)
@app.get("/api/trending/{region}", response_model=TrendingResponse)
async def trending(region: Region):
async with get_db() as db:
cursor = await db.execute("""
SELECT v.video_id, v.title, v.channel_title, v.category_id,
v.thumbnail_url, v.views, v.likes, v.duration
FROM videos v
JOIN video_regions vr ON v.video_id = vr.video_id
WHERE vr.region = ?
ORDER BY v.views DESC
LIMIT 50
""", (region.value,))
rows = await cursor.fetchall()
return TrendingResponse(
region=region.value,
videos=[dict(r) for r in rows],
updated_at=datetime.utcnow(),
)
@app.get("/api/categories")
async def list_categories():
async with get_db() as db:
cursor = await db.execute(
"SELECT id, title FROM categories ORDER BY title"
)
return [dict(r) for r in await cursor.fetchall()]
Dependency Injection for Caching
FastAPI's dependency injection makes adding a cache layer clean:
from functools import lru_cache
from datetime import datetime, timedelta
_cache: dict[str, tuple[datetime, any]] = {}
async def cached_categories() -> list[dict]:
key = "categories"
if key in _cache:
expires, data = _cache[key]
if datetime.utcnow() < expires:
return data
async with get_db() as db:
cursor = await db.execute("SELECT id, title FROM categories ORDER BY title")
data = [dict(r) for r in await cursor.fetchall()]
_cache[key] = (datetime.utcnow() + timedelta(hours=24), data)
return data
@app.get("/api/categories")
async def list_categories(categories: list = Depends(cached_categories)):
return categories
Categories rarely change, so caching them for 24 hours eliminates a database query on every request.
Running and Testing
pip install fastapi uvicorn aiosqlite
uvicorn main:app --host 0.0.0.0 --port 8000 --reload
FastAPI auto-generates OpenAPI docs at /docs. For DailyWatch, this API serves as the backbone for both server-rendered pages and any future mobile or SPA frontend.
This article is part of the Building DailyWatch series. Check out DailyWatch to see these techniques in action.