Architecture & Design

AI Service Client

Location: src/core/ai_client.py:31

The AIClient class provides async HTTP communication with the AI Service:

Features

Async HTTP Client

Uses httpx.AsyncClient for non-blocking I/O
Shared client instance across requests (singleton pattern)
Configurable timeout (default 30s)

Automatic Error Handling

Converts HTTP errors to domain-specific exceptions:
- AIServiceValidationError (400 errors)
- AIServiceUnavailableError (502/503/504 errors)
- AIServiceError (other errors)
Proper error logging for debugging

Graceful Shutdown

Properly closes HTTP connections on app shutdown
Managed via FastAPI lifespan events

Example Usage


from src.core.ai_client import get_ai_client
 
@router.post("/predict_diagnosis")
async def predict_diagnosis(
    anamnesis: AnamnesisRequest,
    ai_client: Annotated[AIClient, Depends(get_ai_client)],
):
    try:
        response = await ai_client.predict_diagnosis(
            anamnesis.model_dump(by_alias=True)
        )
        return DiagnosisPrediction(**response)
    except AIServiceUnavailableError:
        raise HTTPException(status_code=502, detail="AI service unavailable")

Database Layer

Async Sessions

All database operations use the async with get_db() pattern:


from src.database import get_db
 
@router.get("/cases/{case_id}")
async def get_case(
    case_id: str,
    db: Annotated[AsyncSession, Depends(get_db)],
):
    case = await db.get(RnRequest, case_id)
    return case

Connection Pooling

SQLAlchemy manages a connection pool automatically:

Default pool size: 5 connections
Max overflow: 10 additional connections
Automatic connection recycling

Migration Management

Alembic tracks schema versions and handles migrations:


# Create new migration
uv run alembic revision --autogenerate -m "add user table"
 
# Apply migrations
uv run alembic upgrade head
 
# Rollback
uv run alembic downgrade -1

Test Isolation

Tests use testcontainers for real PostgreSQL instances:

Each test gets a fresh database
Migrations applied automatically
Isolated from development database

Schema Design

Location: src/schemas/anamnesis.py

CamelCase Conversion

The API uses camelCase (frontend-friendly) while Python code uses snake_case:


class AnamnesisRequest(BaseModel):
    user_id: str = Field(alias="userId")
    body_location: str = Field(alias="bodyLocation")
 
    class Config:
        populate_by_name = True  # Allow both forms

Field Aliases

Pydantic alias enables automatic conversion:

Request: Client sends {"userId": "123"}
Internal: Python code accesses request.user_id
Response: Client receives {"userId": "123"}

Type Safety

Full type hints and validation:

Enum types for controlled vocabularies
Nested models for complex objects
Optional fields with defaults
Custom validators for business logic

Example Schema


class Symptom(BaseModel):
    type: SymptomType
    severity: int = Field(ge=1, le=10)
    duration_days: int = Field(alias="durationDays")
 
class AnamnesisRequest(BaseModel):
    symptoms: list[Symptom]
    age: int = Field(ge=0, le=120)
    gender: Gender

Dependency Injection

FastAPI’s dependency injection system provides clean separation:

Database Session


async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with async_session_maker() as session:
        yield session

AI Client


_ai_client: AIClient | None = None
 
async def get_ai_client() -> AIClient:
    global _ai_client
    if _ai_client is None:
        _ai_client = AIClient()
    return _ai_client

Usage in Routes


@router.post("/endpoint")
async def endpoint(
    db: Annotated[AsyncSession, Depends(get_db)],
    ai_client: Annotated[AIClient, Depends(get_ai_client)],
):
    # Use db and ai_client
    pass

Lifespan Management

Location: src/main.py:19

The FastAPI app uses lifespan context manager for startup/shutdown:


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    logger.info("Starting DermaDetect API Gateway")
    logger.info(f"Database URL: {settings.database_url}")
    logger.info(f"AI Service URL: {settings.ai_service_url}")
    yield
    # Shutdown
    logger.info("Shutting down DermaDetect API Gateway")
    await close_ai_client()

Benefits:

Clean resource initialization
Graceful connection cleanup
Startup validation logging