Large Application Example

myapp/
├── app.py
├── config.py
├── models/
│   └── user.py
├── schemas/
│   └── user.py
├── services/
│   └── user_service.py
├── repositories/
│   └── user_repository.py
├── routes/
│   ├── __init__.py
│   ├── users.py
│   └── auth.py
├── middleware/
│   └── auth.py
└── utils/
    └── db.py

# config.py
import os
from typing import Any

class Config:
    DEBUG = os.environ.get("DEBUG", "false").lower() == "true"
    SECRET_KEY = os.environ.get("SECRET_KEY", "change-me")
    DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite:///app.db")
    ALLOWED_ORIGINS = os.environ.get("ALLOWED_ORIGINS", "http://localhost:3000").split(",")
    JWT_EXPIRATION = int(os.environ.get("JWT_EXPIRATION", 3600))
    MAX_BODY_SIZE = int(os.environ.get("MAX_BODY_SIZE", 10 * 1024 * 1024))

    @classmethod
    def to_dict(cls) -> dict[str, Any]:
        return {
            "DEBUG": cls.DEBUG,
            "SECRET_KEY": cls.SECRET_KEY,
            "DATABASE_URL": cls.DATABASE_URL,
            "ALLOWED_ORIGINS": cls.ALLOWED_ORIGINS,
            "JWT_EXPIRATION": cls.JWT_EXPIRATION,
            "MAX_BODY_SIZE": cls.MAX_BODY_SIZE,
        }

# models/user.py
from dataclasses import dataclass
from datetime import datetime
from typing import Optional

@dataclass
class User:
    id: Optional[int] = None
    name: str = ""
    email: str = ""
    hashed_password: str = ""
    created_at: Optional[datetime] = None
    updated_at: Optional[datetime] = None

    def to_dict(self) -> dict:
        return {
            "id": self.id,
            "name": self.name,
            "email": self.email,
            "created_at": self.created_at.isoformat() if self.created_at else None,
            "updated_at": self.updated_at.isoformat() if self.updated_at else None,
        }

    @classmethod
    def from_dict(cls, data: dict) -> "User":
        return cls(
            id=data.get("id"),
            name=data.get("name", ""),
            email=data.get("email", ""),
            hashed_password=data.get("hashed_password", ""),
            created_at=data.get("created_at"),
            updated_at=data.get("updated_at"),
        )

# schemas/user.py
from flaxon.validation import Schema, fields

class CreateUser(Schema):
    name = fields.String(required=True, min_length=2, max_length=80)
    email = fields.Email(required=True)
    password = fields.String(required=True, min_length=8)

class UpdateUser(Schema):
    name = fields.String(required=False, min_length=2, max_length=80)
    email = fields.Email(required=False)

class LoginRequest(Schema):
    email = fields.Email(required=True)
    password = fields.String(required=True, min_length=8)

# repositories/user_repository.py
from typing import Optional, List
from models.user import User

class UserRepository:
    def __init__(self, db):
        self.db = db

    async def get_by_id(self, user_id: int) -> Optional[User]:
        row = await self.db.fetch_one("SELECT * FROM users WHERE id = $1", user_id)
        return User.from_dict(row) if row else None

    async def get_by_email(self, email: str) -> Optional[User]:
        row = await self.db.fetch_one("SELECT * FROM users WHERE email = $1", email)
        return User.from_dict(row) if row else None

    async def create(self, user: User) -> User:
        row = await self.db.fetch_one(
            "INSERT INTO users (name, email, hashed_password) VALUES ($1, $2, $3) RETURNING *",
            user.name, user.email, user.hashed_password
        )
        return User.from_dict(row)

    async def update(self, user: User) -> User:
        row = await self.db.fetch_one(
            "UPDATE users SET name = $1, email = $2, updated_at = NOW() WHERE id = $3 RETURNING *",
            user.name, user.email, user.id
        )
        return User.from_dict(row)

    async def delete(self, user_id: int) -> bool:
        await self.db.execute("DELETE FROM users WHERE id = $1", user_id)
        return True

    async def list(self, limit: int = 100, offset: int = 0) -> List[User]:
        rows = await self.db.fetch_all("SELECT * FROM users LIMIT $1 OFFSET $2", limit, offset)
        return [User.from_dict(row) for row in rows]

    async def count(self) -> int:
        return await self.db.fetch_val("SELECT COUNT(*) FROM users")

# services/user_service.py
from typing import Optional, List
from flaxon.security import hash_password, verify_password
from models.user import User
from repositories.user_repository import UserRepository

class UserService:
    def __init__(self, repo: UserRepository):
        self.repo = repo

    async def get_user(self, user_id: int) -> Optional[User]:
        return await self.repo.get_by_id(user_id)

    async def get_user_by_email(self, email: str) -> Optional[User]:
        return await self.repo.get_by_email(email)

    async def create_user(self, name: str, email: str, password: str) -> User:
        if await self.repo.get_by_email(email):
            raise ValueError("Email already registered")

        user = User(
            name=name,
            email=email,
            hashed_password=hash_password(password),
        )
        return await self.repo.create(user)

    async def authenticate(self, email: str, password: str) -> Optional[User]:
        user = await self.repo.get_by_email(email)
        if not user:
            return None
        if not verify_password(password, user.hashed_password):
            return None
        return user

    async def update_user(self, user_id: int, data: dict) -> Optional[User]:
        user = await self.repo.get_by_id(user_id)
        if not user:
            return None
        if "name" in data:
            user.name = data["name"]
        if "email" in data:
            user.email = data["email"]
        return await self.repo.update(user)

    async def delete_user(self, user_id: int) -> bool:
        return await self.repo.delete(user_id)

    async def list_users(self, page: int = 1, per_page: int = 20) -> dict:
        offset = (page - 1) * per_page
        users = await self.repo.list(per_page, offset)
        total = await self.repo.count()

        return {
            "items": [u.to_dict() for u in users],
            "page": page,
            "per_page": per_page,
            "total": total,
            "total_pages": (total + per_page - 1) // per_page,
        }

# routes/auth.py
from flaxon import Router, HTTPException, Response
from flaxon.security import JWTBackend, login_required

from schemas.user import CreateUser, LoginRequest
from services.user_service import UserService
from repositories.user_repository import UserRepository
from utils.db import get_db
from config import Config

router = Router(prefix="/api/v1/auth")
jwt_backend = JWTBackend(secret_key=Config.SECRET_KEY)

@router.post("/register")
async def register(data: CreateUser):
    db = get_db()
    repo = UserRepository(db)
    service = UserService(repo)

    try:
        user = await service.create_user(data.name, data.email, data.password)
        return {"success": True, "user": user.to_dict()}
    except ValueError as exc:
        raise HTTPException(400, str(exc))

@router.post("/login")
async def login(data: LoginRequest):
    db = get_db()
    repo = UserRepository(db)
    service = UserService(repo)

    user = await service.authenticate(data.email, data.password)
    if not user:
        raise HTTPException(401, "Invalid credentials")

    token = await jwt_backend.create_token(user.to_dict())
    return {"token": token, "user": user.to_dict()}

@router.post("/refresh")
@login_required
async def refresh(request):
    user = getattr(request, "user")
    token = await jwt_backend.create_token(user)
    return {"token": token}

@router.post("/logout")
@login_required
async def logout(request):
    # In production, blacklist the token
    return {"success": True}

# routes/users.py
from flaxon import Router, HTTPException
from flaxon.security import login_required, permission_required

from schemas.user import UpdateUser
from services.user_service import UserService
from repositories.user_repository import UserRepository
from utils.db import get_db

router = Router(prefix="/api/v1/users")

@router.get("/")
@login_required
async def list_users(request):
    page = request.query.get_int("page", 1)
    per_page = request.query.get_int("per_page", 20)

    db = get_db()
    repo = UserRepository(db)
    service = UserService(repo)

    return await service.list_users(page, per_page)

@router.get("/")
@login_required
async def get_user(user_id: int):
    db = get_db()
    repo = UserRepository(db)
    service = UserService(repo)

    user = await service.get_user(user_id)
    if not user:
        raise HTTPException(404, "User not found")

    return user.to_dict()

@router.put("/")
@login_required
async def update_user(user_id: int, data: UpdateUser):
    db = get_db()
    repo = UserRepository(db)
    service = UserService(repo)

    user = await service.update_user(user_id, data.to_dict())
    if not user:
        raise HTTPException(404, "User not found")

    return user.to_dict()

@router.delete("/")
@login_required
@permission_required("delete_user")
async def delete_user(user_id: int):
    db = get_db()
    repo = UserRepository(db)
    service = UserService(repo)

    result = await service.delete_user(user_id)
    if not result:
        raise HTTPException(404, "User not found")

    return {"deleted": True}

@router.get("/me")
@login_required
async def get_me(request):
    user = getattr(request, "user")
    return user

# middleware/auth.py
from flaxon.middleware import Middleware

class AuthMiddleware(Middleware):
    async def __call__(self, scope, receive, send):
        if scope.get("type") != "http":
            await self.app(scope, receive, send)
            return

        # Add user to scope for WebSocket authentication
        if "user" not in scope:
            scope["user"] = None

        await self.app(scope, receive, send)

# middleware/logging.py
class LoggingMiddleware(Middleware):
    async def __call__(self, scope, receive, send):
        if scope.get("type") != "http":
            await self.app(scope, receive, send)
            return

        import time
        start = time.perf_counter()
        path = scope.get("path", "/")
        method = scope.get("method", "GET")

        await self.app(scope, receive, send)

        duration = (time.perf_counter() - start) * 1000
        print(f"{method} {path} - {duration:.2f}ms")

# utils/db.py
import asyncpg
from config import Config

_db_pool = None

async def init_db():
    global _db_pool
    _db_pool = await asyncpg.create_pool(Config.DATABASE_URL)

async def close_db():
    if _db_pool:
        await _db_pool.close()

def get_db():
    return _db_pool

# utils/health.py
from flaxon.health import HealthCheck

class DatabaseHealthCheck(HealthCheck):
    def __init__(self):
        super().__init__("database", self._check)

    async def _check(self):
        try:
            db = get_db()
            await db.fetch("SELECT 1")
            return {"status": "healthy"}
        except Exception as exc:
            return {"status": "unhealthy", "message": str(exc)}

# app.py
from flaxon import Flaxon
from flaxon.middleware import RequestIDMiddleware, SecurityHeadersMiddleware, CORSMiddleware
from flaxon.security import AuthenticationMiddleware

from config import Config
from routes import users, auth
from middleware.auth import AuthMiddleware
from middleware.logging import LoggingMiddleware
from utils.db import init_db, close_db
from utils.health import DatabaseHealthCheck
from flaxon.health import HealthRegistry

app = Flaxon("myapp", debug=Config.DEBUG)
app.config.update(Config.to_dict())

# Middleware (order matters)
app.add_middleware(LoggingMiddleware)
app.add_middleware(RequestIDMiddleware)
app.add_middleware(SecurityHeadersMiddleware)
app.add_middleware(
    CORSMiddleware,
    allowed_origins=Config.ALLOWED_ORIGINS,
    allow_credentials=True,
)
app.add_middleware(AuthMiddleware)

# Authentication
from flaxon.security import JWTBackend
jwt_backend = JWTBackend(secret_key=Config.SECRET_KEY)
app.add_middleware(AuthenticationMiddleware, backend=jwt_backend)

# Routes
app.include_router(auth.router)
app.include_router(users.router)

# Health Checks
health = HealthRegistry()
health.register(DatabaseHealthCheck())

@app.get("/health")
async def health_check():
    return await health.check_all()

@app.get("/health/live")
async def liveness():
    return {"status": "alive"}

@app.get("/health/ready")
async def readiness():
    result = await health.check_all()
    all_healthy = all(r["status"] == "healthy" for r in result.values())
    status = 200 if all_healthy else 503
    return result, status

# Lifecycle
@app.on_startup
async def startup():
    await init_db()
    print("✅ Database connected")

@app.on_shutdown
async def shutdown():
    await close_db()
    print("✅ Database disconnected")

@app.get("/")
async def home():
    return {"message": "Welcome to MyApp API", "version": "1.0.0"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)