Architecture & Design

Physician Portal

Location: services/api_gateway/physician_portal/

The Physician Portal is a React 19 single-page application embedded in the API Gateway container. It provides the physician-facing workflow: adaptive anamnesis questionnaire → image capture → AI diagnosis results.

Stack

Dev workflow

npm run dev → Vite dev server :3000
  └── proxy /api/* → FastAPI :8081

Production packaging (multi-stage Docker build)

# Stage 1: Node — builds the React app
FROM node:22-slim AS portal-builder
WORKDIR /portal
COPY services/api_gateway/physician_portal/package*.json ./
RUN npm ci
COPY services/api_gateway/physician_portal/ .
RUN npm run build
 
# Stage 2: Python — copies the dist into the FastAPI container
COPY --from=portal-builder /portal/dist ./physician_portal/dist

FastAPI detects physician_portal/dist/ at startup and:

1. Mounts dist/assets at /assets (static files)
2. Returns index.html for all unmatched routes (SPA fallback)

Authentication

The portal uses JWT Bearer tokens. login stores the token in localStorage; apiFetch in src/api/client.ts attaches it as Authorization: Bearer <token> on every request.

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