Building Services with FastAPI¶

from fastapi import FastAPI, HTTPException, Query, Path
from pydantic import BaseModel, Field
from datetime import datetime

app = FastAPI(
    title="ACME Orders API",
    description="Create and retrieve customer orders",
    version="2.1.0",
)

# ── Pydantic models define the API contract ───────────────────────
class OrderIn(BaseModel):
    customer_id: str  = Field(min_length=1, max_length=50)
    product_id:  str  = Field(min_length=1)
    quantity:    int  = Field(ge=1, le=1000)    # 1 ≤ quantity ≤ 1000
    notes:       str | None = None               # optional field

class OrderOut(OrderIn):
    id:         str
    status:     str = "pending"
    created_at: datetime

# In-memory store (replace with DB in production)
_orders: dict[str, OrderOut] = {}

# ── Route handlers ─────────────────────────────────────────────────
@app.post("/orders", response_model=OrderOut, status_code=201,
          summary="Create a new order")
async def create_order(order: OrderIn) -> OrderOut:
    """Creates a pending order. Returns 422 if validation fails."""
    new_id = f"ord-{len(_orders) + 1:04d}"
    new = OrderOut(id=new_id, created_at=datetime.utcnow(), **order.model_dump())
    _orders[new_id] = new
    return new

@app.get("/orders/{order_id}", response_model=OrderOut,
         summary="Get an order by ID")
async def get_order(
    order_id: str = Path(description="The order identifier"),
) -> OrderOut:
    if order_id not in _orders:
        raise HTTPException(status_code=404, detail=f"Order {order_id!r} not found")
    return _orders[order_id]

@app.get("/orders", response_model=list[OrderOut],
         summary="List orders with optional filters")
async def list_orders(
    customer_id: str | None = Query(default=None, description="Filter by customer"),
    status:      str | None = Query(default=None, description="Filter by status"),
    limit:       int        = Query(default=20, ge=1, le=100),
) -> list[OrderOut]:
    results = list(_orders.values())
    if customer_id:
        results = [o for o in results if o.customer_id == customer_id]
    if status:
        results = [o for o in results if o.status == status]
    return results[:limit]

# Run: uv run uvicorn main:app --reload
# Docs: http://localhost:8000/docs  (Swagger UI, auto-generated)

from fastapi import Depends, HTTPException, Header
from sqlalchemy.orm import Session
from database import SessionLocal

# ── Database session dependency ───────────────────────────────────
def get_db():
    """Yield a DB session per request; close it on completion."""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# ── Authentication dependency ─────────────────────────────────────
def require_auth(x_api_key: str = Header(alias="X-API-Key")) -> str:
    """Extract and validate the API key from the request header."""
    valid_keys = {"secret-key-1", "secret-key-2"}   # use a DB in production
    if x_api_key not in valid_keys:
        raise HTTPException(status_code=401, detail="Invalid API key")
    return x_api_key

# ── Using dependencies in route handlers ─────────────────────────
@app.post("/orders", response_model=OrderOut, status_code=201)
async def create_order(
    order: OrderIn,
    db: Session = Depends(get_db),         # injected DB session
    _: str      = Depends(require_auth),   # authentication enforced
) -> OrderOut:
    # db is a live session; _ is the validated API key (unused here)
    db_order = Order(**order.model_dump())
    db.add(db_order)
    db.commit()
    db.refresh(db_order)
    return OrderOut.model_validate(db_order)

import httpx
import asyncio
from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

# ── Async handler: non-blocking I/O ──────────────────────────────
@app.get("/enriched-orders/{order_id}")
async def get_enriched_order(order_id: str) -> dict:
    """Fetch order AND customer details in parallel — both non-blocking."""
    async with httpx.AsyncClient(timeout=10.0) as client:
        order_task    = client.get(f"https://orders.internal/orders/{order_id}")
        customer_task = client.get(f"https://crm.internal/customers/{order_id}")
        order_resp, customer_resp = await asyncio.gather(order_task, customer_task)
    order_resp.raise_for_status()
    customer_resp.raise_for_status()
    return {**order_resp.json(), "customer": customer_resp.json()}

# ── Background tasks: fire-and-forget ─────────────────────────────
def send_confirmation_email(order_id: str, email: str) -> None:
    # Runs after the response is sent — doesn't delay the client
    import smtplib  # simplified for illustration

@app.post("/orders", status_code=201)
async def create_order_with_notification(
    order: OrderIn,
    background_tasks: BackgroundTasks,
) -> dict:
    order_id = "ord-0001"   # simplified
    background_tasks.add_task(
        send_confirmation_email,
        order_id=order_id,
        email="customer@example.com",
    )
    return {"id": order_id, "status": "pending"}