ডিপেন্ডেন্সি, লাইফস্প্যান, মিডলওয়্যার ও সেটিংস

mkdir -p ~/projects/prompt-api-ops
cd ~/projects/prompt-api-ops
uv init --bare -q
uv add -q "fastapi[standard]" pydantic-settings httpx2
# with pip: python -m venv .venv && source .venv/bin/activate && pip install -q "fastapi[standard]" pydantic-settings httpx2
mkdir -p app
touch app/__init__.py
cat >> pyproject.toml <<'EOF'

[tool.fastapi]
entrypoint = "app.main:app"
EOF

# file: app/errors.py
"""Compact version of the previous page's errors: one JSON shape for every error."""
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse


class AppError(Exception):
    status_code = 400
    code = "bad_request"

    def __init__(self, message: str) -> None:
        super().__init__(message)
        self.message = message


class NotFound(AppError):
    status_code = 404
    code = "not_found"


class Conflict(AppError):
    status_code = 409
    code = "conflict"


class Unauthorized(AppError):
    status_code = 401
    code = "unauthorized"


def install_error_handlers(app: FastAPI) -> None:
    @app.exception_handler(AppError)
    async def handle_app_error(request: Request, exc: AppError) -> JSONResponse:
        body = {"error": {"code": exc.code, "message": exc.message, "details": None}}
        return JSONResponse(body, status_code=exc.status_code)

# file: app/schemas.py
from pydantic import BaseModel, ConfigDict, Field


class PromptIn(BaseModel):
    model_config = ConfigDict(extra="forbid")

    name: str = Field(min_length=3, max_length=60, pattern=r"^[a-z0-9-]+$")
    template: str = Field(min_length=1, max_length=4000)


class PromptOut(BaseModel):
    id: int
    name: str
    version: int
    template: str

# file: app/repository.py
class PromptRepository:
    """In-memory storage. Each instance has its own data: nothing is global."""

    def __init__(self) -> None:
        self._rows: dict[int, dict] = {}

    def add(self, data: dict) -> dict:
        row = {"id": len(self._rows) + 1, "version": 1, **data}
        self._rows[row["id"]] = row
        return row

    def get(self, prompt_id: int) -> dict | None:
        return self._rows.get(prompt_id)

    def get_by_name(self, name: str) -> dict | None:
        return next((row for row in self._rows.values() if row["name"] == name), None)

    def list(self, *, after_id: int = 0, limit: int = 20) -> list[dict]:
        rows = sorted(self._rows.values(), key=lambda row: row["id"])
        return [row for row in rows if row["id"] > after_id][:limit]

    def count(self) -> int:
        return len(self._rows)

# file: app/services.py
from app.errors import Conflict, NotFound
from app.repository import PromptRepository
from app.schemas import PromptIn


class PromptService:
    def __init__(self, repo: PromptRepository, *, max_prompts: int) -> None:
        self.repo = repo
        self.max_prompts = max_prompts

    def create(self, data: PromptIn) -> dict:
        if self.repo.count() >= self.max_prompts:
            raise Conflict(f"The library is full (limit: {self.max_prompts})")
        if self.repo.get_by_name(data.name):
            raise Conflict(f"A prompt named {data.name!r} already exists")
        return self.repo.add(data.model_dump())

    def get(self, prompt_id: int) -> dict:
        row = self.repo.get(prompt_id)
        if row is None:
            raise NotFound(f"Prompt {prompt_id} not found")
        return row

    def list(self, *, after_id: int = 0, limit: int = 20) -> list[dict]:
        return self.repo.list(after_id=after_id, limit=limit)

# file: app/settings.py
from functools import lru_cache

from pydantic import Field, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_prefix="PROMPT_", env_file=".env", extra="ignore")

    api_key: SecretStr                                   # required: no default
    environment: str = "development"
    upstream_url: str = "http://localhost:8800"         # the model provider (the practice API here)
    cors_origins: list[str] = ["http://localhost:3000"]
    log_level: str = "INFO"
    max_prompts: int = Field(default=100, ge=1)


@lru_cache
def get_settings() -> Settings:
    """One Settings object per process. Also a dependency (see below)."""
    return Settings()

from pydantic import ValidationError

from app.settings import Settings

try:
    Settings(_env_file=None)
except ValidationError as exc:
    print([(error["loc"], error["type"]) for error in exc.errors()])

[(('api_key',), 'missing')]

cat > .env <<'EOF'
PROMPT_API_KEY=dev-key-123
PROMPT_ENVIRONMENT=development
PROMPT_UPSTREAM_URL=http://localhost:8800
EOF
echo ".env" >> .gitignore

import os

os.environ["PROMPT_ENVIRONMENT"] = "staging"
os.environ["PROMPT_CORS_ORIGINS"] = '["http://localhost:3000", "https://app.bitbyte.example"]'
settings = Settings()
print(settings.environment, settings.cors_origins, settings.max_prompts)
print(settings.api_key, "|", settings.api_key.get_secret_value())
print(settings)
os.environ.pop("PROMPT_ENVIRONMENT"), os.environ.pop("PROMPT_CORS_ORIGINS")

staging ['http://localhost:3000', 'https://app.bitbyte.example'] 100
********** | dev-key-123
api_key=SecretStr('**********') environment='staging' upstream_url='http://localhost:8800' cors_origins=['http://localhost:3000', 'https://app.bitbyte.example'] log_level='INFO' max_prompts=100

# file: app/logging_setup.py
import json
import logging
from contextvars import ContextVar
from datetime import datetime, timezone

request_id_var: ContextVar[str] = ContextVar("request_id", default="-")


class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        data = {
            "ts": datetime.fromtimestamp(record.created, timezone.utc).isoformat(timespec="milliseconds"),
            "level": record.levelname,
            "logger": record.name,
            "request_id": request_id_var.get(),
            "message": record.getMessage(),
            **getattr(record, "fields", {}),            # extra={"fields": {...}}
        }
        if record.exc_info:
            data["exception"] = self.formatException(record.exc_info)
        return json.dumps(data)


def configure_logging(level: str) -> None:
    handler = logging.StreamHandler()                   # stderr: the platform collects it
    handler.setFormatter(JsonFormatter())
    logger = logging.getLogger("prompt_api")
    logger.handlers[:] = [handler]
    logger.setLevel(level)
    logger.propagate = False

# file: app/middleware.py
import logging
import time
import uuid

from fastapi import FastAPI, Request

from app.logging_setup import request_id_var

log = logging.getLogger("prompt_api.http")


def install_request_context(app: FastAPI) -> None:
    @app.middleware("http")
    async def request_context(request: Request, call_next):
        request_id = request.headers.get("X-Request-ID") or uuid.uuid4().hex[:12]
        token = request_id_var.set(request_id)
        start = time.perf_counter()
        status_code = 500                               # what we log if the app crashes
        try:
            response = await call_next(request)
            status_code = response.status_code
            elapsed_ms = (time.perf_counter() - start) * 1000
            response.headers["X-Request-ID"] = request_id
            response.headers["Server-Timing"] = f"app;dur={elapsed_ms:.1f}"
            return response
        finally:
            elapsed_ms = (time.perf_counter() - start) * 1000
            log.info("request", extra={"fields": {
                "method": request.method, "path": request.url.path,
                "status": status_code, "duration_ms": round(elapsed_ms, 1)}})
            request_id_var.reset(token)

# file: app/deps.py
import secrets
from typing import Annotated

import httpx
from fastapi import Depends, Header, Query, Request

from app.errors import Unauthorized
from app.repository import PromptRepository
from app.services import PromptService
from app.settings import Settings, get_settings

SettingsDep = Annotated[Settings, Depends(get_settings)]


def get_repo(request: Request) -> PromptRepository:
    return request.app.state.repo                      # created in lifespan


def get_http(request: Request) -> httpx.AsyncClient:
    return request.app.state.http                      # created in lifespan


def get_audit(request: Request) -> list[str]:
    return request.app.state.audit


RepoDep = Annotated[PromptRepository, Depends(get_repo)]
HttpDep = Annotated[httpx.AsyncClient, Depends(get_http)]
AuditDep = Annotated[list[str], Depends(get_audit)]


def get_service(repo: RepoDep, settings: SettingsDep) -> PromptService:
    return PromptService(repo, max_prompts=settings.max_prompts)


ServiceDep = Annotated[PromptService, Depends(get_service)]


class Pagination:
    """A class is a dependency too: its __init__ parameters become query parameters."""

    def __init__(self, limit: Annotated[int, Query(ge=1, le=100)] = 20,
                 after_id: Annotated[int, Query(ge=0)] = 0) -> None:
        self.limit = limit
        self.after_id = after_id


PageDep = Annotated[Pagination, Depends()]               # Depends() with no argument = the class itself


def require_api_key(settings: SettingsDep, x_api_key: Annotated[str | None, Header()] = None) -> None:
    """Rejects the request unless X-API-Key matches. Compared in constant time."""
    expected = settings.api_key.get_secret_value().encode()
    if x_api_key is None or not secrets.compare_digest(x_api_key.encode(), expected):
        raise Unauthorized("Missing or wrong X-API-Key")

from typing import Annotated, Iterator

from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient

trace: list[str] = []


def get_session() -> Iterator[str]:
    trace.append("open session")
    try:
        yield "session-1"
    finally:
        trace.append("close session")


SessionDep = Annotated[str, Depends(get_session)]


def get_unit(session: SessionDep) -> str:
    return f"unit({session})"


demo = FastAPI()


@demo.get("/work")
def work(session: SessionDep, unit: Annotated[str, Depends(get_unit)]) -> dict[str, str]:
    trace.append(f"handler got {session} and {unit}")
    return {"ok": "yes"}


with TestClient(demo) as client:
    client.get("/work")
print(trace)

['open session', 'handler got session-1 and unit(session-1)', 'close session']

# file: app/routes.py
import asyncio
import logging
from typing import Annotated

from fastapi import APIRouter, BackgroundTasks, Depends, Path, status

from app.deps import AuditDep, PageDep, ServiceDep, require_api_key
from app.schemas import PromptIn, PromptOut

log = logging.getLogger("prompt_api.routes")

router = APIRouter(prefix="/v1", tags=["prompts"], dependencies=[Depends(require_api_key)])


async def record_audit(audit: list[str], action: str, name: str) -> None:
    await asyncio.sleep(0.5)                            # pretend: write an audit record, call a webhook...
    audit.append(f"{action}:{name}")
    log.info("audit recorded", extra={"fields": {"action": action, "name": name}})


@router.post("/prompts", status_code=status.HTTP_201_CREATED)
def create_prompt(data: PromptIn, service: ServiceDep, audit: AuditDep,
                  background: BackgroundTasks) -> PromptOut:
    row = service.create(data)
    background.add_task(record_audit, audit, "created", row["name"])   # runs after the response is sent
    return row


@router.get("/prompts")
def list_prompts(service: ServiceDep, page: PageDep) -> list[PromptOut]:
    return service.list(after_id=page.after_id, limit=page.limit)


@router.get("/prompts/{prompt_id}")
def get_prompt(prompt_id: Annotated[int, Path(ge=1)], service: ServiceDep) -> PromptOut:
    return service.get(prompt_id)


@router.get("/audit")
def list_audit(audit: AuditDep) -> list[str]:
    return audit

# file: app/main.py
import logging
from contextlib import asynccontextmanager

import httpx
from fastapi import FastAPI, Response
from fastapi.middleware.cors import CORSMiddleware

from app import routes
from app.deps import HttpDep, RepoDep
from app.errors import install_error_handlers
from app.logging_setup import configure_logging
from app.middleware import install_request_context
from app.repository import PromptRepository
from app.settings import Settings, get_settings

log = logging.getLogger("prompt_api")


def create_app(settings: Settings | None = None) -> FastAPI:
    settings = settings or get_settings()

    @asynccontextmanager
    async def lifespan(app: FastAPI):
        configure_logging(settings.log_level)
        app.state.repo = PromptRepository()
        app.state.audit = []
        async with httpx.AsyncClient(base_url=settings.upstream_url, timeout=2.0) as http:
            app.state.http = http                       # one pooled client for the whole process
            log.info("startup", extra={"fields": {"environment": settings.environment}})
            yield
        log.info("shutdown")                            # the client is closed by now

    app = FastAPI(title="BitByte Prompt Library", lifespan=lifespan)
    app.dependency_overrides[get_settings] = lambda: settings
    install_error_handlers(app)
    install_request_context(app)
    app.add_middleware(                                 # added last = outermost
        CORSMiddleware,
        allow_origins=settings.cors_origins,
        allow_methods=["GET", "POST"],
        allow_headers=["X-API-Key", "Content-Type", "X-Request-ID"],
        expose_headers=["X-Request-ID"],
        max_age=600,
    )
    app.include_router(routes.router)

    @app.get("/health", tags=["ops"])
    def health() -> dict[str, str]:
        """Liveness: the process is up. Cheap, and never touches other services."""
        return {"status": "ok"}

    @app.get("/ready", tags=["ops"])
    async def ready(repo: RepoDep, http: HttpDep, response: Response) -> dict:
        """Readiness: can this instance do its job right now?"""
        checks = {"prompt_store": "ok" if repo.count() >= 0 else "down"}
        try:
            (await http.get("/health")).raise_for_status()
            checks["upstream"] = "ok"
        except httpx.HTTPError:
            checks["upstream"] = "down"
        is_ready = all(value == "ok" for value in checks.values())
        if not is_ready:
            response.status_code = 503
        return {"status": "ready" if is_ready else "not_ready", "checks": checks}

    return app


app = create_app()

from app.main import create_app

test_settings = Settings(api_key="test-key", max_prompts=1, log_level="CRITICAL")
headers = {"X-API-Key": "test-key"}
body = {"name": "faq-answer", "template": "FAQ: {input}"}

with TestClient(create_app(test_settings)) as client:
    print(client.post("/v1/prompts", json=body, headers=headers).status_code)
    print(client.post("/v1/prompts", json={**body, "name": "second"}, headers=headers).json())
with TestClient(create_app(test_settings)) as another:
    print(another.get("/v1/prompts", headers=headers).json())

201
{'error': {'code': 'conflict', 'message': 'The library is full (limit: 1)', 'details': None}}
[]

uv run fastapi run --port 8000 > server.log 2>&1 &
export SERVER_PID=$!
curl -s --retry 20 --retry-connrefused --retry-delay 0 http://localhost:8000/health; echo
curl -s -w ' %{http_code}\n' http://localhost:8000/ready

{"status":"ok"}
{"status":"ready","checks":{"prompt_store":"ok","upstream":"ok"}} 200

curl -s -w ' %{http_code}\n' -X POST http://localhost:8000/v1/prompts \
  -H 'Content-Type: application/json' -d '{"name": "faq-answer", "template": "FAQ: {input}"}'
curl -s -D - -X POST http://localhost:8000/v1/prompts -H 'X-API-Key: dev-key-123' -H 'X-Request-ID: nadia-req-001' \
  -H 'Content-Type: application/json' -d '{"name": "faq-answer", "template": "FAQ: {input}"}' \
  | tr -d '\r' | sed -E 's/dur=[0-9.]+/dur=N/' | grep -iE '^(HTTP|x-request-id|server-timing|\{)'
sleep 1

{"error":{"code":"unauthorized","message":"Missing or wrong X-API-Key","details":null}} 401
HTTP/1.1 201 Created
x-request-id: nadia-req-001
server-timing: app;dur=N
{"id":1,"name":"faq-answer","version":1,"template":"FAQ: {input}"}

curl -s -o /dev/null -X POST http://localhost:8000/v1/prompts -H 'X-API-Key: dev-key-123' \
  -H 'Content-Type: application/json' -d '{"name": "summarise", "template": "Summarise: {input}"}'
curl -s http://localhost:8000/v1/audit -H 'X-API-Key: dev-key-123'; echo
sleep 1
curl -s http://localhost:8000/v1/audit -H 'X-API-Key: dev-key-123'; echo

["created:faq-answer"]
["created:faq-answer","created:summarise"]

import asyncio
import time

import httpx
from fastapi import FastAPI

lab = FastAPI()


@lab.get("/async-blocking")
async def async_blocking() -> dict[str, str]:
    time.sleep(0.3)                          # blocks the whole event loop
    return {"done": "yes"}


@lab.get("/sync-blocking")
def sync_blocking() -> dict[str, str]:
    time.sleep(0.3)                          # blocks only one worker thread
    return {"done": "yes"}


@lab.get("/async-waiting")
async def async_waiting() -> dict[str, str]:
    await asyncio.sleep(0.3)                 # yields to the event loop while waiting
    return {"done": "yes"}


async def burst(app: FastAPI, path: str, requests: int = 5) -> float:
    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://lab") as client:
        start = time.perf_counter()
        await asyncio.gather(*(client.get(path) for _ in range(requests)))
        return time.perf_counter() - start


for path in ("/async-blocking", "/sync-blocking", "/async-waiting"):
    seconds = asyncio.run(burst(lab, path))
    print(f"{path:16} 5 requests ran", "one after another" if seconds > 0.9 else "at the same time")

/async-blocking  5 requests ran one after another
/sync-blocking   5 requests ran at the same time
/async-waiting   5 requests ran at the same time

curl -s -D - -o /dev/null -X OPTIONS http://localhost:8000/v1/prompts \
  -H 'Origin: http://localhost:3000' -H 'Access-Control-Request-Method: POST' \
  -H 'Access-Control-Request-Headers: x-api-key,content-type' | tr -d '\r' | grep -iE '^(HTTP|access-control)' | sort
curl -s -w ' %{http_code}\n' -X OPTIONS http://localhost:8000/v1/prompts \
  -H 'Origin: https://evil.example' -H 'Access-Control-Request-Method: POST'

HTTP/1.1 200 OK
access-control-allow-headers: Accept, Accept-Language, Content-Language, Content-Type, X-API-Key, X-Request-ID
access-control-allow-methods: GET, POST
access-control-allow-origin: http://localhost:3000
access-control-max-age: 600
Disallowed CORS origin 400

import json

for line in open("server.log"):
    if line.startswith("{"):
        record = json.loads(line)
        extra = {k: v for k, v in record.items()
                 if k not in ("ts", "level", "logger", "request_id", "message", "duration_ms")}
        rid = record["request_id"]
        shown = rid if rid == "-" or rid.startswith("nadia-") else "(generated)"
        print(f'{shown:14} {record["level"]:5} {record["message"]:15}', extra)

-              INFO  startup         {'environment': 'development'}
(generated)    INFO  request         {'method': 'GET', 'path': '/health', 'status': 200}
(generated)    INFO  request         {'method': 'GET', 'path': '/ready', 'status': 200}
(generated)    INFO  request         {'method': 'POST', 'path': '/v1/prompts', 'status': 401}
nadia-req-001  INFO  request         {'method': 'POST', 'path': '/v1/prompts', 'status': 201}
nadia-req-001  INFO  audit recorded  {'action': 'created', 'name': 'faq-answer'}
(generated)    INFO  request         {'method': 'POST', 'path': '/v1/prompts', 'status': 201}
(generated)    INFO  request         {'method': 'GET', 'path': '/v1/audit', 'status': 200}
(generated)    INFO  audit recorded  {'action': 'created', 'name': 'summarise'}
(generated)    INFO  request         {'method': 'GET', 'path': '/v1/audit', 'status': 200}

kill "$SERVER_PID"
wait "$SERVER_PID" 2>/dev/null
PROMPT_UPSTREAM_URL=http://localhost:9 uv run fastapi run --port 8001 > server2.log 2>&1 &
export SERVER_PID=$!
curl -s --retry 20 --retry-connrefused --retry-delay 0 http://localhost:8001/health; echo
curl -s -w ' %{http_code}\n' http://localhost:8001/ready
kill "$SERVER_PID"

{"status":"ok"}
{"status":"not_ready","checks":{"prompt_store":"ok","upstream":"down"}} 503

# 1. Easy
try:
    Settings(api_key="k", max_prompts=0)
except ValidationError as exc:
    print(exc.errors()[0]["msg"])


# 2. Medium
from app.logging_setup import request_id_var


def get_request_id() -> str:
    return request_id_var.get()


app = create_app(test_settings)


@app.get("/whoami")
def whoami(request_id: Annotated[str, Depends(get_request_id)]) -> dict[str, str]:
    return {"request_id": request_id}


with TestClient(app) as client:
    response = client.get("/whoami", headers={"X-Request-ID": "my-id-42"})
    print(response.json(), response.headers["X-Request-ID"])


# 3. Hard
def slow_work() -> str:
    time.sleep(0.3)
    return "done"


@lab.get("/async-fixed")
async def async_fixed() -> dict[str, str]:
    return {"done": await asyncio.to_thread(slow_work)}       # blocking code moved to a thread


seconds = asyncio.run(burst(lab, "/async-fixed"))
print("5 requests ran", "one after another" if seconds > 0.9 else "at the same time")

Input should be greater than or equal to 1
{'request_id': 'my-id-42'} my-id-42
5 requests ran at the same time