Backend Architecture Guide

Overview

The Noumaris backend is built with FastAPI (Python 3.11+) and follows a modular, scalable architecture designed for healthcare applications requiring HIPAA compliance, auditability, and role-based access control.

Directory Structure

backend/
├── src/noumaris_backend/
│   ├── api/                      # API layer (FastAPI routers)
│   │   ├── main.py              # Main application entry point
│   │   ├── auth.py              # JWT authentication (Keycloak)
│   │   ├── admin_auth.py        # Admin role decorators & helpers
│   │   ├── superadmin.py        # Superadmin API router
│   │   ├── institution_admin.py # Institution admin API router
│   │   ├── invitations.py       # Invitation system router
│   │   ├── permissions.py       # Permission management router
│   │   ├── websocket_auth.py    # WebSocket authentication
│   │   └── websocket_monitoring.py  # WebSocket metrics
│   ├── models/
│   │   └── db_models.py         # SQLAlchemy ORM models
│   └── services/
│       └── permission_service.py # Permission business logic
├── tests/
│   ├── test_admin_models.py     # Phase 1 model tests
│   └── test_admin_apis.py       # Phase 2 API tests
├── alembic/                      # Database migrations
│   ├── versions/                # Migration files
│   └── env.py                   # Alembic configuration
├── scripts/
│   ├── seed_db.py               # Seed templates & tags
│   ├── seed_admin_data.py       # Seed admin test data
│   └── seed_features.py         # Seed feature permissions
├── keycloak/
│   └── realm-export.json        # Keycloak realm configuration
├── Dockerfile                    # Production container
├── API_DOCUMENTATION.md          # Comprehensive API docs
├── BACKEND_ARCHITECTURE.md       # This file
└── requirements.txt              # Python dependencies

Architecture Patterns

1. Layered Architecture

┌─────────────────────────────────────┐
│     API Layer (FastAPI Routers)    │  ← HTTP endpoints, WebSocket, validation
├─────────────────────────────────────┤
│   Service Layer (Business Logic)   │  ← Permission checks, audit logging
├─────────────────────────────────────┤
│    Data Layer (SQLAlchemy ORM)     │  ← Database models, relationships
├─────────────────────────────────────┤
│  Infrastructure (PostgreSQL, etc)  │  ← Database, Keycloak, Deepgram, Claude
└─────────────────────────────────────┘

2. Dependency Injection

FastAPI's Depends() is used extensively for:

Authentication: Depends(get_current_user)
Authorization: Depends(require_superadmin), Depends(require_institution_admin)
Database Sessions: Context managers (db_manager.create_session())

3. Role-Based Access Control (RBAC)

Implemented via decorators in admin_auth.py:

python

@router.get("/admin/superadmin/institutions")
async def list_institutions(current_user: User = Depends(require_superadmin)):
    # Only superadmins can access this endpoint
    pass

Role Hierarchy:

Superadmin → System-wide access
Senior Institution Admin → Can create admins + manage residents
Institution Admin → Manage residents only
Resident → Use granted features
Physician → Individual access

4. Database Session Management

All database operations use context managers for automatic cleanup:

python

with db_manager.create_session() as session:
    # Perform queries
    session.add(new_record)
    session.commit()
    # Session automatically closed

Benefits:

Prevents connection leaks
Automatic rollback on exceptions
Thread-safe sessions

Key Design Decisions

Why FastAPI?

Automatic API Documentation: OpenAPI/Swagger built-in
Pydantic Validation: Strong typing + automatic validation
Async Support: High-performance WebSocket transcription
Dependency Injection: Clean separation of concerns
Modern Python: Type hints, async/await, Python 3.11+

Why Keycloak?

Centralized Authentication: Single sign-on (SSO) support
Role Management: Built-in RBAC with realm roles
Token Standards: JWT with RS256 signatures
Extensibility: Can add MFA, social login, etc.
HIPAA Compliance: Audit logs, session management

Why SQLAlchemy ORM?

Type Safety: Python models map to database tables
Relationship Management: Foreign keys, cascades handled automatically
Migration Support: Alembic for schema changes
Query Builder: Protection against SQL injection
Database Agnostic: Works with PostgreSQL, MySQL, SQLite

Why Pydantic Models?

Request Validation: Automatic validation of API inputs
Response Serialization: Type-safe JSON responses
Documentation: Automatic OpenAPI schema generation
Type Hints: Better IDE support and catch errors early
Custom Validators: Field-level validation (e.g., email format, PGY range)

Security Best Practices

1. Authentication & Authorization

✅ JWT Validation: Every endpoint validates JWT signature against Keycloak public key

python

public_key = get_keycloak_public_key()  # Cached
payload = jwt.decode(token, public_key, algorithms=["RS256"])

✅ Role Checks: Decorator-based role enforcement

python

@require_superadmin  # 403 if not superadmin

✅ Institution Ownership: Admins can only access their own institution

python

def validate_resident_access(admin: InstitutionAdmin, resident_user_id: str):
    if resident.institution_id != admin.institution_id:
        raise HTTPException(403, "Cannot access other institutions")

2. Input Validation

✅ Pydantic Models: Automatic validation before handler execution

python

class InviteResidentRequest(BaseModel):
    email: EmailStr  # Validates email format
    pgy_level: int

    @field_validator('pgy_level')
    def validate_pgy(cls, v):
        if not 1 <= v <= 10:
            raise ValueError('PGY must be 1-10')
        return v

✅ SQL Injection Protection: SQLAlchemy parameterized queries

python

# Safe - parameterized
session.query(User).filter(User.email == user_email).first()

# Unsafe - NEVER DO THIS
session.execute(f"SELECT * FROM users WHERE email = '{user_email}'")

3. Rate Limiting

✅ SlowAPI Integration: Per-IP rate limiting

python

@limiter.limit("50/minute")
async def endpoint(request: Request):
    pass

✅ WebSocket Limits: 3 concurrent connections per user

python

if not ws_rate_limiter.check_limit(user.id):
    raise HTTPException(429, "Too many connections")

4. Audit Logging

✅ All Admin Actions Logged:

python

log_admin_action(
    action="GRANT_PERMISSION",
    user_id=current_user.id,
    username=current_user.username,
    details={'feature_id': 'live_transcription'},
    session=session
)

Logged Events:

Institution creation/modification
Feature grants/revocations
Resident invitations
Permission changes
Subscription status updates

5. Error Handling

✅ Detailed Error Messages: Clear explanations without exposing internals

python

raise HTTPException(
    status_code=409,
    detail="Institution with name 'Test Hospital' already exists"
)

✅ Exception Hierarchy:

400 Bad Request → Invalid input
401 Unauthorized → Missing/invalid token
403 Forbidden → Insufficient permissions
404 Not Found → Resource doesn't exist
409 Conflict → Duplicate record
422 Validation Error → Pydantic validation failed

Database Design Principles

1. Foreign Key Relationships

All relationships use proper foreign keys with cascade behavior:

python

class ResidentProfile(Base):
    institution_id = Column(Integer, ForeignKey('institutions.id', ondelete='CASCADE'))
    # If institution is deleted, all residents are also deleted

2. Soft Deletes

Important records use status fields instead of hard deletes:

python

class ResidentProfile(Base):
    status = Column(String, default='active')  # active/inactive/graduated

3. Timestamps

All models include audit timestamps:

python

created_at = Column(DateTime, server_default=func.now())
updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())

4. Unique Constraints

Prevent duplicates at the database level:

python

class Institution(Base):
    name = Column(String, unique=True)  # No duplicate institution names

5. Indexes

Optimize common queries:

python

__table_args__ = (
    Index('ix_institution_email_status', 'institution_id', 'email', 'status'),
)

Performance Optimization

1. Connection Pooling

SQLAlchemy connection pool configuration:

python

engine = create_engine(
    DATABASE_URL,
    pool_size=20,          # Max connections in pool
    max_overflow=40,       # Allow 40 additional connections
    pool_pre_ping=True,    # Test connections before use
    pool_recycle=3600      # Recycle connections after 1 hour
)

2. Query Optimization

✅ Use .first() instead of .all()[0]:

python

# Good - stops after first match
user = session.query(User).filter(User.email == email).first()

# Bad - loads all matching rows
user = session.query(User).filter(User.email == email).all()[0]

✅ Eager Loading: Load relationships upfront to avoid N+1 queries

python

# Load templates with tags in single query
templates = session.query(NoteTemplate).options(contains_eager(NoteTemplate.tags)).all()

✅ Pagination: Always paginate large result sets

python

query.offset((page - 1) * page_size).limit(page_size).all()

3. Caching

✅ Keycloak Public Key: Cached with @lru_cache

python

@lru_cache(maxsize=1)
def get_keycloak_public_key():
    # Fetched once, then cached
    pass

✅ Feature List: Consider caching active features (low change frequency)

Testing Strategy

1. Unit Tests

Test individual functions in isolation:

python

def test_check_usage_limits():
    usage = check_usage_limits(institution_id, test_db)
    assert usage['residents_used'] >= 0

Location: tests/test_admin_models.py, tests/test_admin_apis.py

2. Integration Tests

Test complete workflows:

python

def test_complete_resident_onboarding():
    # 1. Admin invites resident
    # 2. Resident validates token
    # 3. Resident registers in Keycloak
    # 4. Resident accepts invitation
    # 5. Admin grants permissions
    pass

3. API Tests

Test HTTP endpoints with FastAPI TestClient:

python

response = client.post(
    "/admin/institution/residents/invite",
    json=invite_data,
    headers={"Authorization": f"Bearer {token}"}
)
assert response.status_code == 201

4. Database Tests

Test with in-memory SQLite for speed:

python

TEST_DATABASE_URL = "sqlite:///:memory:"
Base.metadata.create_all(bind=engine)

Run Tests:

bash

cd backend
pytest tests/ -v --tb=short

Deployment

Development

bash

cd backend
poetry install
poetry shell
python -m uvicorn src.noumaris_backend.api.main:app --reload

Ports:

Backend: 8000
PostgreSQL: 5432
Keycloak: 8080

Production (Google Cloud Run)

bash

# Build Docker image
docker build -t gcr.io/noumaris/api:latest .

# Push to Artifact Registry
docker push gcr.io/noumaris/api:latest

# Deploy to Cloud Run
gcloud run deploy noumaris-api \
  --image gcr.io/noumaris/api:latest \
  --region us-central1 \
  --allow-unauthenticated

Environment Variables (Cloud Run):

DATABASE_URL: PostgreSQL connection string
ANTHROPIC_API_KEY: Claude API key
DEEPGRAM_API_KEY: Transcription API key
KEYCLOAK_URL: Keycloak server URL
KEYCLOAK_REALM: Realm name
KEYCLOAK_CLIENT_ID: Client ID

Monitoring & Observability

1. Health Checks

Liveness: /health (fast, no dependencies)
Readiness: /health/ready (checks database)

2. Logging

Structured logging with levels:

python

logger.info(f"Admin {username} created institution {institution_id}")
logger.warning(f"Rate limit exceeded for user {user_id}")
logger.error(f"Database connection failed: {error}", exc_info=True)

Log Levels:

INFO: Normal operations
WARNING: Potential issues
ERROR: Failures requiring attention
CRITICAL: Service-impacting errors

3. Metrics (Future Enhancement)

Consider adding Prometheus metrics:

Request latency (P50, P95, P99)
Error rates by endpoint
Active WebSocket connections
Database connection pool usage

Migration Guide

Adding a New Endpoint

Create Pydantic models (request/response)
Add endpoint to router
Add authentication dependency
Implement business logic
Add tests
Update API documentation

Example:

python

# 1. Pydantic models
class NewFeatureRequest(BaseModel):
    name: str
    description: str

# 2. Add to router
@router.post("/admin/features")
async def create_feature(
    request: NewFeatureRequest,
    current_user: User = Depends(require_superadmin)  # 3. Auth
):
    # 4. Business logic
    with db_manager.create_session() as session:
        feature = Feature(name=request.name, description=request.description)
        session.add(feature)
        session.commit()
        return {"id": feature.id}

Database Schema Changes

Modify models in db_models.py

Generate migration:

bash

alembic revision --autogenerate -m "description"

Review migration file in alembic/versions/
Apply migration:
bash
```
alembic upgrade head
```
1
Update seed scripts if needed
Add tests for new fields

Common Pitfalls & Solutions

Problem: 401 Unauthorized on all endpoints

Cause: JWT token expired or Keycloak public key not cached

Solution:

python

# Clear cache and refetch key
from noumaris_backend.api.auth import get_keycloak_public_key
get_keycloak_public_key.cache_clear()

Problem: Foreign key constraint violations

Cause: Trying to delete a record referenced by another table

Solution: Use cascade deletes or soft deletes

python

ForeignKey('institutions.id', ondelete='CASCADE')

Problem: N+1 query problem

Cause: Loading relationships in a loop

Solution: Use eager loading

python

.options(joinedload(Template.tags))

Problem: Database connection pool exhausted

Cause: Not closing sessions

Solution: Always use context managers

python

with db_manager.create_session() as session:
    # Session auto-closes
    pass

Future Enhancements

Short-term (Phase 3)

[ ] Email notification system (SendGrid/AWS SES)
[ ] Audit log database table (currently app logs only)
[ ] Enhanced rate limiting (per-user, per-endpoint)
[ ] WebSocket reconnection handling
[ ] Batch invitation CSV upload

Medium-term

[ ] Prometheus metrics export
[ ] GraphQL API alongside REST
[ ] Real-time analytics dashboard
[ ] Multi-tenancy isolation improvements
[ ] Advanced reporting (usage, billing)

Long-term

[ ] Machine learning for note quality scoring
[ ] Automated compliance checks (HIPAA, GDPR)
[ ] Blockchain audit trail
[ ] Multi-language support
[ ] Voice biometrics for authentication

Contributing

Code Style

PEP 8: Follow Python style guide
Type Hints: Use everywhere possible
Docstrings: All public functions
Comments: Explain "why", not "what"

Pull Request Process

Create feature branch: feature/your-feature-name
Write tests (aim for 80%+ coverage)
Update documentation
Run tests: pytest tests/ -v
Format code: black .
Submit PR with description

Support

Documentation:

API Docs: /docs (Swagger UI)
Architecture: This file
Testing: tests/README.md

Contact:

Technical Issues: [email protected]
Feature Requests: GitHub Issues
Security: [email protected]

Backend Architecture Guide ​

Overview ​

Directory Structure ​

Architecture Patterns ​

1. Layered Architecture ​

2. Dependency Injection ​

3. Role-Based Access Control (RBAC) ​

4. Database Session Management ​

Key Design Decisions ​

Why FastAPI? ​

Why Keycloak? ​

Why SQLAlchemy ORM? ​

Why Pydantic Models? ​

Security Best Practices ​

1. Authentication & Authorization ​

2. Input Validation ​

3. Rate Limiting ​

4. Audit Logging ​

5. Error Handling ​

Database Design Principles ​

1. Foreign Key Relationships ​

2. Soft Deletes ​

3. Timestamps ​

4. Unique Constraints ​

5. Indexes ​

Performance Optimization ​

1. Connection Pooling ​

2. Query Optimization ​

3. Caching ​

Testing Strategy ​

1. Unit Tests ​

2. Integration Tests ​

3. API Tests ​

4. Database Tests ​

Deployment ​

Development ​

Production (Google Cloud Run) ​

Monitoring & Observability ​

1. Health Checks ​

2. Logging ​

3. Metrics (Future Enhancement) ​

Migration Guide ​

Adding a New Endpoint ​

Database Schema Changes ​

Common Pitfalls & Solutions ​

Problem: 401 Unauthorized on all endpoints ​

Problem: Foreign key constraint violations ​

Problem: N+1 query problem ​

Problem: Database connection pool exhausted ​

Future Enhancements ​

Short-term (Phase 3) ​

Medium-term ​

Long-term ​

Contributing ​

Code Style ​

Pull Request Process ​

Support ​

Backend Architecture Guide

Overview

Directory Structure

Architecture Patterns

1. Layered Architecture

2. Dependency Injection

3. Role-Based Access Control (RBAC)

4. Database Session Management

Key Design Decisions

Why FastAPI?

Why Keycloak?

Why SQLAlchemy ORM?

Why Pydantic Models?

Security Best Practices

1. Authentication & Authorization

2. Input Validation

3. Rate Limiting

4. Audit Logging

5. Error Handling

Database Design Principles

1. Foreign Key Relationships

2. Soft Deletes

3. Timestamps

4. Unique Constraints

5. Indexes

Performance Optimization

1. Connection Pooling

2. Query Optimization

3. Caching

Testing Strategy

1. Unit Tests

2. Integration Tests

3. API Tests

4. Database Tests

Deployment

Development

Production (Google Cloud Run)

Monitoring & Observability

1. Health Checks

2. Logging

3. Metrics (Future Enhancement)

Migration Guide

Adding a New Endpoint

Database Schema Changes

Common Pitfalls & Solutions

Problem: 401 Unauthorized on all endpoints

Problem: Foreign key constraint violations

Problem: N+1 query problem

Problem: Database connection pool exhausted

Future Enhancements

Short-term (Phase 3)

Medium-term

Long-term

Contributing

Code Style

Pull Request Process

Support