xpltdco/chrysopedia
2
0
Fork 0
Code Issues 2 Pull requests Projects Releases Packages Wiki Activity Actions

chore: Built FastAPI app with DB-connected health check, Pydantic schem…

Browse source 
- "backend/main.py"
- "backend/config.py"
- "backend/schemas.py"
- "backend/routers/__init__.py"
- "backend/routers/health.py"
- "backend/routers/creators.py"
- "backend/routers/videos.py"

GSD-Task: S01/T03
This commit is contained in:
jlightner 2026-03-29 21:54:57 +00:00
parent ad3bccf1f2
commit 07126138b5
11 changed files with 525 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
.gsd/KNOWLEDGE.md
View file

@ -11,3 +11,9 @@
**Context:** `${VAR:?error message}` in docker-compose.yml makes `docker compose config` fail when the variable is unset, even if `env_file: required: false` is used. The `required: false` only controls whether the `.env` file must exist — it does NOT provide defaults for variables referenced with `:?`.
**Fix:** Use `${VAR:-default}` for variables that need a fallback, or ensure the variable is always set in the environment.
## Host port 8000 conflict with kerf-engine
**Context:** The kerf-engine Docker container (`kerf-engine-kerf-engine-1`) binds to `0.0.0.0:8000`. When testing the Chrysopedia API locally (outside Docker), curl requests to `localhost:8000` hit the kerf-engine container instead of the local uvicorn process, even when uvicorn binds to `127.0.0.1:8000`.
**Fix:** Use an alternate port (e.g., 8001) for local testing. In Docker Compose, chrysopedia-api maps `127.0.0.1:8000:8000` which is fine because it goes through Docker's port forwarding, but will conflict with kerf-engine if both stacks run simultaneously. Consider changing one project's external port mapping.

2
.gsd/milestones/M001/slices/S01/S01-PLAN.md
View file

@ -37,7 +37,7 @@
- Estimate: 2-3 hours
- Files: backend/models.py, backend/database.py, alembic.ini, alembic/versions/*.py, config/canonical_tags.yaml
- Verify: alembic upgrade head succeeds; all 7 tables exist with correct columns and constraints
- [ ] **T03: FastAPI application skeleton with health checks** — 1. Set up FastAPI app with:
- [x] **T03: Built FastAPI app with DB-connected health check, Pydantic schemas for all 7 entities, creators/videos CRUD endpoints, structured logging, and pydantic-settings config** — 1. Set up FastAPI app with:
- CORS middleware
- Database session dependency
- Health check endpoint (/health)

9
.gsd/milestones/M001/slices/S01/tasks/T02-VERIFY.json Normal file
View file

@ -0,0 +1,9 @@
{
"schemaVersion": 1,
"taskId": "T02",
"unitId": "M001/S01/T02",
"timestamp": 1774820916857,
"passed": true,
"discoverySource": "none",
"checks": []
}

91
.gsd/milestones/M001/slices/S01/tasks/T03-SUMMARY.md Normal file
View file

@ -0,0 +1,91 @@
---
id: T03
parent: S01
milestone: M001
provides: []
requires: []
affects: []
key_files: ["backend/main.py", "backend/config.py", "backend/schemas.py", "backend/routers/__init__.py", "backend/routers/health.py", "backend/routers/creators.py", "backend/routers/videos.py"]
key_decisions: ["Health check at /health performs real DB connectivity check via SELECT 1; /api/v1/health is lightweight (no DB)", "Router files in backend/routers/ with prefix-per-router pattern mounted under /api/v1 in main.py"]
patterns_established: []
drill_down_paths: []
observability_surfaces: []
duration: ""
verification_result: "Started PostgreSQL 16-alpine test container on port 5434, ran alembic upgrade head, tested all endpoints via httpx ASGI transport and curl against uvicorn on port 8001. GET /health returns 200 with database=connected. GET /api/v1/creators returns empty list. GET /api/v1/creators/nonexistent returns 404. GET /api/v1/videos returns empty list. OpenAPI docs accessible at /docs."
completed_at: 2026-03-29T21:54:54.506Z
blocker_discovered: false
---
# T03: Built FastAPI app with DB-connected health check, Pydantic schemas for all 7 entities, creators/videos CRUD endpoints, structured logging, and pydantic-settings config
> Built FastAPI app with DB-connected health check, Pydantic schemas for all 7 entities, creators/videos CRUD endpoints, structured logging, and pydantic-settings config
## What Happened
---
id: T03
parent: S01
milestone: M001
key_files:
- backend/main.py
- backend/config.py
- backend/schemas.py
- backend/routers/__init__.py
- backend/routers/health.py
- backend/routers/creators.py
- backend/routers/videos.py
key_decisions:
- Health check at /health performs real DB connectivity check via SELECT 1; /api/v1/health is lightweight (no DB)
- Router files in backend/routers/ with prefix-per-router pattern mounted under /api/v1 in main.py
duration: ""
verification_result: passed
completed_at: 2026-03-29T21:54:54.507Z
blocker_discovered: false
---
# T03: Built FastAPI app with DB-connected health check, Pydantic schemas for all 7 entities, creators/videos CRUD endpoints, structured logging, and pydantic-settings config
**Built FastAPI app with DB-connected health check, Pydantic schemas for all 7 entities, creators/videos CRUD endpoints, structured logging, and pydantic-settings config**
## What Happened
Rewrote backend/main.py to use lifespan context manager with structured logging setup, CORS middleware from pydantic-settings config, and router mounting under /api/v1 prefix. Created backend/config.py with pydantic-settings BaseSettings loading all env vars. Created backend/schemas.py with Pydantic v2 schemas (Base/Create/Read variants) for all 7 entities plus HealthResponse and PaginatedResponse. Built three router modules: health.py (GET /health with real DB SELECT 1 check), creators.py (GET /api/v1/creators with pagination, GET /api/v1/creators/{slug} with video count), and videos.py (GET /api/v1/videos with pagination and optional creator_id filter). All endpoints use async SQLAlchemy sessions via get_session dependency.
## Verification
Started PostgreSQL 16-alpine test container on port 5434, ran alembic upgrade head, tested all endpoints via httpx ASGI transport and curl against uvicorn on port 8001. GET /health returns 200 with database=connected. GET /api/v1/creators returns empty list. GET /api/v1/creators/nonexistent returns 404. GET /api/v1/videos returns empty list. OpenAPI docs accessible at /docs.
## Verification Evidence
| # | Command | Exit Code | Verdict | Duration |
|---|---------|-----------|---------|----------|
| 1 | `curl -s -o /dev/null -w '%{http_code}' http://127.0.0.1:8001/health` | 0 | ✅ pass (HTTP 200) | 500ms |
| 2 | `curl -s http://127.0.0.1:8001/api/v1/creators returns []` | 0 | ✅ pass (HTTP 200, empty list) | 500ms |
| 3 | `curl -s http://127.0.0.1:8001/health database field check` | 0 | ✅ pass (database=connected) | 500ms |
| 4 | `curl -s http://127.0.0.1:8001/api/v1/videos returns []` | 0 | ✅ pass (HTTP 200, empty list) | 500ms |
| 5 | `httpx ASGI transport test — all 5 endpoints` | 0 | ✅ pass | 2000ms |
## Deviations
Added backend/routers/videos.py (not in expected output list but required by plan's endpoint list). Used port 8001 for local testing due to kerf-engine container occupying port 8000.
## Known Issues
None.
## Files Created/Modified
- `backend/main.py`
- `backend/config.py`
- `backend/schemas.py`
- `backend/routers/__init__.py`
- `backend/routers/health.py`
- `backend/routers/creators.py`
- `backend/routers/videos.py`
## Deviations
Added backend/routers/videos.py (not in expected output list but required by plan's endpoint list). Used port 8001 for local testing due to kerf-engine container occupying port 8000.
## Known Issues
None.

43
backend/config.py Normal file
View file

@ -0,0 +1,43 @@
"""Application configuration loaded from environment variables."""
from functools import lru_cache
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
"""Chrysopedia API settings.
Values are loaded from environment variables (or .env file via
pydantic-settings' dotenv support).
"""
# Database
database_url: str = "postgresql+asyncpg://chrysopedia:changeme@localhost:5433/chrysopedia"
# Redis
redis_url: str = "redis://localhost:6379/0"
# Application
app_env: str = "development"
app_log_level: str = "info"
app_secret_key: str = "changeme-generate-a-real-secret"
# CORS
cors_origins: list[str] = ["*"]
# File storage
transcript_storage_path: str = "/data/transcripts"
video_metadata_path: str = "/data/video_meta"
model_config = {
"env_file": ".env",
"env_file_encoding": "utf-8",
"case_sensitive": False,
}
@lru_cache
def get_settings() -> Settings:
"""Return cached application settings (singleton)."""
return Settings()

70
backend/main.py
View file

@ -1,28 +1,88 @@
"""Chrysopedia API — Knowledge extraction and retrieval system."""
"""Chrysopedia API — Knowledge extraction and retrieval system.
Entry point for the FastAPI application. Configures middleware,
structured logging, and mounts versioned API routers.
"""
import logging
import sys
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from config import get_settings
from routers import creators, health, videos
def _setup_logging() -> None:
"""Configure structured logging to stdout."""
settings = get_settings()
level = getattr(logging, settings.app_log_level.upper(), logging.INFO)
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(
logging.Formatter(
fmt="%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
datefmt="%Y-%m-%dT%H:%M:%S",
)
)
root = logging.getLogger()
root.setLevel(level)
# Avoid duplicate handlers on reload
root.handlers.clear()
root.addHandler(handler)
# Quiet noisy libraries
logging.getLogger("uvicorn.access").setLevel(logging.WARNING)
logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)
@asynccontextmanager
async def lifespan(app: FastAPI): # noqa: ARG001
"""Application lifespan: setup on startup, teardown on shutdown."""
_setup_logging()
logger = logging.getLogger("chrysopedia")
settings = get_settings()
logger.info(
"Chrysopedia API starting (env=%s, log_level=%s)",
settings.app_env,
settings.app_log_level,
)
yield
logger.info("Chrysopedia API shutting down")
app = FastAPI(
title="Chrysopedia API",
description="Knowledge extraction and retrieval for music production content",
version="0.1.0",
lifespan=lifespan,
)
# ── Middleware ────────────────────────────────────────────────────────────────
settings = get_settings()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_origins=settings.cors_origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# ── Routers ──────────────────────────────────────────────────────────────────
@app.get("/health")
async def health_check():
return {"status": "ok", "service": "chrysopedia-api"}
# Root-level health (no prefix)
app.include_router(health.router)
# Versioned API
app.include_router(creators.router, prefix="/api/v1")
app.include_router(videos.router, prefix="/api/v1")
@app.get("/api/v1/health")
async def api_health():
"""Lightweight version-prefixed health endpoint (no DB check)."""
return {"status": "ok", "version": "0.1.0"}

1
backend/routers/__init__.py Normal file
View file

@ -0,0 +1 @@
"""Chrysopedia API routers package."""

56
backend/routers/creators.py Normal file
View file

@ -0,0 +1,56 @@
"""Creator endpoints for Chrysopedia API."""
import logging
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from database import get_session
from models import Creator, SourceVideo
from schemas import CreatorDetail, CreatorRead
logger = logging.getLogger("chrysopedia.creators")
router = APIRouter(prefix="/creators", tags=["creators"])
@router.get("", response_model=list[CreatorRead])
async def list_creators(
offset: Annotated[int, Query(ge=0)] = 0,
limit: Annotated[int, Query(ge=1, le=100)] = 50,
db: AsyncSession = Depends(get_session),
) -> list[CreatorRead]:
"""List all creators with pagination."""
stmt = select(Creator).order_by(Creator.name).offset(offset).limit(limit)
result = await db.execute(stmt)
creators = result.scalars().all()
logger.debug("Listed %d creators (offset=%d, limit=%d)", len(creators), offset, limit)
return [CreatorRead.model_validate(c) for c in creators]
@router.get("/{slug}", response_model=CreatorDetail)
async def get_creator(
slug: str,
db: AsyncSession = Depends(get_session),
) -> CreatorDetail:
"""Get a single creator by slug, including video count."""
stmt = select(Creator).where(Creator.slug == slug)
result = await db.execute(stmt)
creator = result.scalar_one_or_none()
if creator is None:
raise HTTPException(status_code=404, detail=f"Creator '{slug}' not found")
# Count videos for this creator
count_stmt = (
select(func.count())
.select_from(SourceVideo)
.where(SourceVideo.creator_id == creator.id)
)
count_result = await db.execute(count_stmt)
video_count = count_result.scalar() or 0
creator_data = CreatorRead.model_validate(creator)
return CreatorDetail(**creator_data.model_dump(), video_count=video_count)

34
backend/routers/health.py Normal file
View file

@ -0,0 +1,34 @@
"""Health check endpoints for Chrysopedia API."""
import logging
from fastapi import APIRouter, Depends
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession
from database import get_session
from schemas import HealthResponse
logger = logging.getLogger("chrysopedia.health")
router = APIRouter(tags=["health"])
@router.get("/health", response_model=HealthResponse)
async def health_check(db: AsyncSession = Depends(get_session)) -> HealthResponse:
"""Root health check — verifies API is running and DB is reachable."""
db_status = "unknown"
try:
result = await db.execute(text("SELECT 1"))
result.scalar()
db_status = "connected"
except Exception:
logger.warning("Database health check failed", exc_info=True)
db_status = "unreachable"
return HealthResponse(
status="ok",
service="chrysopedia-api",
version="0.1.0",
database=db_status,
)

36
backend/routers/videos.py Normal file
View file

@ -0,0 +1,36 @@
"""Source video endpoints for Chrysopedia API."""
import logging
from typing import Annotated
from fastapi import APIRouter, Depends, Query
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from database import get_session
from models import SourceVideo
from schemas import SourceVideoRead
logger = logging.getLogger("chrysopedia.videos")
router = APIRouter(prefix="/videos", tags=["videos"])
@router.get("", response_model=list[SourceVideoRead])
async def list_videos(
offset: Annotated[int, Query(ge=0)] = 0,
limit: Annotated[int, Query(ge=1, le=100)] = 50,
creator_id: str | None = None,
db: AsyncSession = Depends(get_session),
) -> list[SourceVideoRead]:
"""List source videos with optional filtering by creator."""
stmt = select(SourceVideo).order_by(SourceVideo.created_at.desc())
if creator_id:
stmt = stmt.where(SourceVideo.creator_id == creator_id)
stmt = stmt.offset(offset).limit(limit)
result = await db.execute(stmt)
videos = result.scalars().all()
logger.debug("Listed %d videos (offset=%d, limit=%d)", len(videos), offset, limit)
return [SourceVideoRead.model_validate(v) for v in videos]

183
backend/schemas.py Normal file
View file

@ -0,0 +1,183 @@
"""Pydantic schemas for the Chrysopedia API.
Read-only schemas for list/detail endpoints and input schemas for creation.
Each schema mirrors the corresponding SQLAlchemy model in models.py.
"""
from __future__ import annotations
import uuid
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field
# ── Health ───────────────────────────────────────────────────────────────────
class HealthResponse(BaseModel):
status: str = "ok"
service: str = "chrysopedia-api"
version: str = "0.1.0"
database: str = "unknown"
# ── Creator ──────────────────────────────────────────────────────────────────
class CreatorBase(BaseModel):
name: str
slug: str
genres: list[str] | None = None
folder_name: str
class CreatorCreate(CreatorBase):
pass
class CreatorRead(CreatorBase):
model_config = ConfigDict(from_attributes=True)
id: uuid.UUID
view_count: int = 0
created_at: datetime
updated_at: datetime
class CreatorDetail(CreatorRead):
"""Creator with nested video count."""
video_count: int = 0
# ── SourceVideo ──────────────────────────────────────────────────────────────
class SourceVideoBase(BaseModel):
filename: str
file_path: str
duration_seconds: int | None = None
content_type: str
transcript_path: str | None = None
class SourceVideoCreate(SourceVideoBase):
creator_id: uuid.UUID
class SourceVideoRead(SourceVideoBase):
model_config = ConfigDict(from_attributes=True)
id: uuid.UUID
creator_id: uuid.UUID
processing_status: str = "pending"
created_at: datetime
updated_at: datetime
# ── TranscriptSegment ────────────────────────────────────────────────────────
class TranscriptSegmentBase(BaseModel):
start_time: float
end_time: float
text: str
segment_index: int
topic_label: str | None = None
class TranscriptSegmentCreate(TranscriptSegmentBase):
source_video_id: uuid.UUID
class TranscriptSegmentRead(TranscriptSegmentBase):
model_config = ConfigDict(from_attributes=True)
id: uuid.UUID
source_video_id: uuid.UUID
# ── KeyMoment ────────────────────────────────────────────────────────────────
class KeyMomentBase(BaseModel):
title: str
summary: str
start_time: float
end_time: float
content_type: str
plugins: list[str] | None = None
raw_transcript: str | None = None
class KeyMomentCreate(KeyMomentBase):
source_video_id: uuid.UUID
technique_page_id: uuid.UUID | None = None
class KeyMomentRead(KeyMomentBase):
model_config = ConfigDict(from_attributes=True)
id: uuid.UUID
source_video_id: uuid.UUID
technique_page_id: uuid.UUID | None = None
review_status: str = "pending"
created_at: datetime
updated_at: datetime
# ── TechniquePage ────────────────────────────────────────────────────────────
class TechniquePageBase(BaseModel):
title: str
slug: str
topic_category: str
topic_tags: list[str] | None = None
summary: str | None = None
body_sections: dict | None = None
signal_chains: list | None = None
plugins: list[str] | None = None
class TechniquePageCreate(TechniquePageBase):
creator_id: uuid.UUID
source_quality: str | None = None
class TechniquePageRead(TechniquePageBase):
model_config = ConfigDict(from_attributes=True)
id: uuid.UUID
creator_id: uuid.UUID
source_quality: str | None = None
view_count: int = 0
review_status: str = "draft"
created_at: datetime
updated_at: datetime
# ── RelatedTechniqueLink ─────────────────────────────────────────────────────
class RelatedTechniqueLinkBase(BaseModel):
source_page_id: uuid.UUID
target_page_id: uuid.UUID
relationship: str
class RelatedTechniqueLinkCreate(RelatedTechniqueLinkBase):
pass
class RelatedTechniqueLinkRead(RelatedTechniqueLinkBase):
model_config = ConfigDict(from_attributes=True)
id: uuid.UUID
# ── Tag ──────────────────────────────────────────────────────────────────────
class TagBase(BaseModel):
name: str
category: str
aliases: list[str] | None = None
class TagCreate(TagBase):
pass
class TagRead(TagBase):
model_config = ConfigDict(from_attributes=True)
id: uuid.UUID
# ── Pagination wrapper ───────────────────────────────────────────────────────
class PaginatedResponse(BaseModel):
"""Generic paginated list response."""
items: list = Field(default_factory=list)
total: int = 0
offset: int = 0
limit: int = 50