feat(backend): add API routes

This commit is contained in:
2025-04-20 17:52:49 +01:00
parent 375787aa60
commit 2f0e2a40a6
6 changed files with 319 additions and 0 deletions

View File

View File

@@ -0,0 +1,3 @@
from src.api.routes import categories, patterns, questions, stats
__all__ = ["categories", "patterns", "questions", "stats"]

View File

@@ -0,0 +1,39 @@
from collections.abc import AsyncGenerator
from typing import Annotated
from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from src.db.database import get_db
from src.schemas import CategoryListResponse, CategoryResponse
from src.services import CategoryService
router = APIRouter()
async def get_category_service(
db: Annotated[AsyncSession, Depends(get_db)],
) -> AsyncGenerator[CategoryService, None]:
"""Dependency for CategoryService."""
yield CategoryService(db)
@router.get("", response_model=CategoryListResponse)
async def list_categories(
service: Annotated[CategoryService, Depends(get_category_service)],
) -> CategoryListResponse:
"""List all categories with question counts."""
categories_with_counts = await service.get_categories()
items = [
CategoryResponse(
id=category.id,
name=category.name,
slug=category.slug,
description=category.description,
question_count=count,
)
for category, count in categories_with_counts
]
return CategoryListResponse(items=items)

View File

@@ -0,0 +1,81 @@
from collections.abc import AsyncGenerator
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from src.db.database import get_db
from src.schemas import (
PatternDetailResponse,
PatternListResponse,
PatternResponse,
PatternTutorialResponse,
)
from src.services import PatternService
router = APIRouter()
async def get_pattern_service(
db: Annotated[AsyncSession, Depends(get_db)],
) -> AsyncGenerator[PatternService, None]:
"""Dependency for PatternService."""
yield PatternService(db)
@router.get("", response_model=PatternListResponse)
async def list_patterns(
service: Annotated[PatternService, Depends(get_pattern_service)],
) -> PatternListResponse:
"""List all patterns with question counts."""
patterns_with_counts = await service.get_patterns()
items = [
PatternResponse(
id=pattern.id,
name=pattern.name,
slug=pattern.slug,
description=pattern.description,
when_to_use=pattern.when_to_use,
question_count=count,
)
for pattern, count in patterns_with_counts
]
return PatternListResponse(items=items)
@router.get("/{slug}", response_model=PatternDetailResponse)
async def get_pattern(
slug: str,
service: Annotated[PatternService, Depends(get_pattern_service)],
) -> PatternDetailResponse:
"""Get a single pattern by slug with details."""
result = await service.get_pattern_by_slug(slug)
if not result:
raise HTTPException(status_code=404, detail="Pattern not found")
pattern, count = result
return PatternDetailResponse(
id=pattern.id,
name=pattern.name,
slug=pattern.slug,
description=pattern.description,
when_to_use=pattern.when_to_use,
question_count=count,
)
@router.get("/{slug}/tutorial", response_model=PatternTutorialResponse)
async def get_pattern_tutorial(
slug: str,
service: Annotated[PatternService, Depends(get_pattern_service)],
) -> PatternTutorialResponse:
"""Get full pattern tutorial with educational content and learning progression."""
tutorial = await service.get_pattern_tutorial(slug)
if not tutorial:
raise HTTPException(status_code=404, detail="Pattern not found")
return tutorial

View File

@@ -0,0 +1,170 @@
from collections.abc import AsyncGenerator
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.ext.asyncio import AsyncSession
from src.config import get_settings
from src.db.database import get_db
from src.models import Difficulty
from src.schemas import (
QuestionDetail,
QuestionListItem,
QuestionListResponse,
SubmissionRequest,
SubmissionResponse,
TestResult,
)
from src.services import QuestionService
router = APIRouter()
settings = get_settings()
async def get_question_service(
db: Annotated[AsyncSession, Depends(get_db)],
) -> AsyncGenerator[QuestionService, None]:
yield QuestionService(db)
@router.get("", response_model=QuestionListResponse)
async def list_questions(
service: Annotated[QuestionService, Depends(get_question_service)],
page: Annotated[int, Query(ge=1)] = 1,
limit: Annotated[int, Query(ge=1, le=100)] = 20,
difficulty: Annotated[str | None, Query(description="Comma-separated difficulties")] = None,
category: Annotated[str | None, Query(description="Category slug")] = None,
pattern: Annotated[str | None, Query(description="Pattern slug")] = None,
search: Annotated[str | None, Query(max_length=100, description="Search in title")] = None,
) -> QuestionListResponse:
difficulties: list[Difficulty] | None = None
if difficulty:
try:
difficulties = [Difficulty(d.strip()) for d in difficulty.split(",")]
except ValueError as e:
raise HTTPException(
status_code=400,
detail="Invalid difficulty value. Must be: easy, medium, hard",
) from e
questions, total = await service.get_questions(
page=page,
limit=limit,
difficulties=difficulties,
category_slug=category,
pattern_slug=pattern,
search=search,
)
pages = (total + limit - 1) // limit if total > 0 else 0
return QuestionListResponse(
items=[QuestionListItem.model_validate(q) for q in questions],
total=total,
page=page,
limit=limit,
pages=pages,
)
@router.get("/{slug}", response_model=QuestionDetail)
async def get_question(
slug: str,
service: Annotated[QuestionService, Depends(get_question_service)],
) -> QuestionDetail:
question = await service.get_question_by_slug(slug)
if not question:
raise HTTPException(status_code=404, detail="Question not found")
return QuestionDetail.model_validate(question)
def _json_sort_key(x: object) -> str:
import json
return json.dumps(x, sort_keys=True, default=str)
def compare_outputs(actual: object, expected: object) -> bool:
"""Compare test outputs with flexible matching for unordered arrays and nested structures."""
import json
try:
if json.dumps(actual, sort_keys=True) == json.dumps(expected, sort_keys=True):
return True
except (TypeError, ValueError):
pass
if actual == expected:
return True
if isinstance(actual, list) and isinstance(expected, list):
if len(actual) != len(expected):
return False
try:
sorted_actual = sorted(actual, key=_json_sort_key)
sorted_expected = sorted(expected, key=_json_sort_key)
return json.dumps(sorted_actual) == json.dumps(sorted_expected)
except (TypeError, ValueError):
return sorted(str(x) for x in actual) == sorted(str(x) for x in expected)
return False
@router.post("/{slug}/submit", response_model=SubmissionResponse)
async def submit_solution(
slug: str,
submission: SubmissionRequest,
service: Annotated[QuestionService, Depends(get_question_service)],
) -> SubmissionResponse:
"""Validate user's solution outputs against hidden test expected values."""
question = await service.get_question_by_slug(slug)
if not question:
raise HTTPException(status_code=404, detail="Question not found")
if not question.test_cases:
raise HTTPException(status_code=400, detail="Question has no test cases")
visible_tests = question.test_cases.get("visible", [])
hidden_tests = question.test_cases.get("hidden", [])
output_lookup = {ho.test_id: ho.output for ho in submission.hidden_outputs}
visible_results: list[TestResult] = []
hidden_results: list[TestResult] = []
for i, test in enumerate(hidden_tests):
test_id = len(visible_tests) + i
user_output = output_lookup.get(test_id)
expected = test["expected"]
if user_output is None:
hidden_results.append(
TestResult(
test_id=test_id,
passed=False,
actual=None,
error="No output provided",
)
)
else:
passed = compare_outputs(user_output, expected)
hidden_results.append(
TestResult(
test_id=test_id,
passed=passed,
actual=user_output,
)
)
total_passed = sum(1 for r in hidden_results if r.passed)
total_tests = len(hidden_results)
return SubmissionResponse(
passed=total_passed == total_tests,
visible_results=visible_results,
hidden_results=hidden_results,
total_passed=total_passed,
total_tests=total_tests,
)

View File

@@ -0,0 +1,26 @@
from collections.abc import AsyncGenerator
from typing import Annotated
from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from src.db.database import get_db
from src.schemas import StatsResponse
from src.services import StatsService
router = APIRouter()
async def get_stats_service(
db: Annotated[AsyncSession, Depends(get_db)],
) -> AsyncGenerator[StatsService, None]:
"""Dependency for StatsService."""
yield StatsService(db)
@router.get("", response_model=StatsResponse)
async def get_stats(
service: Annotated[StatsService, Depends(get_stats_service)],
) -> StatsResponse:
"""Get question counts by difficulty, category, and pattern."""
return await service.get_stats()