c0df369018
- "backend/search_service.py" - "backend/schemas.py" - "backend/routers/search.py" - "backend/routers/techniques.py" - "backend/routers/topics.py" - "backend/routers/creators.py" - "backend/main.py" GSD-Task: S05/T01
134 lines
4.4 KiB
Python
134 lines
4.4 KiB
Python
"""Technique page endpoints — list and detail with eager-loaded relations."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Annotated
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from database import get_session
|
|
from models import Creator, KeyMoment, RelatedTechniqueLink, TechniquePage
|
|
from schemas import (
|
|
CreatorInfo,
|
|
KeyMomentSummary,
|
|
PaginatedResponse,
|
|
RelatedLinkItem,
|
|
TechniquePageDetail,
|
|
TechniquePageRead,
|
|
)
|
|
|
|
logger = logging.getLogger("chrysopedia.techniques")
|
|
|
|
router = APIRouter(prefix="/techniques", tags=["techniques"])
|
|
|
|
|
|
@router.get("", response_model=PaginatedResponse)
|
|
async def list_techniques(
|
|
category: Annotated[str | None, Query()] = None,
|
|
creator_slug: Annotated[str | None, Query()] = None,
|
|
offset: Annotated[int, Query(ge=0)] = 0,
|
|
limit: Annotated[int, Query(ge=1, le=100)] = 50,
|
|
db: AsyncSession = Depends(get_session),
|
|
) -> PaginatedResponse:
|
|
"""List technique pages with optional category/creator filtering."""
|
|
stmt = select(TechniquePage)
|
|
|
|
if category:
|
|
stmt = stmt.where(TechniquePage.topic_category == category)
|
|
|
|
if creator_slug:
|
|
# Join to Creator to filter by slug
|
|
stmt = stmt.join(Creator, TechniquePage.creator_id == Creator.id).where(
|
|
Creator.slug == creator_slug
|
|
)
|
|
|
|
# Count total before pagination
|
|
from sqlalchemy import func
|
|
|
|
count_stmt = select(func.count()).select_from(stmt.subquery())
|
|
count_result = await db.execute(count_stmt)
|
|
total = count_result.scalar() or 0
|
|
|
|
stmt = stmt.order_by(TechniquePage.created_at.desc()).offset(offset).limit(limit)
|
|
result = await db.execute(stmt)
|
|
pages = result.scalars().all()
|
|
|
|
return PaginatedResponse(
|
|
items=[TechniquePageRead.model_validate(p) for p in pages],
|
|
total=total,
|
|
offset=offset,
|
|
limit=limit,
|
|
)
|
|
|
|
|
|
@router.get("/{slug}", response_model=TechniquePageDetail)
|
|
async def get_technique(
|
|
slug: str,
|
|
db: AsyncSession = Depends(get_session),
|
|
) -> TechniquePageDetail:
|
|
"""Get full technique page detail with key moments, creator, and related links."""
|
|
stmt = (
|
|
select(TechniquePage)
|
|
.where(TechniquePage.slug == slug)
|
|
.options(
|
|
selectinload(TechniquePage.key_moments),
|
|
selectinload(TechniquePage.creator),
|
|
selectinload(TechniquePage.outgoing_links).selectinload(
|
|
RelatedTechniqueLink.target_page
|
|
),
|
|
selectinload(TechniquePage.incoming_links).selectinload(
|
|
RelatedTechniqueLink.source_page
|
|
),
|
|
)
|
|
)
|
|
result = await db.execute(stmt)
|
|
page = result.scalar_one_or_none()
|
|
|
|
if page is None:
|
|
raise HTTPException(status_code=404, detail=f"Technique '{slug}' not found")
|
|
|
|
# Build key moments (ordered by start_time)
|
|
key_moments = sorted(page.key_moments, key=lambda km: km.start_time)
|
|
key_moment_items = [KeyMomentSummary.model_validate(km) for km in key_moments]
|
|
|
|
# Build creator info
|
|
creator_info = None
|
|
if page.creator:
|
|
creator_info = CreatorInfo(
|
|
name=page.creator.name,
|
|
slug=page.creator.slug,
|
|
genres=page.creator.genres,
|
|
)
|
|
|
|
# Build related links (outgoing + incoming)
|
|
related_links: list[RelatedLinkItem] = []
|
|
for link in page.outgoing_links:
|
|
if link.target_page:
|
|
related_links.append(
|
|
RelatedLinkItem(
|
|
target_title=link.target_page.title,
|
|
target_slug=link.target_page.slug,
|
|
relationship=link.relationship.value if hasattr(link.relationship, 'value') else str(link.relationship),
|
|
)
|
|
)
|
|
for link in page.incoming_links:
|
|
if link.source_page:
|
|
related_links.append(
|
|
RelatedLinkItem(
|
|
target_title=link.source_page.title,
|
|
target_slug=link.source_page.slug,
|
|
relationship=link.relationship.value if hasattr(link.relationship, 'value') else str(link.relationship),
|
|
)
|
|
)
|
|
|
|
base = TechniquePageRead.model_validate(page)
|
|
return TechniquePageDetail(
|
|
**base.model_dump(),
|
|
key_moments=key_moment_items,
|
|
creator_info=creator_info,
|
|
related_links=related_links,
|
|
)
|