feat(backend): add API routes
This commit is contained in:
3
backend/src/api/routes/__init__.py
Normal file
3
backend/src/api/routes/__init__.py
Normal file
@@ -0,0 +1,3 @@
|
||||
from src.api.routes import categories, patterns, questions, stats
|
||||
|
||||
__all__ = ["categories", "patterns", "questions", "stats"]
|
||||
39
backend/src/api/routes/categories.py
Normal file
39
backend/src/api/routes/categories.py
Normal 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)
|
||||
81
backend/src/api/routes/patterns.py
Normal file
81
backend/src/api/routes/patterns.py
Normal 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
|
||||
170
backend/src/api/routes/questions.py
Normal file
170
backend/src/api/routes/questions.py
Normal 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,
|
||||
)
|
||||
26
backend/src/api/routes/stats.py
Normal file
26
backend/src/api/routes/stats.py
Normal 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()
|
||||
Reference in New Issue
Block a user