FastAPI আর্কিটেকচার: রাউটার, স্কিমা ও এরর হ্যান্ডলিং

prompt-api/
├── pyproject.toml            dependencies + [tool.fastapi] entrypoint
└── app/
    ├── __init__.py
    ├── main.py               create_app(): metadata, error handlers, routers
    ├── errors.py             AppError classes and the one error format
    ├── schemas.py            Pydantic models: what goes in and out over HTTP
    ├── repository.py         storage (in memory now, a database later)
    ├── services.py           business rules: no HTTP, no SQL
    └── routers/
        ├── __init__.py
        ├── health.py
        ├── prompts_v1.py     /v1/prompts
        └── prompts_v2.py     /v2/prompts

mkdir -p ~/projects/prompt-api
cd ~/projects/prompt-api
uv init --bare -q
uv add -q "fastapi[standard]"
# pip দিয়ে: python -m venv .venv && source .venv/bin/activate && pip install -q "fastapi[standard]"
mkdir -p app/routers
touch app/__init__.py app/routers/__init__.py
cat >> pyproject.toml <<'EOF'

[tool.fastapi]
entrypoint = "app.main:app"
EOF
tail -3 pyproject.toml

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

# file: app/schemas.py
"""Request and response models for the Prompt Library."""
import string
from typing import Literal

from pydantic import BaseModel, ConfigDict, Field, computed_field, field_validator, model_validator


def template_variables(template: str) -> list[str]:
    """Names used as {placeholders} in a template, in order, without duplicates."""
    names = [field for _, field, _, _ in string.Formatter().parse(template) if field]
    return list(dict.fromkeys(names))


class ModelSettings(BaseModel):
    """How the LLM is called. Nested inside every prompt."""
    model_config = ConfigDict(extra="forbid")

    model: Literal["practice-llm-1", "practice-llm-2"] = "practice-llm-1"
    temperature: float = Field(default=0.2, ge=0, le=2)
    max_tokens: int = Field(default=512, ge=1, le=4096)


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

    name: str = Field(min_length=3, max_length=60, pattern=r"^[a-z0-9-]+$",
                      description="Unique slug, e.g. support-answer")
    template: str = Field(min_length=1, max_length=4000)
    tags: list[str] = Field(default_factory=list, max_length=5)
    settings: ModelSettings = Field(default_factory=ModelSettings)

    @field_validator("tags")
    @classmethod
    def normalise_tags(cls, tags: list[str]) -> list[str]:
        return sorted({tag.strip().lower() for tag in tags if tag.strip()})

    @model_validator(mode="after")
    def template_needs_input(self) -> "PromptIn":
        if "input" not in template_variables(self.template):
            raise ValueError("template must contain an {input} placeholder")
        return self


class PromptOut(BaseModel):
    id: int
    name: str
    version: int
    template: str
    tags: list[str]
    settings: ModelSettings

    @computed_field
    @property
    def variables(self) -> list[str]:
        return template_variables(self.template)


class RenderIn(BaseModel):
    values: dict[str, str] = Field(description="A value for every template variable")


class RenderOut(BaseModel):
    prompt_id: int
    version: int
    text: str
    settings: ModelSettings


class PromptPage(BaseModel):
    """v2 list envelope: items plus a cursor for the next page."""
    items: list[PromptOut]
    next_cursor: int | None

from pydantic import ValidationError

from app.schemas import PromptIn, PromptOut

data = {
    "name": "  support-answer ",
    "template": "You are BitByte's support assistant.\nAnswer in {language}: {input}",
    "tags": ["Support", "support", " RAG "],
    "settings": {"temperature": 0.3},
}
prompt = PromptIn.model_validate(data)
print(repr(prompt.name), prompt.tags)
print(prompt.settings)

try:
    PromptIn.model_validate({"name": "Bad Name!", "template": "Summarise this.", "colour": "red",
                             "settings": {"temperature": 3}})
except ValidationError as exc:
    for error in exc.errors():
        print(error["loc"], error["type"], "-", error["msg"])

'support-answer' ['rag', 'support']
model='practice-llm-1' temperature=0.3 max_tokens=512
('name',) string_pattern_mismatch - String should match pattern '^[a-z0-9-]+$'
('settings', 'temperature') less_than_equal - Input should be less than or equal to 2
('colour',) extra_forbidden - Extra inputs are not permitted

out = PromptOut(id=1, version=1, **prompt.model_dump())
print(out.model_dump(include={"name", "variables"}))
print(out.model_dump_json(exclude={"template", "settings"}))

{'name': 'support-answer', 'variables': ['language', 'input']}
{"id":1,"name":"support-answer","version":1,"tags":["rag","support"],"variables":["language","input"]}

from datetime import datetime, timedelta, timezone

from pydantic import BaseModel, ConfigDict
from pydantic.alias_generators import to_camel


class UsageOut(BaseModel):
    model_config = ConfigDict(alias_generator=to_camel, validate_by_name=True, serialize_by_alias=True)

    prompt_tokens: int
    completion_tokens: int
    created_at: datetime


usage = UsageOut(prompt_tokens=12, completion_tokens=30,
                 created_at=datetime(2026, 1, 15, 10, 0, tzinfo=timezone(timedelta(hours=6))))
print(usage.model_dump_json())
print(UsageOut.model_validate({"promptTokens": 1, "completionTokens": 2, "createdAt": "2026-01-15T04:00:00Z"}))

{"promptTokens":12,"completionTokens":30,"createdAt":"2026-01-15T10:00:00+06:00"}
prompt_tokens=1 completion_tokens=2 created_at=datetime.datetime(2026, 1, 15, 4, 0, tzinfo=TzInfo(0))

{
  "error": {
    "code": "not_found",
    "message": "Prompt 42 not found",
    "details": null
  }
}

# file: app/errors.py
"""Application errors and the handlers that turn every error into one JSON shape."""
import logging
from typing import Any

from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from starlette.exceptions import HTTPException as StarletteHTTPException

logger = logging.getLogger("prompt_api")


class ErrorBody(BaseModel):
    code: str
    message: str
    details: Any = None


class ErrorResponse(BaseModel):
    error: ErrorBody


class AppError(Exception):
    """Base class for errors the service raises on purpose."""
    status_code = 400
    code = "bad_request"

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


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


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


class MissingVariables(AppError):
    status_code = 422
    code = "missing_variables"


HTTP_CODES = {401: "unauthorized", 403: "forbidden", 404: "not_found", 405: "method_not_allowed"}


def error_response(status: int, code: str, message: str, details: Any = None,
                   headers: dict[str, str] | None = None) -> JSONResponse:
    body = ErrorResponse(error=ErrorBody(code=code, message=message, details=details))
    return JSONResponse(body.model_dump(), status_code=status, headers=headers)


def install_error_handlers(app: FastAPI) -> None:
    @app.exception_handler(AppError)
    async def handle_app_error(request: Request, exc: AppError) -> JSONResponse:
        return error_response(exc.status_code, exc.code, exc.message, exc.details)

    @app.exception_handler(RequestValidationError)
    async def handle_validation(request: Request, exc: RequestValidationError) -> JSONResponse:
        details = [
            {"field": ".".join(str(part) for part in err["loc"]), "message": err["msg"], "type": err["type"]}
            for err in exc.errors()
        ]
        return error_response(422, "validation_error", "The request is not valid", details)

    @app.exception_handler(StarletteHTTPException)
    async def handle_http(request: Request, exc: StarletteHTTPException) -> JSONResponse:
        code = HTTP_CODES.get(exc.status_code, f"http_{exc.status_code}")
        return error_response(exc.status_code, code, str(exc.detail), headers=exc.headers)

    @app.exception_handler(Exception)
    async def handle_unexpected(request: Request, exc: Exception) -> JSONResponse:
        logger.exception("unhandled error on %s %s", request.method, request.url.path)
        return error_response(500, "internal_error", "Something went wrong on our side")

# file: app/repository.py
"""Prompt storage. In memory for now: the interface is what matters."""


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

    def add(self, data: dict) -> dict:
        row = {"id": self._next_id, "version": 1, **data}
        self._rows[row["id"]] = row
        self._next_id += 1
        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, *, tag: str | None = None, after_id: int = 0, limit: int = 20) -> list[dict]:
        rows = [row for row in self._rows.values()
                if row["id"] > after_id and (tag is None or tag in row["tags"])]
        return sorted(rows, key=lambda row: row["id"])[:limit]

    def replace(self, prompt_id: int, data: dict) -> dict:
        row = self._rows[prompt_id]
        row.update(data, version=row["version"] + 1)
        return row

    def delete(self, prompt_id: int) -> None:
        del self._rows[prompt_id]

# file: app/services.py
"""Business rules for prompts. No HTTP and no storage details in here."""
from app.errors import Conflict, MissingVariables, NotFound
from app.repository import PromptRepository
from app.schemas import PromptIn, template_variables


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

    def create(self, data: PromptIn) -> dict:
        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, *, tag: str | None = None, after_id: int = 0, limit: int = 20) -> list[dict]:
        return self.repo.list(tag=tag, after_id=after_id, limit=limit)

    def new_version(self, prompt_id: int, data: PromptIn) -> dict:
        current = self.get(prompt_id)
        if data.name != current["name"]:
            raise Conflict("The name of a prompt cannot change; create a new prompt instead")
        return self.repo.replace(prompt_id, data.model_dump())

    def delete(self, prompt_id: int) -> None:
        self.get(prompt_id)
        self.repo.delete(prompt_id)

    def render(self, prompt_id: int, values: dict[str, str]) -> tuple[dict, str]:
        row = self.get(prompt_id)
        missing = [name for name in template_variables(row["template"]) if name not in values]
        if missing:
            raise MissingVariables("Some template variables have no value", details={"missing": missing})
        return row, row["template"].format(**values)


_service = PromptService(PromptRepository())


def get_prompt_service() -> PromptService:
    """Dependency: one shared service (the next page replaces this with lifespan state)."""
    return _service

from app.errors import AppError
from app.repository import PromptRepository
from app.services import PromptService

service = PromptService(PromptRepository())
row = service.create(prompt)
print(row["id"], row["name"], row["version"])
for attempt in (lambda: service.create(prompt), lambda: service.get(7), lambda: service.render(1, {"input": "hi"})):
    try:
        attempt()
    except AppError as exc:
        print(type(exc).__name__, exc.status_code, exc.code, "-", exc.message, exc.details or "")
print(service.render(1, {"input": "How do I reset my password?", "language": "Bangla"})[1])

1 support-answer 1
Conflict 409 conflict - A prompt named 'support-answer' already exists 
NotFound 404 not_found - Prompt 7 not found 
MissingVariables 422 missing_variables - Some template variables have no value {'missing': ['language']}
You are BitByte's support assistant.
Answer in Bangla: How do I reset my password?

# file: app/routers/prompts_v1.py
from typing import Annotated

from fastapi import APIRouter, Depends, Path, Query, Response, status

from app.errors import ErrorResponse
from app.schemas import PromptIn, PromptOut, RenderIn, RenderOut
from app.services import PromptService, get_prompt_service

router = APIRouter(
    prefix="/v1/prompts",
    tags=["prompts"],
    responses={404: {"model": ErrorResponse, "description": "Prompt not found"}},
)

ServiceDep = Annotated[PromptService, Depends(get_prompt_service)]
PromptId = Annotated[int, Path(ge=1, description="The prompt's id")]


@router.get("", deprecated=True, summary="List prompts (deprecated: use /v2/prompts)")
def list_prompts(
    service: ServiceDep,
    response: Response,
    tag: Annotated[str | None, Query(max_length=30)] = None,
    limit: Annotated[int, Query(ge=1, le=100)] = 20,
) -> list[PromptOut]:
    response.headers["Deprecation"] = "@1798761600"          # 2027-01-01 (RFC 9745)
    response.headers["Sunset"] = "Fri, 01 Jan 2027 00:00:00 GMT"   # RFC 8594
    response.headers["Link"] = '</v2/prompts>; rel="successor-version"'
    return service.list(tag=tag, limit=limit)


@router.post("", status_code=status.HTTP_201_CREATED,
             responses={409: {"model": ErrorResponse, "description": "Name already used"}})
def create_prompt(data: PromptIn, service: ServiceDep) -> PromptOut:
    return service.create(data)


@router.get("/{prompt_id}")
def get_prompt(prompt_id: PromptId, service: ServiceDep) -> PromptOut:
    return service.get(prompt_id)


@router.put("/{prompt_id}", summary="Save a new version of a prompt")
def update_prompt(prompt_id: PromptId, data: PromptIn, service: ServiceDep) -> PromptOut:
    return service.new_version(prompt_id, data)


@router.delete("/{prompt_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_prompt(prompt_id: PromptId, service: ServiceDep) -> None:
    service.delete(prompt_id)


@router.post("/{prompt_id}/render",
             responses={422: {"model": ErrorResponse, "description": "Missing or invalid values"}})
def render_prompt(prompt_id: PromptId, body: RenderIn, service: ServiceDep) -> RenderOut:
    row, text = service.render(prompt_id, body.values)
    return RenderOut(prompt_id=row["id"], version=row["version"], text=text, settings=row["settings"])

# file: app/routers/health.py
from fastapi import APIRouter

router = APIRouter(tags=["ops"])


@router.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok"}

# file: app/main.py
import re

from fastapi import FastAPI
from fastapi.routing import APIRoute

from app.errors import install_error_handlers
from app.routers import health, prompts_v1, prompts_v2

DESCRIPTION = """
Store **versioned prompt templates** for LLM features and render them with values.

All errors use one shape: `{"error": {"code", "message", "details"}}`.
"""


def operation_id(route: APIRoute) -> str:
    """Readable, unique operation ids such as v1-get_prompt (they become method names in generated clients)."""
    first = route.path.strip("/").split("/")[0]
    return f"{first}-{route.name}" if re.fullmatch(r"v\d+", first) else route.name


def create_app() -> FastAPI:
    app = FastAPI(
        title="BitByte Prompt Library",
        version="1.2.0",
        summary="Versioned prompt templates for BitByte's AI features.",
        description=DESCRIPTION,
        contact={"name": "BitByte AI team", "email": "ai-team@example.com"},
        license_info={"name": "MIT", "identifier": "MIT"},
        openapi_tags=[
            {"name": "prompts", "description": "Create, version and render prompts."},
            {"name": "ops", "description": "Health checks for load balancers and monitoring."},
        ],
        generate_unique_id_function=operation_id,
    )
    install_error_handlers(app)
    app.include_router(health.router)
    app.include_router(prompts_v1.router)
    app.include_router(prompts_v2.router)
    return app


app = create_app()

# file: app/routers/prompts_v2.py
from typing import Annotated

from fastapi import APIRouter, Depends, Query

from app.schemas import PromptOut, PromptPage
from app.services import PromptService, get_prompt_service

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

ServiceDep = Annotated[PromptService, Depends(get_prompt_service)]


@router.get("", summary="List prompts with cursor pagination")
def list_prompts(
    service: ServiceDep,
    tag: Annotated[str | None, Query(max_length=30)] = None,
    cursor: Annotated[int, Query(ge=0, description="next_cursor from the previous page")] = 0,
    limit: Annotated[int, Query(ge=1, le=100)] = 20,
) -> PromptPage:
    rows = service.list(tag=tag, after_id=cursor, limit=limit + 1)
    items = [PromptOut.model_validate(row) for row in rows[:limit]]
    next_cursor = items[-1].id if len(rows) > limit else None
    return PromptPage(items=items, next_cursor=next_cursor)

uv run fastapi run --port 8000 > server.log 2>&1 &
echo $! > server.pid
curl -s --retry 20 --retry-connrefused --retry-delay 0 http://localhost:8000/health

{"status":"ok"}

curl -s -w ' %{http_code}\n' -X POST http://localhost:8000/v1/prompts \
  -H 'Content-Type: application/json' \
  -d '{"name": "support-answer", "template": "Answer in {language}: {input}", "tags": ["Support", "RAG"]}'
curl -s -w ' %{http_code}\n' -X POST http://localhost:8000/v1/prompts \
  -H 'Content-Type: application/json' \
  -d '{"name": "summarise", "template": "Summarise in 3 bullet points:\n{input}", "tags": ["batch"], "settings": {"model": "practice-llm-2", "max_tokens": 300}}'
curl -s -w ' %{http_code}\n' -X POST http://localhost:8000/v1/prompts \
  -H 'Content-Type: application/json' \
  -d '{"name": "summarise", "template": "Summary: {input}"}'

{"id":1,"name":"support-answer","version":1,"template":"Answer in {language}: {input}","tags":["rag","support"],"settings":{"model":"practice-llm-1","temperature":0.2,"max_tokens":512},"variables":["language","input"]} 201
{"id":2,"name":"summarise","version":1,"template":"Summarise in 3 bullet points:\n{input}","tags":["batch"],"settings":{"model":"practice-llm-2","temperature":0.2,"max_tokens":300},"variables":["input"]} 201
{"error":{"code":"conflict","message":"A prompt named 'summarise' already exists","details":null}} 409

curl -s -X POST http://localhost:8000/v1/prompts -H 'Content-Type: application/json' \
  -d '{"name": "x", "template": "No placeholder here", "settings": {"temperature": 5}}' | python -m json.tool

{
    "error": {
        "code": "validation_error",
        "message": "The request is not valid",
        "details": [
            {
                "field": "body.name",
                "message": "String should have at least 3 characters",
                "type": "string_too_short"
            },
            {
                "field": "body.settings.temperature",
                "message": "Input should be less than or equal to 2",
                "type": "less_than_equal"
            }
        ]
    }
}

curl -s -w ' %{http_code}\n' http://localhost:8000/v1/prompts/42
curl -s -w ' %{http_code}\n' http://localhost:8000/v1/prompts/abc
curl -s -w ' %{http_code}\n' http://localhost:8000/v1/nothing-here
curl -s -w ' %{http_code}\n' -X PATCH http://localhost:8000/v1/prompts/1
curl -s -w ' %{http_code}\n' -X POST http://localhost:8000/v1/prompts/1/render \
  -H 'Content-Type: application/json' -d '{"values": {"input": "Where is my certificate?"}}'

{"error":{"code":"not_found","message":"Prompt 42 not found","details":null}} 404
{"error":{"code":"validation_error","message":"The request is not valid","details":[{"field":"path.prompt_id","message":"Input should be a valid integer, unable to parse string as an integer","type":"int_parsing"}]}} 422
{"error":{"code":"not_found","message":"Not Found","details":null}} 404
{"error":{"code":"method_not_allowed","message":"Method Not Allowed","details":null}} 405
{"error":{"code":"missing_variables","message":"Some template variables have no value","details":{"missing":["language"]}}} 422

curl -s -X POST http://localhost:8000/v1/prompts/1/render -H 'Content-Type: application/json' \
  -d '{"values": {"input": "Where is my certificate?", "language": "Bangla"}}' | python -m json.tool
curl -s 'http://localhost:8000/v2/prompts?limit=1'; echo
curl -s 'http://localhost:8000/v2/prompts?limit=1&cursor=1'; echo

{
    "prompt_id": 1,
    "version": 1,
    "text": "Answer in Bangla: Where is my certificate?",
    "settings": {
        "model": "practice-llm-1",
        "temperature": 0.2,
        "max_tokens": 512
    }
}
{"items":[{"id":1,"name":"support-answer","version":1,"template":"Answer in {language}: {input}","tags":["rag","support"],"settings":{"model":"practice-llm-1","temperature":0.2,"max_tokens":512},"variables":["language","input"]}],"next_cursor":1}
{"items":[{"id":2,"name":"summarise","version":1,"template":"Summarise in 3 bullet points:\n{input}","tags":["batch"],"settings":{"model":"practice-llm-2","temperature":0.2,"max_tokens":300},"variables":["input"]}],"next_cursor":null}

curl -s -D - -o /dev/null 'http://localhost:8000/v1/prompts?tag=rag' | tr -d '\r' | grep -iE '^(HTTP|deprecation|sunset|link)'
curl -s 'http://localhost:8000/v1/prompts?tag=rag' | python -c "import sys, json; print([p['name'] for p in json.load(sys.stdin)])"

HTTP/1.1 200 OK
deprecation: @1798761600
sunset: Fri, 01 Jan 2027 00:00:00 GMT
link: </v2/prompts>; rel="successor-version"
['support-answer']

import httpx

spec = httpx.get("http://localhost:8000/openapi.json").json()
print(spec["openapi"], "|", spec["info"]["title"], spec["info"]["version"], "|", spec["info"]["license"])
print([tag["name"] for tag in spec["tags"]])
for path, methods in spec["paths"].items():
    for method, op in methods.items():
        flag = " (deprecated)" if op.get("deprecated") else ""
        print(f"{method.upper():6} {path:30} {op['operationId']}{flag}")
print(sorted(spec["paths"]["/v1/prompts/{prompt_id}"]["get"]["responses"]))
print(spec["components"]["schemas"]["ErrorBody"]["required"])

3.1.0 | BitByte Prompt Library 1.2.0 | {'name': 'MIT', 'identifier': 'MIT'}
['prompts', 'ops']
GET    /health                        health
GET    /v1/prompts                    v1-list_prompts (deprecated)
POST   /v1/prompts                    v1-create_prompt
GET    /v1/prompts/{prompt_id}        v1-get_prompt
PUT    /v1/prompts/{prompt_id}        v1-update_prompt
DELETE /v1/prompts/{prompt_id}        v1-delete_prompt
POST   /v1/prompts/{prompt_id}/render v1-render_prompt
GET    /v2/prompts                    v2-list_prompts
['200', '404', '422']
['code', 'message']

kill "$(cat server.pid)"

# 1. সহজ
from app.errors import AppError


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


exc = Unauthorized("Log in to use the prompt library")
print(exc.status_code, exc.code, exc.message)


# 2. মাঝারি
from pydantic import field_validator

from app.schemas import RenderIn


class SafeRenderIn(RenderIn):
    @field_validator("values")
    @classmethod
    def limit_length(cls, values: dict[str, str]) -> dict[str, str]:
        too_long = [name for name, value in values.items() if len(value) > 2000]
        if too_long:
            raise ValueError(f"values too long (max 2000 characters): {too_long}")
        return values


try:
    SafeRenderIn(values={"input": "x" * 2001})
except ValidationError as exc:
    print(exc.errors()[0]["msg"])

401 unauthorized Log in to use the prompt library
Value error, values too long (max 2000 characters): ['input']

uv add -q --dev httpx2

# 3. কঠিন
from fastapi.testclient import TestClient

from app.main import create_app

with TestClient(create_app()) as tc:
    body = {"name": "faq-answer", "template": "FAQ: {input}"}
    created = tc.post("/v1/prompts", json=body).json()
    v2 = tc.put(f"/v1/prompts/{created['id']}", json={**body, "template": "FAQ (short): {input}"})
    print(v2.status_code, v2.json()["version"], v2.json()["template"])
    renamed = tc.put(f"/v1/prompts/{created['id']}", json={**body, "name": "faq-other"})
    print(renamed.status_code, renamed.json())

with TestClient(create_app()) as another:          # একটা "নতুন" অ্যাপ...
    print([p["name"] for p in another.get("/v2/prompts").json()["items"]])

200 2 FAQ (short): {input}
409 {'error': {'code': 'conflict', 'message': 'The name of a prompt cannot change; create a new prompt instead', 'details': None}}
['faq-answer']