Testing, middleware, and uploads

# pip install pytest httpx
from typing import Annotated
from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient

app = FastAPI()

def current_user():
    return {"id": 1}

@app.get("/me")
def me(user: Annotated[dict, Depends(current_user)]):
    return user

@app.get("/books/{book_id}")
def book(book_id: int):
    return {"id": book_id}

def test_protected_me():
    app.dependency_overrides[current_user] = lambda: {"id": 7}
    try:
        with TestClient(app) as client:  # runs lifespan
            response = client.get("/me")
        assert response.status_code == 200
        assert response.json()["id"] == 7
    finally:
        app.dependency_overrides.clear()

def test_validation():
    with TestClient(app) as client:
        response = client.get("/books/not-an-int")
    assert response.status_code == 422

# Test database for the app's existing get_db dependency.
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from sqlalchemy.pool import StaticPool
from app.main import app
from app.db import Base, get_db
from app import models  # register mappings before create_all

@pytest.fixture
def client():
    test_engine = create_engine("sqlite://",
        connect_args={"check_same_thread": False}, poolclass=StaticPool)
    Base.metadata.create_all(test_engine)
    connection = test_engine.connect()
    transaction = connection.begin()
    session = Session(bind=connection, join_transaction_mode="create_savepoint")

    def override_db():
        yield session

    app.dependency_overrides[get_db] = override_db
    try:
        with TestClient(app) as client:
            yield client
    finally:
        app.dependency_overrides.clear()
        session.close()
        transaction.rollback()
        connection.close()
        test_engine.dispose()

# SQLite is handy for unit tests; PostgreSQL test DB is better
# for PostgreSQL-specific constraints, locking, and SQL.

# pip install pytest anyio httpx asgi-lifespan
import pytest
from asgi_lifespan import LifespanManager
from httpx import ASGITransport, AsyncClient
from app.main import app

@pytest.mark.anyio
async def test_async_endpoint():
    async with LifespanManager(app):
        transport = ASGITransport(app=app)
        async with AsyncClient(transport=transport,
                               base_url="http://test") as client:
            response = await client.get("/health")
    assert response.status_code == 200

pytest -q
pytest tests/test_api.py -q
pytest -q -k 'authorization or relationship'

# TestClient runs startup/shutdown inside `with TestClient(app)`.
# HTTPX AsyncClient with ASGITransport alone does not.

from time import perf_counter
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://dashboard.example.com"],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PATCH", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
)

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

from typing import Annotated
from fastapi import Depends, FastAPI
from config import Settings, get_settings

app = FastAPI()

@app.get("/info")
def info(settings: Annotated[Settings, Depends(get_settings)]):
    return {"debug": settings.debug}

# Tests may override get_settings via app.dependency_overrides.
# Avoid returning secrets from a settings endpoint.

import hashlib
from fastapi import FastAPI, File, Form, HTTPException, UploadFile

app = FastAPI()

@app.post("/documents")
async def upload_document(
    title: str = Form(),
    file: UploadFile = File(),
):
    allowed = {"application/pdf"}
    if file.content_type not in allowed:
        raise HTTPException(415, "PDF required")
    size = 0
    digest = hashlib.sha256()
    try:
        while chunk := await file.read(1024 * 1024):
            size += len(chunk)
            if size > 10 * 1024 * 1024:
                raise HTTPException(413, "Too large")
            digest.update(chunk)  # process each chunk without buffering the file
        return {"title": title, "filename": file.filename,
                "size": size, "sha256": digest.hexdigest()}
    finally:
        await file.close()

from fastapi.responses import FileResponse, StreamingResponse

@app.get("/reports/latest")
def report():
    return FileResponse("/private/reports/latest.pdf",
                        media_type="application/pdf",
                        filename="latest.pdf")

def rows():
    yield "id,name\n"
    for i in range(3):
        yield f"{i},book-{i}\n"

@app.get("/export")
def export():
    return StreamingResponse(rows(), media_type="text/csv")