Python FastAPI

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: initialize DB pool, load ML model, etc.
    app.state.db = await create_pool()
    yield
    # Shutdown: close connections, flush buffers
    await app.state.db.close()

app = FastAPI(
    title="domus-api",
    version="0.1.0",
    lifespan=lifespan,
)

from fastapi import FastAPI, status
from pydantic import BaseModel

class Device(BaseModel):
    hostname: str
    ip: str
    vlan: int

class DeviceOut(BaseModel):
    id: int
    hostname: str
    ip: str

@app.get("/devices", response_model=list[DeviceOut])
async def list_devices():
    return await db.fetch_all("SELECT * FROM devices")

@app.post("/devices", response_model=DeviceOut, status_code=status.HTTP_201_CREATED)
async def create_device(device: Device):
    return await db.insert(device.model_dump())

@app.put("/devices/{device_id}", response_model=DeviceOut)
async def update_device(device_id: int, device: Device):
    return await db.update(device_id, device.model_dump())

@app.delete("/devices/{device_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_device(device_id: int):
    await db.delete(device_id)

from fastapi import Query

@app.get("/devices/{device_id}")
async def get_device(device_id: int):
    """Path param -- FastAPI validates type automatically."""
    ...

@app.get("/devices")
async def search_devices(
    vlan: int | None = None,
    hostname: str | None = Query(None, min_length=3, max_length=64),
    skip: int = Query(0, ge=0),
    limit: int = Query(20, ge=1, le=100),
):
    """Query params -- all optional with validation constraints."""
    ...

from pydantic import BaseModel, Field

class PolicyUpdate(BaseModel):
    name: str = Field(..., min_length=1, max_length=128, description="Policy set name")
    vlan: int = Field(..., ge=1, le=4094, description="Target VLAN ID")
    enabled: bool = Field(True, description="Whether policy is active")

@app.put("/policies/{policy_id}")
async def update_policy(policy_id: int, body: PolicyUpdate):
    # body is already validated -- access fields directly
    return {"id": policy_id, **body.model_dump()}

from fastapi import Depends, Header, HTTPException

async def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

async def verify_api_key(x_api_key: str = Header(...)):
    if x_api_key != settings.api_key:
        raise HTTPException(status_code=403, detail="Invalid API key")

@app.get("/secure-data", dependencies=[Depends(verify_api_key)])
async def secure_endpoint(db: Session = Depends(get_db)):
    return db.query(Data).all()

from fastapi import HTTPException, Request
from fastapi.responses import JSONResponse

@app.get("/devices/{device_id}")
async def get_device(device_id: int):
    device = await db.get(device_id)
    if not device:
        raise HTTPException(
            status_code=404,
            detail=f"Device {device_id} not found",
        )
    return device

class DeviceNotFoundError(Exception):
    def __init__(self, device_id: int):
        self.device_id = device_id

@app.exception_handler(DeviceNotFoundError)
async def device_not_found_handler(request: Request, exc: DeviceNotFoundError):
    return JSONResponse(
        status_code=404,
        content={"detail": f"Device {exc.device_id} not found"},
    )

import time
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://docs.domusdigitalis.dev"],
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.middleware("http")
async def add_timing_header(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    elapsed = time.perf_counter() - start
    response.headers["X-Process-Time"] = f"{elapsed:.4f}"
    return response

from fastapi.testclient import TestClient

client = TestClient(app)

def test_list_devices():
    response = client.get("/devices")
    assert response.status_code == 200
    assert isinstance(response.json(), list)

def test_create_device():
    payload = {"hostname": "switch-01", "ip": "10.50.1.100", "vlan": 10}
    response = client.post("/devices", json=payload)
    assert response.status_code == 201
    data = response.json()
    assert data["hostname"] == "switch-01"

def test_not_found():
    response = client.get("/devices/99999")
    assert response.status_code == 404
    assert "not found" in response.json()["detail"].lower()

from fastapi import APIRouter

router = APIRouter(prefix="/devices", tags=["devices"])

@router.get("/")
async def list_devices():
    ...

@router.get("/{device_id}")
async def get_device(device_id: int):
    ...

# In main.py
app.include_router(router)