Updated terminal client REPL to work with refactored GameState structure where current_batter/pitcher/catcher are now LineupPlayerState objects instead of integer IDs. Also standardized all documentation to properly show 'uv run' prefixes for Python commands. REPL Updates: - terminal_client/display.py: Access lineup_id from LineupPlayerState objects - terminal_client/repl.py: Fix typos (self.current_game → self.current_game_id) - tests/unit/terminal_client/test_commands.py: Create proper LineupPlayerState objects in test fixtures (2 tests fixed, all 105 terminal client tests passing) Documentation Updates (100+ command examples): - CLAUDE.md: Updated pytest examples to use 'uv run' prefix - terminal_client/CLAUDE.md: Updated ~40 command examples - tests/CLAUDE.md: Updated all test commands (unit, integration, debugging) - app/*/CLAUDE.md: Updated test and server startup commands (5 files) All Python commands now consistently use 'uv run' prefix to align with project's UV migration, improving developer experience and preventing confusion about virtual environment activation. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
23 KiB
Backend API Layer - REST Endpoints
Overview
The app/api/ directory contains REST API endpoints for the Paper Dynasty game backend. This layer provides traditional HTTP request/response endpoints for operations like authentication, game creation, health checks, and data retrieval.
Architecture Note: Most real-time gameplay happens via WebSocket (see app/websocket/). REST API is primarily for:
- Authentication & authorization
- Game lifecycle (create, list, retrieve)
- Health monitoring
- Data queries that don't require real-time updates
Directory Structure
app/api/
├── __init__.py # Empty package marker
├── routes/ # API route modules
│ ├── __init__.py # Empty package marker
│ ├── health.py # Health check endpoints (✅ complete)
│ ├── auth.py # Authentication endpoints (stub)
│ └── games.py # Game CRUD endpoints (stub)
└── CLAUDE.md # This file
Note: No dependencies.py file yet. Common dependencies (auth, DB sessions) will be added as needed during Phase 2+ implementation.
Current Implementation Status
✅ Fully Implemented
Health Endpoints (routes/health.py)
Purpose: Service health monitoring for deployment and operations
Endpoints:
-
GET
/api/health- Application health check- Returns: Service status, timestamp, environment, version
- Use: Load balancer health probes, monitoring systems
- Response time: < 10ms
-
GET
/api/health/db- Database connectivity check- Returns: Database connection status
- Use: Verify database availability
- Response time: < 50ms
- Graceful error handling (returns unhealthy status instead of 500)
Example Responses:
// GET /api/health
{
"status": "healthy",
"timestamp": "2025-10-31T14:23:45.123456+00:00",
"environment": "development",
"version": "1.0.0"
}
// GET /api/health/db (success)
{
"status": "healthy",
"database": "connected",
"timestamp": "2025-10-31T14:23:45.123456+00:00"
}
// GET /api/health/db (failure)
{
"status": "unhealthy",
"database": "disconnected",
"error": "connection refused",
"timestamp": "2025-10-31T14:23:45.123456+00:00"
}
🟡 Stub Implementation (Phase 1)
Authentication Endpoints (routes/auth.py)
Purpose: User authentication via Discord OAuth (planned for Phase 1, currently stub)
Endpoints:
-
POST
/api/auth/token- Create JWT token- Request:
TokenRequest(user_id, username, discord_id) - Response:
TokenResponse(access_token, token_type) - Current: Directly creates token from provided data (no OAuth yet)
- TODO: Implement full Discord OAuth flow
- Request:
-
GET
/api/auth/verify- Verify authentication status- Response: Stub acknowledgment
- TODO: Implement token verification with user context
TODO Items:
- Discord OAuth callback handler
- State parameter validation
- Token refresh endpoint
- User session management
- Authentication middleware/dependency
Game Endpoints (routes/games.py)
Purpose: Game lifecycle management (planned for Phase 2+, currently stub)
Endpoints:
-
GET
/api/games/- List all games- Response: List of
GameListItem(game_id, league_id, status, teams) - TODO: Implement with database query, pagination, filters
- Response: List of
-
GET
/api/games/{game_id}- Get game details- Response: Full game state and metadata
- TODO: Load from database, include plays, lineups, current state
-
POST
/api/games/- Create new game- Request: Game creation parameters (league, teams, settings)
- Response: Created game details
- TODO: Create game in DB, initialize state, return game_id
TODO Items:
- Game creation with validation
- Game listing with filters (by league, status, user)
- Game detail retrieval with related data
- Game deletion/cancellation
- Game history and statistics
FastAPI Patterns Used
1. APIRouter Pattern
Each route module creates its own APIRouter instance:
from fastapi import APIRouter
router = APIRouter()
@router.get("/endpoint")
async def handler():
...
Routers are registered in app/main.py:
from app.api.routes import health, auth, games
app.include_router(health.router, prefix="/api", tags=["health"])
app.include_router(auth.router, prefix="/api/auth", tags=["auth"])
app.include_router(games.router, prefix="/api/games", tags=["games"])
Conventions:
- Each module exports a
routervariable - Prefix and tags applied during registration in main.py
- Tags group endpoints in Swagger UI docs
2. Pydantic Request/Response Models
Use Pydantic models for type safety and validation:
from pydantic import BaseModel
class TokenRequest(BaseModel):
"""Request model for token creation"""
user_id: str
username: str
discord_id: str
class TokenResponse(BaseModel):
"""Response model for token creation"""
access_token: str
token_type: str = "bearer"
@router.post("/token", response_model=TokenResponse)
async def create_auth_token(request: TokenRequest):
# FastAPI automatically validates request body against TokenRequest
# and serializes return value according to TokenResponse
...
Benefits:
- Automatic request validation with clear error messages
- Automatic response serialization
- OpenAPI schema generation for docs
- Type hints for IDE support
3. Async Handlers
All route handlers use async def:
@router.get("/health/db")
async def database_health():
# Can use await for DB queries, external APIs, etc.
async with engine.connect() as conn:
await conn.execute(text("SELECT 1"))
...
Why Async:
- Non-blocking I/O operations (database, external APIs)
- Better concurrency for high-traffic endpoints
- Consistent with WebSocket handlers
- Required for async database operations (asyncpg)
4. Logging Pattern
Module-level logger with descriptive name:
import logging
logger = logging.getLogger(f'{__name__}.health')
@router.get("/endpoint")
async def handler():
logger.info("Endpoint called")
logger.error(f"Error occurred: {error}", exc_info=True)
Logging Levels:
DEBUG: Detailed diagnostic informationINFO: Normal operation events (endpoint calls, state changes)WARNING: Unusual but handled situationsERROR: Error conditions that need attentionCRITICAL: Severe errors requiring immediate action
5. Error Handling
Use FastAPI's HTTPException for error responses:
from fastapi import HTTPException
@router.post("/endpoint")
async def handler():
try:
# operation
return result
except ValidationError as e:
logger.error(f"Validation failed: {e}")
raise HTTPException(status_code=400, detail="Invalid request data")
except Exception as e:
logger.error(f"Unexpected error: {e}", exc_info=True)
raise HTTPException(status_code=500, detail="Internal server error")
Status Codes:
- 200: Success
- 201: Created
- 400: Bad request (validation error)
- 401: Unauthorized (missing/invalid token)
- 403: Forbidden (valid token, insufficient permissions)
- 404: Not found
- 500: Internal server error
6. Settings Injection
Use get_settings() for configuration:
from app.config import get_settings
settings = get_settings()
@router.get("/endpoint")
async def handler():
api_url = settings.sba_api_url
...
Available Settings:
app_env: "development", "staging", "production"debug: Enable debug modedatabase_url: PostgreSQL connection stringsecret_key: JWT signing keydiscord_client_id,discord_client_secret: OAuth credentialssba_api_url,pd_api_url: League API endpointscors_origins: Allowed frontend origins
Integration Points
With Other Backend Components
Database Layer (app/database/)
from app.database.session import get_session
from app.database.operations import DatabaseOperations
@router.get("/games/{game_id}")
async def get_game(game_id: str):
db_ops = DatabaseOperations()
game_data = await db_ops.load_game_state(game_id)
return game_data
Models (app/models/)
from app.models import GameState
from app.models.db_models import Game
@router.post("/games/")
async def create_game(request: CreateGameRequest):
# Use Pydantic models for validation
# Use SQLAlchemy models for database operations
...
Authentication (app/utils/auth.py)
from app.utils.auth import create_token, verify_token
@router.post("/auth/token")
async def create_auth_token(request: TokenRequest):
token = create_token(user_data)
return TokenResponse(access_token=token)
State Manager (app/core/state_manager.py)
from app.core.state_manager import state_manager
@router.get("/games/{game_id}")
async def get_game(game_id: str):
# Check in-memory state first (fast)
state = state_manager.get_state(game_id)
if not state:
# Fall back to database
state = await state_manager.recover_game(game_id)
return state
With Frontend
Frontend applications make HTTP requests to these endpoints:
// Example frontend code (Vue/Nuxt)
const api = axios.create({
baseURL: 'http://localhost:8000/api',
headers: {
'Authorization': `Bearer ${token}`
}
});
// Health check
const health = await api.get('/health');
// Create game
const game = await api.post('/games/', {
league_id: 'sba',
home_team_id: 1,
away_team_id: 2
});
// List games
const games = await api.get('/games/');
Common Tasks
Adding a New Endpoint
-
Choose appropriate route module (or create new one)
# If creating new module touch app/api/routes/teams.py -
Define Pydantic models for request/response
from pydantic import BaseModel class CreateTeamRequest(BaseModel): name: str league_id: str owner_id: str class TeamResponse(BaseModel): id: int name: str league_id: str created_at: str -
Create route handler
import logging from fastapi import APIRouter, HTTPException from app.database.operations import DatabaseOperations logger = logging.getLogger(f'{__name__}.teams') router = APIRouter() @router.post("/", response_model=TeamResponse) async def create_team(request: CreateTeamRequest): """Create a new team""" logger.info(f"Creating team: {request.name}") try: db_ops = DatabaseOperations() team = await db_ops.create_team( name=request.name, league_id=request.league_id, owner_id=request.owner_id ) return TeamResponse(**team) except Exception as e: logger.error(f"Failed to create team: {e}", exc_info=True) raise HTTPException(status_code=500, detail="Failed to create team") -
Register router in main.py
from app.api.routes import teams app.include_router(teams.router, prefix="/api/teams", tags=["teams"]) -
Test the endpoint
# Start server uv run python -m app.main # Test with curl curl -X POST http://localhost:8000/api/teams/ \ -H "Content-Type: application/json" \ -d '{"name": "Test Team", "league_id": "sba", "owner_id": "user123"}' # Or use Swagger UI # http://localhost:8000/docs
Adding Authentication Dependency
When authentication is implemented, add a dependency for protected endpoints:
from fastapi import Depends, HTTPException, Header
from app.utils.auth import verify_token
async def get_current_user(authorization: str = Header(...)):
"""Extract and verify user from JWT token"""
try:
scheme, token = authorization.split()
if scheme.lower() != 'bearer':
raise HTTPException(status_code=401, detail="Invalid auth scheme")
payload = verify_token(token)
return payload
except Exception as e:
raise HTTPException(status_code=401, detail="Invalid token")
@router.post("/games/")
async def create_game(
request: CreateGameRequest,
user = Depends(get_current_user) # Automatically extracts and validates token
):
# user contains decoded JWT payload
owner_id = user['user_id']
...
Adding Database Operations
For endpoints that need database access:
from app.database.operations import DatabaseOperations
from app.database.session import get_session
@router.get("/games/{game_id}")
async def get_game(game_id: str):
db_ops = DatabaseOperations()
# Load game with all related data
game_data = await db_ops.load_game_state(game_id)
if not game_data:
raise HTTPException(status_code=404, detail="Game not found")
return game_data
Adding Request Validation
Pydantic provides powerful validation:
from pydantic import BaseModel, Field, validator
class CreateGameRequest(BaseModel):
league_id: str = Field(..., regex="^(sba|pd)$")
home_team_id: int = Field(..., gt=0)
away_team_id: int = Field(..., gt=0)
game_mode: str = Field(default="friendly", regex="^(ranked|friendly|practice)$")
@validator('away_team_id')
def teams_must_differ(cls, v, values):
if 'home_team_id' in values and v == values['home_team_id']:
raise ValueError('Home and away teams must be different')
return v
Adding Query Parameters
For list endpoints with filters:
from typing import Optional
@router.get("/games/")
async def list_games(
league_id: Optional[str] = None,
status: Optional[str] = None,
limit: int = 50,
offset: int = 0
):
"""
List games with optional filters
Query params:
- league_id: Filter by league ('sba' or 'pd')
- status: Filter by status ('pending', 'active', 'completed')
- limit: Number of results (default 50, max 100)
- offset: Pagination offset (default 0)
"""
db_ops = DatabaseOperations()
games = await db_ops.list_games(
league_id=league_id,
status=status,
limit=min(limit, 100),
offset=offset
)
return games
Testing
Manual Testing with Swagger UI
FastAPI automatically generates interactive API documentation:
-
Start the backend server:
uv run python -m app.main -
Open Swagger UI:
http://localhost:8000/docs -
Features:
- View all endpoints grouped by tags
- See request/response schemas
- Try out endpoints with test data
- View response bodies and status codes
- Copy curl commands
Manual Testing with curl
# Health check
curl http://localhost:8000/api/health
# Database health check
curl http://localhost:8000/api/health/db
# Create token (stub)
curl -X POST http://localhost:8000/api/auth/token \
-H "Content-Type: application/json" \
-d '{
"user_id": "test123",
"username": "testuser",
"discord_id": "123456789"
}'
# List games (stub)
curl http://localhost:8000/api/games/
Unit Testing
Create test files in tests/unit/api/:
import pytest
from httpx import AsyncClient
from app.main import app
@pytest.mark.asyncio
async def test_health_endpoint():
async with AsyncClient(app=app, base_url="http://test") as client:
response = await client.get("/api/health")
assert response.status_code == 200
data = response.json()
assert data["status"] == "healthy"
assert "timestamp" in data
assert "version" in data
@pytest.mark.asyncio
async def test_create_token():
async with AsyncClient(app=app, base_url="http://test") as client:
response = await client.post("/api/auth/token", json={
"user_id": "test123",
"username": "testuser",
"discord_id": "123456789"
})
assert response.status_code == 200
data = response.json()
assert "access_token" in data
assert data["token_type"] == "bearer"
Integration Testing
Test with real database:
import pytest
from httpx import AsyncClient
from app.main import app
from app.database.session import init_db
@pytest.mark.integration
@pytest.mark.asyncio
async def test_database_health_endpoint():
await init_db()
async with AsyncClient(app=app, base_url="http://test") as client:
response = await client.get("/api/health/db")
assert response.status_code == 200
data = response.json()
assert data["status"] == "healthy"
assert data["database"] == "connected"
Troubleshooting
Issue: 422 Unprocessable Entity
Cause: Request body doesn't match Pydantic model schema
Solution:
- Check Swagger UI for expected schema
- Verify all required fields are present
- Check field types match (string vs int, etc.)
- Check for custom validators
Example Error:
{
"detail": [
{
"loc": ["body", "user_id"],
"msg": "field required",
"type": "value_error.missing"
}
]
}
Issue: 401 Unauthorized
Cause: Missing or invalid authentication token
Solution:
- Verify token is included in Authorization header
- Check token format:
Bearer <token> - Verify token hasn't expired
- Check secret_key matches between token creation and verification
Issue: 500 Internal Server Error
Cause: Unhandled exception in route handler
Solution:
- Check backend logs for stack trace
- Add try/except blocks with specific error handling
- Use HTTPException for expected errors
- Log errors with
exc_info=Truefor full stack trace
Example:
try:
result = await some_operation()
except ValidationError as e:
# Expected error - return 400
logger.warning(f"Validation failed: {e}")
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
# Unexpected error - return 500 and log
logger.error(f"Unexpected error: {e}", exc_info=True)
raise HTTPException(status_code=500, detail="Internal server error")
Issue: CORS Errors in Frontend
Cause: Frontend origin not in CORS allowed origins
Solution:
- Check frontend URL in browser console error
- Add origin to
.env:CORS_ORIGINS=["http://localhost:3000", "http://localhost:3001", "http://your-frontend.com"] - Restart backend server
Issue: Route Not Found (404)
Cause: Route not registered or incorrect path
Solution:
- Verify router is imported and registered in
app/main.py - Check prefix matches:
/api/games/vs/api/game/ - Verify route path in decorator matches request
- Check method matches (GET vs POST)
Example Registration:
# In app/main.py
from app.api.routes import games
app.include_router(
games.router,
prefix="/api/games", # All routes will be /api/games/*
tags=["games"]
)
Best Practices
1. Keep Route Handlers Thin
Route handlers should orchestrate, not implement logic:
# ❌ Bad - business logic in route handler
@router.post("/games/")
async def create_game(request: CreateGameRequest):
# Validate teams exist
# Check user permissions
# Create game record
# Initialize state
# Send notifications
# ... 100 lines of logic
# ✅ Good - orchestrate using service layer
@router.post("/games/")
async def create_game(request: CreateGameRequest, user = Depends(get_current_user)):
game_service = GameService()
game = await game_service.create_game(request, user['user_id'])
return game
2. Use Explicit Response Models
Always specify response_model for type safety and documentation:
# ❌ Bad - no response model
@router.get("/games/{game_id}")
async def get_game(game_id: str):
return {"game_id": game_id, "status": "active"}
# ✅ Good - explicit response model
@router.get("/games/{game_id}", response_model=GameResponse)
async def get_game(game_id: str):
return GameResponse(game_id=game_id, status="active")
3. Document Endpoints
Use docstrings for endpoint documentation:
@router.get("/games/", response_model=List[GameListItem])
async def list_games(
league_id: Optional[str] = None,
status: Optional[str] = None
):
"""
List games with optional filters.
Query Parameters:
- league_id: Filter by league ('sba' or 'pd')
- status: Filter by status ('pending', 'active', 'completed')
Returns:
- List of GameListItem objects
Raises:
- 400: Invalid filter parameters
- 500: Database error
"""
...
4. Handle Errors Gracefully
Return meaningful error messages:
@router.get("/games/{game_id}")
async def get_game(game_id: str):
game = await db_ops.get_game(game_id)
if not game:
raise HTTPException(
status_code=404,
detail=f"Game {game_id} not found"
)
return game
5. Use Dependency Injection
Share common logic via dependencies:
from fastapi import Depends
async def get_db_ops():
"""Dependency for database operations"""
return DatabaseOperations()
@router.get("/games/{game_id}")
async def get_game(
game_id: str,
db_ops: DatabaseOperations = Depends(get_db_ops)
):
# db_ops injected automatically
game = await db_ops.get_game(game_id)
return game
6. Version Your API
When making breaking changes, use API versioning:
# Option 1: URL path versioning
app.include_router(games_v1.router, prefix="/api/v1/games")
app.include_router(games_v2.router, prefix="/api/v2/games")
# Option 2: Header versioning (more complex)
@router.get("/games/")
async def list_games(api_version: str = Header(default="v1")):
if api_version == "v2":
return new_format_games()
else:
return legacy_format_games()
References
- FastAPI Documentation: https://fastapi.tiangolo.com/
- Pydantic Documentation: https://docs.pydantic.dev/
- Main Application:
app/main.py(router registration) - Settings:
app/config.py(application configuration) - Database Layer:
app/database/operations.py(DB operations) - WebSocket Layer:
app/websocket/handlers.py(real-time events) - Authentication:
app/utils/auth.py(JWT utilities)
Future Enhancements
Phase 1 (Weeks 1-4)
- Complete Discord OAuth flow in
auth.py - Add authentication middleware
- Create user session endpoints
- Add token refresh mechanism
Phase 2+ (Weeks 5-13)
- Implement game CRUD in
games.py - Add game listing with filters and pagination
- Create team management endpoints
- Add roster/lineup endpoints
- Create statistics/analytics endpoints
- Add game history endpoints
- Implement WebSocket-REST synchronization
Beyond MVP
- Add comprehensive API rate limiting
- Implement API key authentication for external integrations
- Create admin endpoints for moderation
- Add bulk operations endpoints
- Create export endpoints (CSV, JSON)
- Add advanced search and filtering
Note: This directory is currently 33% complete (health endpoints only). Most endpoints are stubs awaiting Phase 2+ implementation. Focus on health endpoints for current operational needs.
Current Phase: Phase 2 - Week 8 Last Updated: 2025-10-31