diff --git a/CHANGELOG.md b/CHANGELOG.md index db8ac20..74eba64 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,49 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.2.0] - 2026-01-10 + +### Added + +#### Builder Pattern API +- `FastAPIBootstrap` - Fluent builder pattern for intuitive app configuration +- `bootstrap()` - Convenience function to create builder instance +- Chainable methods: `.title()`, `.version()`, `.stage()`, `.prefix()`, `.with_*()`, `.add_router()`, `.build()` + +#### Configuration Management +- `BootstrapSettings` - Centralized Pydantic-based configuration class +- `LoggingSettings`, `CORSSettings`, `SecuritySettings` - Typed configuration models +- `get_settings()` - Cached settings loader from environment variables +- `mask_sensitive_data()` - Utility for masking sensitive fields in logs + +#### Security Middleware +- `SecurityHeadersMiddleware` - HSTS, CSP, X-Frame-Options, X-Content-Type-Options +- `RequestIDMiddleware` - Automatic request ID generation and propagation +- `RequestTimingMiddleware` - Response time tracking (X-Response-Time header) +- `MaxRequestSizeMiddleware` - Request body size limiting + +#### Prometheus Metrics +- `MetricsMiddleware` - Automatic request metrics collection +- `get_metrics_router()` - Prometheus-compatible `/metrics` endpoint +- `MetricsRegistry` - Counter, Gauge, Histogram implementations +- Request count, latency histogram, in-progress requests, error tracking + +#### New Examples +- `examples/builder/` - Builder pattern usage example + +### Changed +- Updated minimum dependency versions for better compatibility +- Status upgraded from Alpha to Beta +- Enhanced `__init__.py` exports for easier imports + +### Fixed +- Various type hints improvements + +## [0.1.2] - 2026-01-05 + +### Fixed +- Minor bug fixes and improvements + ## [0.1.1] - 2026-01-02 ### Changed diff --git a/README.ko.md b/README.ko.md index 30b696d..9bb19c7 100644 --- a/README.ko.md +++ b/README.ko.md @@ -10,7 +10,7 @@ [![Python Version](https://img.shields.io/badge/python-3.12%2B-blue)](https://www.python.org/downloads/) [![License: MIT](https://img.shields.io/badge/License-MIT-green.svg)](https://opensource.org/licenses/MIT) -[![Status](https://img.shields.io/badge/status-alpha-yellow)](https://github.com/bestend/fastapi_bootstrap) +[![Status](https://img.shields.io/badge/status-beta-brightgreen)](https://github.com/bestend/fastapi_bootstrap) [![Tests](https://github.com/bestend/fastapi_bootstrap/actions/workflows/tests.yml/badge.svg)](https://github.com/bestend/fastapi_bootstrap/actions/workflows/tests.yml) @@ -19,7 +19,7 @@ ## ✨ 개요 -**FastAPI Bootstrap**은 강력한 API를 빠르게 구축하는 데 필요한 모든 것을 포함하는 프로덕션 준비 FastAPI 보일러플레이트입니다. 사전 구성된 로깅, 에러 핸들링, 요청/응답 추적 등을 즉시 사용할 수 있습니다. +**FastAPI Bootstrap**은 강력한 API를 빠르게 구축하는 데 필요한 모든 것을 포함하는 프로덕션 준비 FastAPI 보일러플레이트입니다. 사전 구성된 로깅, 에러 핸들링, 요청/응답 추적, 메트릭, 보안 헤더 등을 즉시 사용할 수 있습니다. 매 FastAPI 프로젝트마다 같은 보일러플레이트 코드를 작성하는 것을 멈추세요. FastAPI Bootstrap으로 바로 기능 개발을 시작하세요. @@ -36,6 +36,9 @@ - **📚 자동 문서화** — 자동 OpenAPI/Swagger UI 생성 - **🔧 높은 설정성** — 로깅, CORS, 미들웨어 등을 커스터마이징 가능 - **🚀 프로덕션 준비** — Graceful shutdown, 환경 기반 설정 +- **📊 Prometheus 메트릭** — 요청 통계가 포함된 내장 메트릭 엔드포인트 *(NEW)* +- **🔒 보안 헤더** — HSTS, CSP, X-Frame-Options 미들웨어 *(NEW)* +- **🏗️ Builder 패턴** — 직관적인 앱 설정을 위한 Fluent API *(NEW)* --- @@ -43,6 +46,12 @@ ```bash pip install fastapi_bootstrap + +# 인증 지원 포함 +pip install fastapi_bootstrap[auth] + +# 모든 선택적 의존성 포함 +pip install fastapi_bootstrap[all] ``` --- @@ -51,30 +60,18 @@ pip install fastapi_bootstrap 완전한 예제는 [examples/](./examples/) 디렉토리를 참조하세요. -### 간단한 예제 - -```bash -# 예제 실행 -python examples/simple/app.py - -# 접속 -http://localhost:8000/v1/docs -``` - -### 기본 사용법 +### 기본 사용법 (전통적 방식) ```python from fastapi import APIRouter from fastapi_bootstrap import create_app, LoggingAPIRoute -# API 라우터 생성 router = APIRouter(route_class=LoggingAPIRoute) @router.get("/hello") async def hello(): return {"message": "안녕하세요!"} -# 최소 설정으로 앱 생성 app = create_app( [router], title="내 API", @@ -82,13 +79,105 @@ app = create_app( ) ``` +### 기본 사용법 (Builder 패턴) ✨ NEW + +```python +from fastapi import APIRouter +from fastapi_bootstrap import bootstrap, LoggingAPIRoute + +router = APIRouter(route_class=LoggingAPIRoute) + +@router.get("/hello") +async def hello(): + return {"message": "안녕하세요!"} + +app = ( + bootstrap() + .title("내 API") + .version("1.0.0") + .stage("prod") + .with_cors(origins=["https://myapp.com"]) + .with_security_headers() + .with_metrics() + .with_request_id() + .add_router(router) + .build() +) +``` + ### 앱 실행 ```bash uvicorn main:app --host 0.0.0.0 --port 8000 ``` -### 전체 설정 예제 +--- + +## 🏗️ Builder 패턴 *(NEW)* + +Fluent API로 직관적인 앱 설정: + +```python +from fastapi_bootstrap import bootstrap, FastAPIBootstrap + +# Fluent 체이닝 +app = ( + bootstrap() + .title("내 API") + .version("1.0.0") + .stage("prod") + .with_cors(origins=["https://myapp.com"]) + .with_security_headers() + .with_metrics(endpoint="/metrics") + .with_request_id() + .with_request_timing() + .add_router(router) + .build() +) + +# 또는 환경 설정에서 구성 +from fastapi_bootstrap.config import BootstrapSettings +settings = BootstrapSettings() # 환경변수에서 자동 로드 +``` + +--- + +## 📊 Prometheus 메트릭 *(NEW)* + +```python +from fastapi_bootstrap import bootstrap + +app = bootstrap().title("내 API").with_metrics().build() + +# GET /metrics 에서 확인: +# http_requests_total{method="GET", path="/hello", status="200"} 42 +# http_request_duration_seconds_bucket{le="0.1"} 35 +``` + +--- + +## 🔒 보안 헤더 *(NEW)* + +```python +app = ( + bootstrap() + .with_security_headers( + hsts_max_age=31536000, # 1년 + content_security_policy="default-src 'self'", + x_frame_options="DENY", + ) + .build() +) + +# 응답 헤더: +# Strict-Transport-Security: max-age=31536000; includeSubDomains +# X-Content-Type-Options: nosniff +# X-Frame-Options: DENY +``` + +--- + +## ⚙️ 전체 설정 예제 ```python from fastapi import APIRouter @@ -337,7 +426,16 @@ python examples/simple/app.py # http://localhost:8000/v1/docs 접속 ``` -### 2. [Auth Example](./examples/auth/) +### 2. [Builder Example](./examples/builder/) *(NEW)* +메트릭, 보안 헤더, request ID를 포함한 Fluent Builder API. + +```bash +python examples/builder/app.py +# http://localhost:8000/docs 접속 +# http://localhost:8000/metrics 에서 메트릭 확인 +``` + +### 3. [Auth Example](./examples/auth/) 역할 기반 접근 제어를 포함한 OIDC/Keycloak 인증. ```bash @@ -349,7 +447,7 @@ python examples/auth/app.py # http://localhost:8000/v1/docs 접속 ``` -### 3. [CORS Example](./examples/cors/) +### 4. [CORS Example](./examples/cors/) 환경별 CORS 설정 및 보안 모범 사례. ```bash @@ -360,7 +458,7 @@ python examples/cors/app.py STAGE=prod ALLOWED_ORIGINS="https://myapp.com" python examples/cors/app.py ``` -### 4. [External Auth Example](./examples/external_auth/) +### 5. [External Auth Example](./examples/external_auth/) API Gateway/Ingress 인증 및 Swagger UI Bearer token 지원. ```bash diff --git a/README.md b/README.md index 401974e..8388084 100644 --- a/README.md +++ b/README.md @@ -10,7 +10,7 @@ [![Python Version](https://img.shields.io/badge/python-3.12%2B-blue)](https://www.python.org/downloads/) [![License: MIT](https://img.shields.io/badge/License-MIT-green.svg)](https://opensource.org/licenses/MIT) -[![Status](https://img.shields.io/badge/status-alpha-yellow)](https://github.com/bestend/fastapi_bootstrap) +[![Status](https://img.shields.io/badge/status-beta-brightgreen)](https://github.com/bestend/fastapi_bootstrap) [![Tests](https://github.com/bestend/fastapi_bootstrap/actions/workflows/tests.yml/badge.svg)](https://github.com/bestend/fastapi_bootstrap/actions/workflows/tests.yml) @@ -19,7 +19,7 @@ ## ✨ Overview -**FastAPI Bootstrap** is a production-ready FastAPI boilerplate that includes everything you need to build robust APIs quickly. It provides pre-configured logging, error handling, request/response tracking, and more out of the box. +**FastAPI Bootstrap** is a production-ready FastAPI boilerplate that includes everything you need to build robust APIs quickly. It provides pre-configured logging, error handling, request/response tracking, metrics, security headers, and more out of the box. Stop writing the same boilerplate code for every FastAPI project. Start building features immediately with FastAPI Bootstrap. @@ -36,13 +36,22 @@ Stop writing the same boilerplate code for every FastAPI project. Start building - **📚 Auto Documentation** — Automatic OpenAPI/Swagger UI generation - **🔧 Highly Configurable** — Customize logging, CORS, middleware, and more - **🚀 Production Ready** — Graceful shutdown, environment-based configuration +- **📊 Prometheus Metrics** — Built-in metrics endpoint with request stats *(NEW)* +- **🔒 Security Headers** — HSTS, CSP, X-Frame-Options middleware *(NEW)* +- **🏗️ Builder Pattern** — Fluent API for intuitive app configuration *(NEW)* --- ## 📦 Installation ```bash -pip install fastapi_bootstrap +pip install fastapi-bootstrap + +# With authentication support +pip install fastapi-bootstrap[auth] + +# With all optional dependencies +pip install fastapi-bootstrap[all] ``` --- @@ -51,31 +60,18 @@ pip install fastapi_bootstrap See [examples/](./examples/) directory for complete, runnable examples. -### Simple Example - -```bash -# Run the example -python examples/simple/app.py - -# Visit -http://localhost:8000/v1/docs -``` - -### Basic Usage - +### Basic Usage (Traditional) ```python from fastapi import APIRouter from fastapi_bootstrap import create_app, LoggingAPIRoute -# Create your API router router = APIRouter(route_class=LoggingAPIRoute) @router.get("/hello") async def hello(): return {"message": "Hello, World!"} -# Create the app with minimal configuration app = create_app( [router], title="My API", @@ -83,6 +79,32 @@ app = create_app( ) ``` +### Basic Usage (Builder Pattern) ✨ NEW + +```python +from fastapi import APIRouter +from fastapi_bootstrap import bootstrap, LoggingAPIRoute + +router = APIRouter(route_class=LoggingAPIRoute) + +@router.get("/hello") +async def hello(): + return {"message": "Hello, World!"} + +app = ( + bootstrap() + .title("My API") + .version("1.0.0") + .stage("prod") + .with_cors(origins=["https://myapp.com"]) + .with_security_headers() + .with_metrics() + .with_request_id() + .add_router(router) + .build() +) +``` + ### Run the app ```bash @@ -92,7 +114,7 @@ uvicorn main:app --host 0.0.0.0 --port 8000 ### With Full Configuration ```python -from fastapi import APIRouter, Depends +from fastapi import APIRouter from fastapi_bootstrap import create_app, LoggingAPIRoute, get_logger logger = get_logger() @@ -106,18 +128,9 @@ async def hello(): async def startup_handler(app): logger.info("Application starting up...") - # Initialize database, connections, etc. async def shutdown_handler(app): logger.info("Application shutting down...") - # Cleanup resources - -# Or without app parameter (both styles are supported) -async def simple_startup(): - logger.info("Simple startup...") - -async def simple_shutdown(): - logger.info("Simple shutdown...") app = create_app( api_list=[router], @@ -126,16 +139,129 @@ app = create_app( prefix_url="/api/v1", graceful_timeout=10, docs_enable=True, - docs_prefix_url="/api/v1", health_check_api="/healthz", startup_coroutines=[startup_handler], shutdown_coroutines=[shutdown_handler], - stage="prod", # dev, staging, prod + stage="prod", ) ``` --- +## 🏗️ Builder Pattern (NEW in v0.2.0) + +The new Builder pattern provides a fluent, chainable API for configuring your FastAPI application: + +```python +from fastapi_bootstrap import bootstrap + +app = ( + bootstrap() + # Basic configuration + .title("My Production API") + .version("2.0.0") + .description("A production-ready API") + .stage("prod") + .prefix("/api/v2") + + # Logging configuration + .with_logging(level="INFO", json_output=True) + + # CORS configuration + .with_cors( + origins=["https://myapp.com", "https://admin.myapp.com"], + allow_credentials=True, + ) + + # Security headers (HSTS, CSP, X-Frame-Options, etc.) + .with_security_headers( + hsts_max_age=31536000, + content_security_policy="default-src 'self'", + ) + + # Prometheus metrics at /metrics + .with_metrics(endpoint="/metrics") + + # Request ID and timing headers + .with_request_id() + .with_request_timing() + + # Health check endpoint + .with_health_check(endpoint="/health") + + # Graceful shutdown + .with_graceful_shutdown(timeout=30) + + # Add routers + .add_router(users_router, prefix="/users") + .add_router(posts_router, prefix="/posts") + + # Lifecycle hooks + .on_startup(init_database) + .on_shutdown(cleanup_database) + + .build() +) +``` + +--- + +## 📊 Prometheus Metrics (NEW) + +Built-in Prometheus-compatible metrics endpoint: + +```python +from fastapi_bootstrap import bootstrap, MetricsMiddleware, get_metrics_router + +# Using Builder pattern +app = bootstrap().with_metrics().add_router(router).build() + +# Or manually +app.add_middleware(MetricsMiddleware) +app.include_router(get_metrics_router()) +``` + +Visit `/metrics` to see: +``` +# HELP fastapi_bootstrap_http_requests_total Total HTTP requests +# TYPE fastapi_bootstrap_http_requests_total counter +fastapi_bootstrap_http_requests_total{method="GET",path="/api/users",status="200"} 150 + +# HELP fastapi_bootstrap_http_request_duration_seconds HTTP request duration +# TYPE fastapi_bootstrap_http_request_duration_seconds histogram +fastapi_bootstrap_http_request_duration_seconds_bucket{method="GET",path="/api/users",le="0.1"} 145 +... +``` + +--- + +## 🔒 Security Headers (NEW) + +Automatic security headers middleware: + +```python +from fastapi_bootstrap import SecurityHeadersMiddleware + +app.add_middleware( + SecurityHeadersMiddleware, + stage="prod", + hsts_max_age=31536000, + content_security_policy="default-src 'self'", + x_frame_options="DENY", +) +``` + +Added headers: +- `Strict-Transport-Security` (HSTS) +- `Content-Security-Policy` +- `X-Frame-Options` +- `X-Content-Type-Options` +- `X-XSS-Protection` +- `Referrer-Policy` +- `Permissions-Policy` + +--- + ## 📖 Core Components ### 1. `create_app()` @@ -337,22 +463,25 @@ Basic usage with logging, response formatting, and pagination. ```bash python examples/simple/app.py -# Visit http://localhost:8000/v1/docs ``` -### 2. [Auth Example](./examples/auth/) +### 2. [Builder Example](./examples/builder/) ✨ NEW +Builder pattern with metrics, security headers, and request tracking. + +```bash +python examples/builder/app.py +``` + +### 3. [Auth Example](./examples/auth/) OIDC/Keycloak authentication with role-based access control. ```bash -# Set up environment export OIDC_ISSUER="https://keycloak.example.com/realms/myrealm" export OIDC_CLIENT_ID="my-api" - python examples/auth/app.py -# Visit http://localhost:8000/v1/docs ``` -### 3. [CORS Example](./examples/cors/) +### 4. [CORS Example](./examples/cors/) Environment-specific CORS configuration and security best practices. ```bash @@ -363,12 +492,11 @@ python examples/cors/app.py STAGE=prod ALLOWED_ORIGINS="https://myapp.com" python examples/cors/app.py ``` -### 4. [External Auth Example](./examples/external_auth/) +### 5. [External Auth Example](./examples/external_auth/) API Gateway/Ingress authentication with Swagger UI Bearer token support. ```bash python examples/external_auth/app.py -# Visit http://localhost:8000/docs ``` See [examples/README.md](./examples/README.md) for detailed documentation. diff --git a/examples/README.md b/examples/README.md index 4b40ecc..528b58d 100644 --- a/examples/README.md +++ b/examples/README.md @@ -10,6 +10,14 @@ python examples/simple/app.py Logging, standardized responses, pagination +### [Builder](./builder/) - Builder Pattern (NEW!) + +```bash +python examples/builder/app.py +``` + +Fluent builder API, metrics, security headers, request ID/timing + ### [Auth](./auth/) - OIDC Authentication ```bash @@ -40,7 +48,7 @@ API Gateway/Ingress authentication, Swagger UI Bearer token support ## Quick Reference -### Basic App +### Basic App (Traditional) ```python from fastapi_bootstrap import create_app @@ -48,6 +56,24 @@ from fastapi_bootstrap import create_app app = create_app([router], title="My API", prefix_url="/v1") ``` +### Basic App (Builder Pattern) ✨ + +```python +from fastapi_bootstrap import bootstrap + +app = ( + bootstrap() + .title("My API") + .version("1.0.0") + .stage("prod") + .with_cors(origins=["https://myapp.com"]) + .with_security_headers() + .with_metrics() + .add_router(router) + .build() +) +``` + ### Auto Logging ```python @@ -78,6 +104,55 @@ async def get_me(user = Depends(auth.get_current_user)): return {"email": user.email} ``` +### Prometheus Metrics + +```python +from fastapi_bootstrap import get_metrics_router, MetricsMiddleware + +app.add_middleware(MetricsMiddleware) +app.include_router(get_metrics_router()) +# Metrics available at GET /metrics +``` + +### Security Headers + +```python +from fastapi_bootstrap import SecurityHeadersMiddleware + +app.add_middleware( + SecurityHeadersMiddleware, + stage="prod", + hsts_max_age=31536000, + content_security_policy="default-src 'self'", +) +``` + +### Configuration + +```python +from fastapi_bootstrap import BootstrapSettings, get_settings + +# Load from environment variables +settings = get_settings() + +# Or configure programmatically +settings = BootstrapSettings( + stage="prod", + logging=LoggingSettings(level="WARNING"), + cors=CORSSettings(origins=["https://myapp.com"]), +) +``` + +### Sensitive Data Masking + +```python +from fastapi_bootstrap import mask_sensitive_data + +data = {"username": "john", "password": "secret123"} +masked = mask_sensitive_data(data) +# {"username": "john", "password": "***MASKED***"} +``` + ### CORS ```python diff --git a/examples/builder/README.md b/examples/builder/README.md new file mode 100644 index 0000000..7ca4d8c --- /dev/null +++ b/examples/builder/README.md @@ -0,0 +1,84 @@ +# Builder Pattern Example + +This example demonstrates the new fluent Builder API for creating FastAPI applications. + +## Features Demonstrated + +- **Builder Pattern**: Chainable, intuitive API for app configuration +- **Automatic Metrics**: Prometheus-compatible `/metrics` endpoint +- **Security Headers**: Auto-configured for production +- **Request ID**: Every request gets a unique ID in `X-Request-ID` header +- **Request Timing**: Response time in `X-Response-Time` header +- **Health Check**: Configurable health endpoint + +## Running the Example + +```bash +# Install dependencies +pip install fastapi-bootstrap[all] + +# Run the server +cd examples/builder +uvicorn app:app --reload +``` + +## Endpoints + +- `GET /` - Root endpoint +- `GET /users` - List users +- `GET /users/{id}` - Get specific user +- `GET /health` - Health check +- `GET /metrics` - Prometheus metrics +- `GET /docs` - Swagger UI + +## Builder Pattern vs Traditional + +### Traditional (still supported) +```python +app = create_app( + api_list=[router], + title="My API", + version="1.0.0", + stage="prod", + cors_origins=["https://myapp.com"], +) +``` + +### Builder Pattern (new) +```python +app = ( + bootstrap() + .title("My API") + .version("1.0.0") + .stage("prod") + .with_cors(origins=["https://myapp.com"]) + .with_security_headers() + .with_metrics() + .add_router(router) + .build() +) +``` + +## Response Headers + +When making requests, you'll see these headers in responses: + +``` +X-Request-ID: 550e8400-e29b-41d4-a716-446655440000 +X-Response-Time: 1.23ms +``` + +## Metrics Output + +Visit `/metrics` to see Prometheus-format metrics: + +``` +# HELP fastapi_bootstrap_http_requests_total Total HTTP requests +# TYPE fastapi_bootstrap_http_requests_total counter +fastapi_bootstrap_http_requests_total{method="GET",path="/",status="200"} 5 + +# HELP fastapi_bootstrap_http_request_duration_seconds HTTP request duration +# TYPE fastapi_bootstrap_http_request_duration_seconds histogram +fastapi_bootstrap_http_request_duration_seconds_bucket{method="GET",path="/",le="0.005"} 3 +... +``` diff --git a/examples/builder/app.py b/examples/builder/app.py new file mode 100644 index 0000000..dcae4d0 --- /dev/null +++ b/examples/builder/app.py @@ -0,0 +1,69 @@ +"""Example application using the Builder pattern. + +This example demonstrates the new fluent Builder API for creating +FastAPI applications with fastapi_bootstrap. + +Run: + uvicorn app:app --reload +""" + +from fastapi import APIRouter + +from fastapi_bootstrap import LoggingAPIRoute, bootstrap + +# Create router with automatic request/response logging +router = APIRouter(route_class=LoggingAPIRoute) + + +@router.get("/") +async def root(): + """Root endpoint.""" + return {"message": "Hello from Builder Pattern!"} + + +@router.get("/users") +async def list_users(): + """List all users.""" + return { + "users": [ + {"id": 1, "name": "Alice"}, + {"id": 2, "name": "Bob"}, + ] + } + + +@router.get("/users/{user_id}") +async def get_user(user_id: int): + """Get a specific user.""" + return {"id": user_id, "name": "User"} + + +# Build the application using the fluent builder pattern +app = ( + bootstrap() + .title("Builder Pattern Example") + .version("1.0.0") + .stage("dev") # Use "prod" for production + # Configure logging + .with_logging(level="DEBUG", json_output=False) + # Enable CORS (auto-configured for dev stage) + .with_cors() + # Add security headers (skipped in dev by default) + .with_security_headers() + # Enable Prometheus metrics + .with_metrics(endpoint="/metrics") + # Add request ID and timing headers + .with_request_id() + .with_request_timing() + # Configure health check + .with_health_check(endpoint="/health") + # Add the router + .add_router(router) + .build() +) + + +if __name__ == "__main__": + import uvicorn + + uvicorn.run(app, host="0.0.0.0", port=8000) diff --git a/pyproject.toml b/pyproject.toml index ce0418d..7995c2f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -3,18 +3,21 @@ requires = ["setuptools>=65.0", "setuptools-scm[toml]>=6.2"] build-backend = "setuptools.build_meta" [project] -name = "fastapi_bootstrap" +# Distribution name: fastapi-bootstrap (PyPI convention with hyphen) +# Python import name: fastapi_bootstrap (underscore) +# pip install fastapi-bootstrap +name = "fastapi-bootstrap" version = "0.0.dev" # Auto-updated by GitHub Actions when tag is pushed -description = "🚀 Easy FastAPI Setup Kit - Production-ready FastAPI boilerplate with logging, error handling, and more" +description = "🚀 Easy FastAPI Setup Kit - Production-ready FastAPI boilerplate with logging, error handling, metrics, security headers, and more" readme = "README.md" requires-python = ">=3.12" license = {text = "MIT"} authors = [ {name = "JunSeok Kim", email = "infend@gmail.com"} ] -keywords = ["fastapi", "logging", "error-handling", "api", "web", "framework", "production"] +keywords = ["fastapi", "logging", "error-handling", "api", "web", "framework", "production", "metrics", "security", "opentelemetry"] classifiers = [ - "Development Status :: 3 - Alpha", + "Development Status :: 4 - Beta", "Intended Audience :: Developers", "License :: OSI Approved :: MIT License", "Programming Language :: Python :: 3", @@ -22,30 +25,39 @@ classifiers = [ "Programming Language :: Python :: 3.13", "Framework :: FastAPI", "Topic :: Software Development :: Libraries :: Python Modules", + "Topic :: Internet :: WWW/HTTP :: HTTP Servers", + "Typing :: Typed", ] dependencies = [ - "fastapi>=0.100.0", - "uvicorn>=0.20.0", - "pydantic>=2.0.0", - "loguru>=0.7.0", - "opentelemetry-api>=1.20.0", - "PyYAML>=6.0", + "fastapi>=0.109.0", + "uvicorn>=0.27.0", + "pydantic>=2.5.0", + "loguru>=0.7.2", + "opentelemetry-api>=1.22.0", + "PyYAML>=6.0.1", ] [project.optional-dependencies] auth = [ "python-jose[cryptography]>=3.3.0", - "httpx>=0.24.0", + "httpx>=0.26.0", +] +metrics = [ + # Future: prometheus-client if needed +] +all = [ + "python-jose[cryptography]>=3.3.0", + "httpx>=0.26.0", ] dev = [ - "pytest>=7.0.0", - "pytest-cov>=4.0.0", - "pytest-asyncio>=0.21.0", - "pytest-mock>=3.10.0", - "httpx>=0.24.0", - "ruff>=0.1.0", - "mypy>=1.0.0", - "types-PyYAML>=6.0.0", + "pytest>=8.0.0", + "pytest-cov>=4.1.0", + "pytest-asyncio>=0.23.0", + "pytest-mock>=3.12.0", + "httpx>=0.26.0", + "ruff>=0.2.0", + "mypy>=1.8.0", + "types-PyYAML>=6.0.12", "python-jose[cryptography]>=3.3.0", ] diff --git a/src/fastapi_bootstrap/__init__.py b/src/fastapi_bootstrap/__init__.py index 7a69293..cd88734 100644 --- a/src/fastapi_bootstrap/__init__.py +++ b/src/fastapi_bootstrap/__init__.py @@ -3,9 +3,9 @@ 🚀 Production-ready FastAPI boilerplate with batteries included. This package provides a complete FastAPI setup with logging, error handling, -request/response tracking, and more out of the box. +request/response tracking, metrics, security headers, and more out of the box. -**Quick Start:** +**Quick Start (Traditional):** ```python from fastapi import APIRouter from fastapi_bootstrap import create_app, LoggingAPIRoute @@ -19,24 +19,59 @@ async def hello(): app = create_app([router], title="My API", version="1.0.0") ``` -**Recommended import style:** +**Quick Start (Builder Pattern):** ```python -from fastapi_bootstrap import create_app, LoggingAPIRoute, BaseModel -from fastapi_bootstrap.log import get_logger -from fastapi_bootstrap.exception import add_exception_handler +from fastapi import APIRouter +from fastapi_bootstrap import bootstrap, LoggingAPIRoute + +router = APIRouter(route_class=LoggingAPIRoute) + +@router.get("/hello") +async def hello(): + return {"message": "Hello, World!"} + +app = ( + bootstrap() + .title("My API") + .version("1.0.0") + .stage("prod") + .with_cors(origins=["https://myapp.com"]) + .with_security_headers() + .with_metrics() + .add_router(router) + .build() +) ``` -**Avoid importing from internal modules:** +**Recommended import style:** ```python -# ❌ Don't do this -from fastapi_bootstrap.base import create_app -from fastapi_bootstrap.exception.handler import add_exception_handler +from fastapi_bootstrap import create_app, bootstrap, LoggingAPIRoute, BaseModel +from fastapi_bootstrap.log import get_logger +from fastapi_bootstrap.config import BootstrapSettings +from fastapi_bootstrap.metrics import get_metrics_router ``` """ from .base import create_app +from .builder import FastAPIBootstrap, bootstrap +from .config import ( + BootstrapSettings, + CORSSettings, + LoggingSettings, + SecuritySettings, + Stage, + get_settings, + mask_sensitive_data, +) from .log import get_logger from .logging_api_route import LoggingAPIRoute +from .metrics import MetricsMiddleware, get_metrics_registry, get_metrics_router +from .middleware import ( + MaxRequestSizeMiddleware, + RequestIDMiddleware, + RequestTimingMiddleware, + SecurityHeadersMiddleware, +) from .response import ResponseFormatter from .type import BaseModel @@ -45,23 +80,66 @@ async def hello(): from .auth import OIDCAuth, OIDCConfig, TokenPayload __all__ = [ + # Core "BaseModel", "LoggingAPIRoute", - "OIDCAuth", - "OIDCConfig", "ResponseFormatter", - "TokenPayload", "create_app", "get_logger", + # Builder + "FastAPIBootstrap", + "bootstrap", + # Config + "BootstrapSettings", + "CORSSettings", + "LoggingSettings", + "SecuritySettings", + "Stage", + "get_settings", + "mask_sensitive_data", + # Metrics + "MetricsMiddleware", + "get_metrics_registry", + "get_metrics_router", + # Middleware + "MaxRequestSizeMiddleware", + "RequestIDMiddleware", + "RequestTimingMiddleware", + "SecurityHeadersMiddleware", + # Auth (optional) + "OIDCAuth", + "OIDCConfig", + "TokenPayload", ] except ImportError: # Auth dependencies not installed __all__ = [ + # Core "BaseModel", "LoggingAPIRoute", "ResponseFormatter", "create_app", "get_logger", + # Builder + "FastAPIBootstrap", + "bootstrap", + # Config + "BootstrapSettings", + "CORSSettings", + "LoggingSettings", + "SecuritySettings", + "Stage", + "get_settings", + "mask_sensitive_data", + # Metrics + "MetricsMiddleware", + "get_metrics_registry", + "get_metrics_router", + # Middleware + "MaxRequestSizeMiddleware", + "RequestIDMiddleware", + "RequestTimingMiddleware", + "SecurityHeadersMiddleware", ] -__version__ = "0.0.dev" +__version__ = "0.2.0" diff --git a/src/fastapi_bootstrap/builder.py b/src/fastapi_bootstrap/builder.py new file mode 100644 index 0000000..6c71284 --- /dev/null +++ b/src/fastapi_bootstrap/builder.py @@ -0,0 +1,586 @@ +"""Builder pattern API for FastAPI Bootstrap. + +This module provides a fluent builder interface for creating FastAPI applications +with a more intuitive and chainable API. + +Example: + ```python + from fastapi_bootstrap.builder import FastAPIBootstrap + + app = ( + FastAPIBootstrap() + .title("My API") + .version("1.0.0") + .stage("prod") + .with_logging(level="INFO", json_output=True) + .with_cors(origins=["https://myapp.com"]) + .with_security_headers() + .with_metrics() + .with_auth(oidc_config) + .add_router(router) + .build() + ) + ``` +""" + +from collections.abc import Callable +from typing import Any, Self + +from fastapi import APIRouter, FastAPI + +from fastapi_bootstrap.base import create_app +from fastapi_bootstrap.config import ( + BootstrapSettings, + CORSSettings, + GracefulShutdownSettings, + HealthCheckSettings, + LogFormat, + LoggingSettings, + MetricsSettings, + RateLimitSettings, + Stage, +) + + +class FastAPIBootstrap: + """Fluent builder for creating FastAPI applications. + + This class provides a chainable API for configuring and creating + production-ready FastAPI applications with all the features of + fastapi_bootstrap. + + Example: + ```python + # Simple usage + app = ( + FastAPIBootstrap() + .title("My API") + .version("1.0.0") + .add_router(router) + .build() + ) + + # Full configuration + app = ( + FastAPIBootstrap() + .title("Production API") + .version("2.0.0") + .description("A production-ready API") + .stage("prod") + .prefix("/api/v2") + .with_logging(level="WARNING", json_output=True) + .with_cors( + origins=["https://myapp.com"], + allow_credentials=True, + ) + .with_security_headers( + hsts_max_age=31536000, + content_security_policy="default-src 'self'", + ) + .with_metrics(endpoint="/metrics") + .with_health_check(endpoint="/health") + .with_graceful_shutdown(timeout=30) + .add_router(users_router, prefix="/users") + .add_router(posts_router, prefix="/posts") + .on_startup(init_database) + .on_shutdown(cleanup_database) + .build() + ) + ``` + """ + + def __init__(self): + """Initialize the builder with default settings.""" + self._settings = BootstrapSettings() + self._routers: list[tuple[APIRouter, str | None, list[Any] | None]] = [] + self._middlewares: list[Any] = [] + self._startup_coroutines: list[Callable] = [] + self._shutdown_coroutines: list[Callable] = [] + self._dependencies: list[Any] = [] + self._prefix_url: str = "" + self._docs_enable: bool = True + self._docs_prefix_url: str = "" + self._add_bearer_auth: bool = False + self._swagger_ui_init_oauth: dict[str, Any] | None = None + self._oidc_config: Any = None + self._enable_security_headers: bool = False + self._security_headers_config: dict[str, Any] = {} + self._enable_metrics: bool = False + self._enable_request_id: bool = False + self._request_id_header: str = "X-Request-ID" + self._enable_request_timing: bool = False + self._request_timing_header: str = "X-Response-Time" + + def title(self, title: str) -> Self: + """Set the API title. + + Args: + title: API title for documentation + + Returns: + Self for chaining + """ + self._settings.title = title + return self + + def version(self, version: str) -> Self: + """Set the API version. + + Args: + version: API version string + + Returns: + Self for chaining + """ + self._settings.version = version + return self + + def description(self, description: str) -> Self: + """Set the API description. + + Args: + description: API description for documentation + + Returns: + Self for chaining + """ + self._settings.description = description + return self + + def stage(self, stage: str | Stage) -> Self: + """Set the application stage. + + Args: + stage: Environment stage ("dev", "staging", or "prod") + + Returns: + Self for chaining + """ + if isinstance(stage, str): + stage = Stage(stage.lower()) + self._settings.stage = stage + return self + + def prefix(self, prefix: str) -> Self: + """Set the URL prefix for all routes. + + Args: + prefix: URL prefix (e.g., "/api/v1") + + Returns: + Self for chaining + """ + self._prefix_url = prefix + return self + + def with_logging( + self, + level: str = "INFO", + json_output: bool = False, + max_string_length: int = 5000, + truncation_threshold: int = 2000, + ) -> Self: + """Configure logging settings. + + Args: + level: Log level (DEBUG, INFO, WARNING, ERROR, CRITICAL) + json_output: Enable JSON log output + max_string_length: Maximum length for logged strings + truncation_threshold: Threshold for truncating strings in structures + + Returns: + Self for chaining + """ + self._settings.logging = LoggingSettings( + level=level, + format=LogFormat.JSON if json_output else LogFormat.TEXT, + json_output=json_output, + string_max_length=max_string_length, + truncation_threshold=truncation_threshold, + ) + return self + + def with_cors( + self, + origins: list[str] | None = None, + allow_credentials: bool = True, + allow_methods: list[str] | None = None, + allow_headers: list[str] | None = None, + expose_headers: list[str] | None = None, + max_age: int = 600, + ) -> Self: + """Configure CORS settings. + + Args: + origins: Allowed origins (None for stage-based defaults) + allow_credentials: Allow credentials in CORS requests + allow_methods: Allowed HTTP methods + allow_headers: Allowed HTTP headers + expose_headers: Headers to expose to browser + max_age: Max age for preflight cache + + Returns: + Self for chaining + """ + self._settings.cors = CORSSettings( + origins=origins or [], + allow_credentials=allow_credentials, + allow_methods=allow_methods or ["GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS"], + allow_headers=allow_headers or ["*"], + expose_headers=expose_headers or ["X-Request-ID", "X-Trace-ID"], + max_age=max_age, + ) + return self + + def with_security_headers( + self, + hsts_max_age: int = 31536000, + hsts_include_subdomains: bool = True, + content_security_policy: str | None = "default-src 'self'", + x_frame_options: str = "DENY", + enable_in_dev: bool = False, + ) -> Self: + """Enable security headers middleware. + + Args: + hsts_max_age: HSTS max-age in seconds + hsts_include_subdomains: Include subdomains in HSTS + content_security_policy: CSP header value + x_frame_options: X-Frame-Options value + enable_in_dev: Enable in development mode + + Returns: + Self for chaining + """ + self._enable_security_headers = True + self._security_headers_config = { + "hsts_max_age": hsts_max_age, + "hsts_include_subdomains": hsts_include_subdomains, + "content_security_policy": content_security_policy, + "x_frame_options": x_frame_options, + "enable_in_dev": enable_in_dev, + } + return self + + def with_metrics( + self, + endpoint: str = "/metrics", + include_in_schema: bool = False, + ) -> Self: + """Enable Prometheus metrics. + + Args: + endpoint: Metrics endpoint path + include_in_schema: Include in OpenAPI schema + + Returns: + Self for chaining + """ + self._enable_metrics = True + self._settings.metrics = MetricsSettings( + enabled=True, + endpoint=endpoint, + include_in_schema=include_in_schema, + ) + return self + + def with_health_check( + self, + endpoint: str = "/healthz", + include_in_schema: bool = False, + ) -> Self: + """Configure health check endpoint. + + Args: + endpoint: Health check endpoint path + include_in_schema: Include in OpenAPI schema + + Returns: + Self for chaining + """ + self._settings.health_check = HealthCheckSettings( + endpoint=endpoint, + include_in_schema=include_in_schema, + ) + return self + + def with_graceful_shutdown(self, timeout: int = 10) -> Self: + """Configure graceful shutdown. + + Args: + timeout: Seconds to wait for in-flight requests + + Returns: + Self for chaining + """ + self._settings.graceful_shutdown = GracefulShutdownSettings(timeout=timeout) + return self + + def with_rate_limit( + self, + requests_per_minute: int = 60, + burst_size: int = 10, + ) -> Self: + """Enable rate limiting. + + Args: + requests_per_minute: Max requests per minute per client + burst_size: Allowed burst above limit + + Returns: + Self for chaining + """ + self._settings.rate_limit = RateLimitSettings( + enabled=True, + requests_per_minute=requests_per_minute, + burst_size=burst_size, + ) + return self + + def with_request_id(self, header_name: str = "X-Request-ID") -> Self: + """Enable request ID middleware. + + Args: + header_name: Header name for request ID + + Returns: + Self for chaining + """ + self._enable_request_id = True + self._request_id_header = header_name + return self + + def with_request_timing(self, header_name: str = "X-Response-Time") -> Self: + """Enable request timing middleware. + + Args: + header_name: Header name for response time + + Returns: + Self for chaining + """ + self._enable_request_timing = True + self._request_timing_header = header_name + return self + + def with_auth(self, oidc_config: Any, enable_swagger_ui: bool = True) -> Self: + """Configure OIDC authentication. + + Args: + oidc_config: OIDCConfig instance + enable_swagger_ui: Enable OAuth2 flow in Swagger UI + + Returns: + Self for chaining + """ + self._oidc_config = oidc_config + self._add_bearer_auth = True + if enable_swagger_ui and oidc_config: + self._swagger_ui_init_oauth = { + "clientId": oidc_config.client_id, + "usePkceWithAuthorizationCodeGrant": True, + } + return self + + def with_docs( + self, + enable: bool = True, + prefix_url: str = "", + ) -> Self: + """Configure API documentation. + + Args: + enable: Enable docs endpoints + prefix_url: URL prefix for docs + + Returns: + Self for chaining + """ + self._docs_enable = enable + self._docs_prefix_url = prefix_url + return self + + def add_router( + self, + router: APIRouter, + prefix: str = "", + dependencies: list[Any] | None = None, + ) -> Self: + """Add an API router. + + Args: + router: FastAPI APIRouter instance + prefix: Optional prefix for this router + dependencies: Optional dependencies for this router + + Returns: + Self for chaining + + Note: + When prefix or dependencies are specified, routes are copied to a + new router. Router-level properties (tags, responses, callbacks) + should be set on individual routes or use FastAPI's include_router + directly for complex configurations. + """ + self._routers.append((router, prefix, dependencies)) + return self + + def add_middleware(self, middleware: Any, **kwargs: Any) -> Self: + """Add a custom middleware. + + Args: + middleware: Middleware class + **kwargs: Middleware configuration + + Returns: + Self for chaining + """ + self._middlewares.append((middleware, kwargs)) + return self + + def add_dependency(self, dependency: Any) -> Self: + """Add a global dependency. + + Args: + dependency: FastAPI dependency + + Returns: + Self for chaining + """ + self._dependencies.append(dependency) + return self + + def on_startup(self, coroutine: Callable) -> Self: + """Add a startup coroutine. + + Args: + coroutine: Async function to run on startup + + Returns: + Self for chaining + """ + self._startup_coroutines.append(coroutine) + return self + + def on_shutdown(self, coroutine: Callable) -> Self: + """Add a shutdown coroutine. + + Args: + coroutine: Async function to run on shutdown + + Returns: + Self for chaining + """ + self._shutdown_coroutines.append(coroutine) + return self + + def build(self) -> FastAPI: + """Build the FastAPI application. + + Returns: + Configured FastAPI application instance + """ + # Prepare routers - combine with individual prefixes and dependencies + api_list = [] + for router, prefix, deps in self._routers: + if prefix or deps: + # Create a new router with the combined prefix and dependencies + combined_router = APIRouter() + combined_router.include_router( + router, + prefix=prefix or "", + dependencies=deps or [], + ) + api_list.append(combined_router) + else: + api_list.append(router) + + # Get CORS config based on stage + cors_config = self._settings.get_cors_config_for_stage() + + # Create base app using create_app + app = create_app( + api_list=api_list, + title=self._settings.title, + version=self._settings.version, + prefix_url=self._prefix_url, + graceful_timeout=self._settings.graceful_shutdown.timeout, + dependencies=self._dependencies if self._dependencies else None, + startup_coroutines=self._startup_coroutines if self._startup_coroutines else None, + shutdown_coroutines=self._shutdown_coroutines if self._shutdown_coroutines else None, + health_check_api=self._settings.health_check.endpoint, + metrics_api=self._settings.metrics.endpoint, + docs_enable=self._docs_enable, + docs_prefix_url=self._docs_prefix_url, + add_external_basic_auth=self._add_bearer_auth, + stage=self._settings.stage.value + if isinstance(self._settings.stage, Stage) + else self._settings.stage, + cors_origins=cors_config.origins if cors_config.origins else None, + cors_allow_credentials=cors_config.allow_credentials, + cors_allow_methods=cors_config.allow_methods if cors_config.allow_methods else None, + cors_allow_headers=cors_config.allow_headers if cors_config.allow_headers else None, + swagger_ui_init_oauth=self._swagger_ui_init_oauth, + ) + + # Add security headers middleware if enabled + if self._enable_security_headers: + from fastapi_bootstrap.middleware import SecurityHeadersMiddleware + + app.add_middleware( + SecurityHeadersMiddleware, + stage=self._settings.stage.value + if isinstance(self._settings.stage, Stage) + else self._settings.stage, + **self._security_headers_config, + ) + + # Add request ID middleware if enabled + if self._enable_request_id: + from fastapi_bootstrap.middleware import RequestIDMiddleware + + app.add_middleware(RequestIDMiddleware, header_name=self._request_id_header) + + # Add request timing middleware if enabled + if self._enable_request_timing: + from fastapi_bootstrap.middleware import RequestTimingMiddleware + + app.add_middleware(RequestTimingMiddleware, header_name=self._request_timing_header) + + # Add metrics middleware if enabled + if self._enable_metrics: + from fastapi_bootstrap.metrics import MetricsMiddleware, get_metrics_router + + app.add_middleware(MetricsMiddleware) + app.include_router( + get_metrics_router( + path=self._settings.metrics.endpoint, + include_in_schema=self._settings.metrics.include_in_schema, + ) + ) + + # Add custom middlewares + for middleware, kwargs in self._middlewares: + app.add_middleware(middleware, **kwargs) + + return app + + +# Convenient function to create a builder +def bootstrap() -> FastAPIBootstrap: + """Create a new FastAPIBootstrap builder. + + Returns: + FastAPIBootstrap instance for chaining + + Example: + ```python + from fastapi_bootstrap.builder import bootstrap + + app = bootstrap().title("My API").add_router(router).build() + ``` + """ + return FastAPIBootstrap() + + +__all__ = ["FastAPIBootstrap", "bootstrap"] diff --git a/src/fastapi_bootstrap/config.py b/src/fastapi_bootstrap/config.py new file mode 100644 index 0000000..88d39a0 --- /dev/null +++ b/src/fastapi_bootstrap/config.py @@ -0,0 +1,493 @@ +"""Configuration management for FastAPI Bootstrap. + +This module provides centralized configuration using Pydantic Settings, +supporting environment variables, .env files, and programmatic configuration. + +Example: + ```python + from fastapi_bootstrap.config import BootstrapSettings + + # Load from environment variables + settings = BootstrapSettings() + + # Or override programmatically + settings = BootstrapSettings( + stage="prod", + log_level="WARNING", + cors_origins=["https://myapp.com"] + ) + ``` +""" + +import os +from enum import Enum +from functools import lru_cache +from typing import Any + +from pydantic import BaseModel, Field, field_validator, model_validator + + +class Stage(str, Enum): + """Application environment stage.""" + + DEV = "dev" + STAGING = "staging" + PROD = "prod" + + +class LogFormat(str, Enum): + """Log output format.""" + + JSON = "json" + TEXT = "text" + + +class LoggingSettings(BaseModel): + """Logging configuration settings.""" + + level: str = Field( + default="INFO", + description="Log level (DEBUG, INFO, WARNING, ERROR, CRITICAL)", + ) + format: LogFormat = Field( + default=LogFormat.TEXT, + description="Log output format (json or text)", + ) + json_output: bool = Field( + default=False, + description="Enable JSON log output (deprecated, use format instead)", + ) + string_max_length: int = Field( + default=5000, + ge=100, + le=100000, + description="Maximum length for logged strings before truncation", + ) + truncation_threshold: int = Field( + default=2000, + ge=100, + le=50000, + description="Threshold for truncating strings in nested structures", + ) + traceback_limit: int = Field( + default=10, + ge=1, + le=100, + description="Maximum traceback depth to show in logs", + ) + + @field_validator("level") + @classmethod + def validate_level(cls, v: str) -> str: + """Validate and normalize log level.""" + valid_levels = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"} + normalized = v.upper() + if normalized not in valid_levels: + raise ValueError( + f"Invalid log level: '{v}' (normalized: '{normalized}'). " + f"Must be one of {sorted(valid_levels)}" + ) + return normalized + + @model_validator(mode="after") + def validate_format_consistency(self) -> "LoggingSettings": + """Warn if json_output conflicts with format setting.""" + import warnings + + if self.json_output and self.format == LogFormat.TEXT: + warnings.warn( + "LoggingSettings has json_output=True but format=TEXT. " + "The 'json_output' field is deprecated; use format=LogFormat.JSON instead.", + DeprecationWarning, + stacklevel=2, + ) + return self + + +class CORSSettings(BaseModel): + """CORS (Cross-Origin Resource Sharing) configuration.""" + + origins: list[str] = Field( + default_factory=list, + description="Allowed origins. Empty list means no origins allowed.", + ) + allow_credentials: bool = Field( + default=True, + description="Allow credentials in CORS requests", + ) + allow_methods: list[str] = Field( + default_factory=lambda: ["GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS"], + description="Allowed HTTP methods", + ) + allow_headers: list[str] = Field( + default_factory=lambda: ["*"], + description="Allowed HTTP headers", + ) + expose_headers: list[str] = Field( + default_factory=lambda: ["X-Request-ID", "X-Trace-ID"], + description="Headers exposed to the browser", + ) + max_age: int = Field( + default=600, + ge=0, + le=86400, + description="Max age for CORS preflight cache (seconds)", + ) + + +class SecuritySettings(BaseModel): + """Security-related configuration.""" + + # Security Headers + enable_security_headers: bool = Field( + default=True, + description="Enable security headers middleware", + ) + hsts_max_age: int = Field( + default=31536000, # 1 year + ge=0, + description="HSTS max-age in seconds (0 to disable)", + ) + content_security_policy: str | None = Field( + default="default-src 'self'", + description="Content-Security-Policy header value", + ) + x_frame_options: str = Field( + default="DENY", + description="X-Frame-Options header value", + ) + x_content_type_options: str = Field( + default="nosniff", + description="X-Content-Type-Options header value", + ) + + # Request limits + max_request_size: int = Field( + default=10 * 1024 * 1024, # 10MB + ge=1024, + description="Maximum request body size in bytes", + ) + + # Sensitive data masking + mask_sensitive_fields: list[str] = Field( + default_factory=lambda: [ + "password", + "passwd", + "secret", + "token", + "api_key", + "apikey", + "authorization", + "auth", + "credential", + "private_key", + "access_token", + "refresh_token", + ], + description="Field names to mask in logs", + ) + + +class RateLimitSettings(BaseModel): + """Rate limiting configuration.""" + + enabled: bool = Field( + default=False, + description="Enable rate limiting", + ) + requests_per_minute: int = Field( + default=60, + ge=1, + description="Maximum requests per minute per client", + ) + burst_size: int = Field( + default=10, + ge=1, + description="Allowed burst size above the limit", + ) + + +class MetricsSettings(BaseModel): + """Metrics and monitoring configuration.""" + + enabled: bool = Field( + default=False, + description="Enable metrics endpoint", + ) + endpoint: str = Field( + default="/metrics", + description="Metrics endpoint path", + ) + include_in_schema: bool = Field( + default=False, + description="Include metrics endpoint in OpenAPI schema", + ) + + +class HealthCheckSettings(BaseModel): + """Health check configuration.""" + + endpoint: str = Field( + default="/healthz", + description="Health check endpoint path", + ) + include_in_schema: bool = Field( + default=False, + description="Include health endpoint in OpenAPI schema", + ) + + +class GracefulShutdownSettings(BaseModel): + """Graceful shutdown configuration.""" + + timeout: int = Field( + default=10, + ge=0, + le=300, + description="Seconds to wait for in-flight requests during shutdown", + ) + + +class BootstrapSettings(BaseModel): + """Main configuration class for FastAPI Bootstrap. + + This class aggregates all configuration settings and can be loaded + from environment variables or configured programmatically. + + Environment Variables: + - STAGE: Application stage (dev, staging, prod) + - LOG_LEVEL: Logging level + - LOG_JSON: Enable JSON logging ("true" or "false") + - LOG_STRING_LENGTH: Max string length in logs + - CORS_ORIGINS: Comma-separated list of allowed origins + - RATE_LIMIT_ENABLED: Enable rate limiting + - RATE_LIMIT_RPM: Requests per minute + + Example: + ```python + # Load from environment + settings = BootstrapSettings() + + # Override specific values + settings = BootstrapSettings( + stage=Stage.PROD, + logging=LoggingSettings(level="WARNING"), + cors=CORSSettings(origins=["https://myapp.com"]), + ) + ``` + """ + + stage: Stage = Field( + default=Stage.DEV, + description="Application environment stage", + ) + + # Sub-configurations + logging: LoggingSettings = Field(default_factory=LoggingSettings) + cors: CORSSettings = Field(default_factory=CORSSettings) + security: SecuritySettings = Field(default_factory=SecuritySettings) + rate_limit: RateLimitSettings = Field(default_factory=RateLimitSettings) + metrics: MetricsSettings = Field(default_factory=MetricsSettings) + health_check: HealthCheckSettings = Field(default_factory=HealthCheckSettings) + graceful_shutdown: GracefulShutdownSettings = Field(default_factory=GracefulShutdownSettings) + + # Application metadata + title: str = Field(default="FastAPI Application", description="API title") + version: str = Field(default="0.1.0", description="API version") + description: str = Field(default="", description="API description") + + model_config = {"use_enum_values": True} + + @classmethod + def from_env(cls) -> "BootstrapSettings": + """Load settings from environment variables. + + Returns: + BootstrapSettings instance configured from environment + + Raises: + ValueError: If environment variable cannot be parsed to expected type + + Example: + ```python + # Set environment variables + # STAGE=prod + # LOG_LEVEL=WARNING + # CORS_ORIGINS=https://myapp.com,https://api.myapp.com + + settings = BootstrapSettings.from_env() + ``` + """ + + def parse_int_env(name: str, default: str) -> int: + """Parse integer from environment variable with error handling.""" + value = os.getenv(name, default) + try: + return int(value) + except ValueError as e: + raise ValueError( + f"Environment variable {name} must be an integer, got: '{value}'" + ) from e + + # Parse stage + stage_str = os.getenv("STAGE", "dev").lower() + stage = Stage(stage_str) if stage_str in [s.value for s in Stage] else Stage.DEV + + # Parse logging settings + log_json = os.getenv("LOG_JSON", "false").lower() in ("true", "1", "yes") + logging_settings = LoggingSettings( + level=os.getenv("LOG_LEVEL", "INFO"), + format=LogFormat.JSON if log_json else LogFormat.TEXT, + json_output=log_json, + string_max_length=parse_int_env("LOG_STRING_LENGTH", "5000"), + truncation_threshold=parse_int_env("LOG_TRUNCATION_THRESHOLD", "2000"), + traceback_limit=parse_int_env("TRACEBACKLIMIT", "10"), + ) + + # Parse CORS settings + cors_origins_str = os.getenv("CORS_ORIGINS", "") + cors_origins = [o.strip() for o in cors_origins_str.split(",") if o.strip()] + allow_credentials = os.getenv("CORS_ALLOW_CREDENTIALS", "true").lower() in ( + "true", + "1", + "yes", + ) + + # Validate CORS configuration - browsers reject credentials with wildcard + if allow_credentials and "*" in cors_origins: + raise ValueError( + "Invalid CORS configuration: allow_credentials=True cannot be used " + 'with origins containing "*". Per the CORS specification, browsers ' + "will reject this combination. Configure explicit origins instead, " + 'e.g. CORS_ORIGINS="https://example.com,https://api.example.com"' + ) + + cors_settings = CORSSettings( + origins=cors_origins, + allow_credentials=allow_credentials, + ) + + # Parse rate limit settings + rate_limit_settings = RateLimitSettings( + enabled=os.getenv("RATE_LIMIT_ENABLED", "false").lower() in ("true", "1", "yes"), + requests_per_minute=parse_int_env("RATE_LIMIT_RPM", "60"), + ) + + # Parse metrics settings + metrics_settings = MetricsSettings( + enabled=os.getenv("METRICS_ENABLED", "false").lower() in ("true", "1", "yes"), + endpoint=os.getenv("METRICS_ENDPOINT", "/metrics"), + ) + + # Parse graceful shutdown + graceful_settings = GracefulShutdownSettings( + timeout=parse_int_env("GRACEFUL_SHUTDOWN_TIMEOUT", "10"), + ) + + return cls( + stage=stage, + logging=logging_settings, + cors=cors_settings, + rate_limit=rate_limit_settings, + metrics=metrics_settings, + graceful_shutdown=graceful_settings, + title=os.getenv("APP_TITLE", "FastAPI Application"), + version=os.getenv("APP_VERSION", "0.1.0"), + description=os.getenv("APP_DESCRIPTION", ""), + ) + + def get_cors_config_for_stage(self) -> CORSSettings: + """Get CORS configuration adjusted for the current stage. + + Returns: + CORSSettings with stage-appropriate defaults applied + """ + # If origins are explicitly set, use them + if self.cors.origins: + return self.cors + + # Apply stage-specific defaults + if self.stage == Stage.DEV: + return CORSSettings( + origins=["*"], + allow_methods=["*"], + allow_headers=["*"], + allow_credentials=True, + ) + elif self.stage == Stage.STAGING: + return CORSSettings( + origins=self.cors.origins or [], + allow_methods=self.cors.allow_methods, + allow_headers=self.cors.allow_headers, + allow_credentials=True, + ) + else: # PROD + return CORSSettings( + origins=self.cors.origins, # Must be explicitly set + allow_methods=["GET", "POST", "PUT", "DELETE", "PATCH"], + allow_headers=["Content-Type", "Authorization", "X-Request-ID"], + allow_credentials=True, + ) + + +@lru_cache() +def get_settings() -> BootstrapSettings: + """Get cached settings instance. + + Returns: + Cached BootstrapSettings instance loaded from environment + + Example: + ```python + from fastapi_bootstrap.config import get_settings + + settings = get_settings() + print(settings.stage) # "dev" + ``` + """ + return BootstrapSettings.from_env() + + +def mask_sensitive_data(data: Any, sensitive_fields: list[str] | None = None) -> Any: + """Mask sensitive fields in a data structure. + + Recursively traverses dictionaries and lists to mask values of + fields that match sensitive field names. + + Args: + data: Data structure to mask (dict, list, or primitive) + sensitive_fields: List of field names to mask. If None, loads from + get_settings().security.mask_sensitive_fields (cached via @lru_cache). + + Note: + When sensitive_fields is None, this function calls get_settings() which + requires BootstrapSettings to be loadable from environment. For standalone + usage without environment configuration, provide explicit sensitive_fields. + + Returns: + Data structure with sensitive values replaced with "***MASKED***" + + Example: + ```python + data = {"username": "john", "password": "secret123"} + masked = mask_sensitive_data(data) + # {"username": "john", "password": "***MASKED***"} + ``` + """ + if sensitive_fields is None: + sensitive_fields = get_settings().security.mask_sensitive_fields + + sensitive_set = {f.lower() for f in sensitive_fields} + + def _mask(obj: Any) -> Any: + if isinstance(obj, dict): + return { + k: "***MASKED***" if k.lower() in sensitive_set else _mask(v) + for k, v in obj.items() + } + elif isinstance(obj, list): + return [_mask(item) for item in obj] + else: + return obj + + return _mask(data) diff --git a/src/fastapi_bootstrap/metrics.py b/src/fastapi_bootstrap/metrics.py new file mode 100644 index 0000000..a5d221c --- /dev/null +++ b/src/fastapi_bootstrap/metrics.py @@ -0,0 +1,457 @@ +"""Prometheus metrics for FastAPI Bootstrap. + +This module provides Prometheus-compatible metrics collection and export, +including request counts, latencies, and error rates. + +Example: + ```python + from fastapi_bootstrap.metrics import MetricsMiddleware, get_metrics_router + + app.add_middleware(MetricsMiddleware) + app.include_router(get_metrics_router()) + ``` +""" + +import time +from collections import defaultdict +from collections.abc import Callable +from dataclasses import dataclass, field +from threading import Lock + +from fastapi import APIRouter, Request, Response +from starlette.middleware.base import BaseHTTPMiddleware +from starlette.routing import Match +from starlette.types import ASGIApp + + +@dataclass +class Histogram: + """A simple histogram implementation for tracking distributions. + + Tracks count, sum, and bucket counts for Prometheus histogram format. + """ + + buckets: tuple[float, ...] = (0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0) + _sum: float = 0.0 + _count: int = 0 + _lock: Lock = field(default_factory=Lock) + _counts: dict[float, int] = field(default_factory=dict) # Per-instance dict, not shared + + def __post_init__(self) -> None: + """Initialize per-bucket counts to 0.""" + for bucket in self.buckets: + if bucket not in self._counts: + self._counts[bucket] = 0 + + def observe(self, value: float) -> None: + """Record an observation. + + Increments only the smallest bucket that can contain the value. + Cumulative counts are computed in get_bucket_counts(). + """ + with self._lock: + self._sum += value + self._count += 1 + # Find the smallest bucket that can contain this value + for bucket in sorted(self.buckets): + if value <= bucket: + self._counts[bucket] += 1 + break # Only increment one bucket, not all matching buckets + + @property + def sum(self) -> float: + """Total sum of all observations.""" + return self._sum + + @property + def count(self) -> int: + """Total count of observations.""" + return self._count + + def get_bucket_counts(self) -> dict[float, int]: + """Get cumulative bucket counts (Prometheus histogram format). + + Returns counts where each bucket includes all observations <= bucket boundary. + """ + cumulative = {} + total = 0 + for bucket in sorted(self.buckets): + total += self._counts.get(bucket, 0) + cumulative[bucket] = total + return cumulative + + +@dataclass +class Counter: + """A simple counter implementation.""" + + _value: int = 0 + _lock: Lock = field(default_factory=Lock) + + def inc(self, value: int = 1) -> None: + """Increment the counter.""" + with self._lock: + self._value += value + + @property + def value(self) -> int: + """Current counter value.""" + return self._value + + +@dataclass +class Gauge: + """A simple gauge implementation.""" + + _value: float = 0.0 + _lock: Lock = field(default_factory=Lock) + + def set(self, value: float) -> None: + """Set the gauge value.""" + with self._lock: + self._value = value + + def inc(self, value: float = 1.0) -> None: + """Increment the gauge.""" + with self._lock: + self._value += value + + def dec(self, value: float = 1.0) -> None: + """Decrement the gauge.""" + with self._lock: + self._value -= value + + @property + def value(self) -> float: + """Current gauge value.""" + return self._value + + +class MetricsRegistry: + """Central registry for all application metrics. + + This class maintains all metrics and provides export functionality + in Prometheus text format. + + Example: + ```python + registry = MetricsRegistry() + registry.request_count.labels(method="GET", path="/api/users").inc() + ``` + """ + + def __init__(self, app_name: str = "fastapi_bootstrap"): + """Initialize the metrics registry. + + Args: + app_name: Application name prefix for metrics + """ + self.app_name = app_name + self._lock = Lock() + + # Request metrics + self._request_count: dict[tuple, Counter] = {} + self._request_latency: dict[tuple, Histogram] = {} + self._request_in_progress: dict[tuple, Gauge] = {} + + # Error metrics + self._error_count: dict[tuple, Counter] = {} + + # App info + self._app_info: dict[str, str] = {} + + def set_app_info(self, version: str, python_version: str) -> None: + """Set application info for info metric.""" + self._app_info = { + "version": version, + "python_version": python_version, + } + + def get_request_count(self, method: str, path: str, status: int) -> Counter: + """Get or create a request counter.""" + key = (method, path, status) + if key not in self._request_count: + with self._lock: + if key not in self._request_count: + self._request_count[key] = Counter() + return self._request_count[key] + + def get_request_latency(self, method: str, path: str) -> Histogram: + """Get or create a request latency histogram.""" + key = (method, path) + if key not in self._request_latency: + with self._lock: + if key not in self._request_latency: + self._request_latency[key] = Histogram() + return self._request_latency[key] + + def get_requests_in_progress(self, method: str, path: str) -> Gauge: + """Get or create an in-progress requests gauge.""" + key = (method, path) + if key not in self._request_in_progress: + with self._lock: + if key not in self._request_in_progress: + self._request_in_progress[key] = Gauge() + return self._request_in_progress[key] + + def get_error_count(self, method: str, path: str, error_type: str) -> Counter: + """Get or create an error counter.""" + key = (method, path, error_type) + if key not in self._error_count: + with self._lock: + if key not in self._error_count: + self._error_count[key] = Counter() + return self._error_count[key] + + def export(self) -> str: + """Export all metrics in Prometheus text format. + + Returns: + Prometheus-compatible metrics text + """ + lines: list[str] = [] + + # App info + if self._app_info: + lines.append(f"# HELP {self.app_name}_app_info Application information") + lines.append(f"# TYPE {self.app_name}_app_info gauge") + labels = ",".join(f'{k}="{v}"' for k, v in self._app_info.items()) + lines.append(f"{self.app_name}_app_info{{{labels}}} 1") + lines.append("") + + # Request count + if self._request_count: + lines.append(f"# HELP {self.app_name}_http_requests_total Total HTTP requests") + lines.append(f"# TYPE {self.app_name}_http_requests_total counter") + for (method, path, status), counter in sorted(self._request_count.items()): + lines.append( + f'{self.app_name}_http_requests_total{{method="{method}",path="{path}",status="{status}"}} {counter.value}' + ) + lines.append("") + + # Request latency histogram + if self._request_latency: + lines.append( + f"# HELP {self.app_name}_http_request_duration_seconds HTTP request duration" + ) + lines.append(f"# TYPE {self.app_name}_http_request_duration_seconds histogram") + for (method, path), histogram in sorted(self._request_latency.items()): + bucket_counts = histogram.get_bucket_counts() + for bucket, count in bucket_counts.items(): + lines.append( + f'{self.app_name}_http_request_duration_seconds_bucket{{method="{method}",path="{path}",le="{bucket}"}} {count}' + ) + lines.append( + f'{self.app_name}_http_request_duration_seconds_bucket{{method="{method}",path="{path}",le="+Inf"}} {histogram.count}' + ) + lines.append( + f'{self.app_name}_http_request_duration_seconds_sum{{method="{method}",path="{path}"}} {histogram.sum}' + ) + lines.append( + f'{self.app_name}_http_request_duration_seconds_count{{method="{method}",path="{path}"}} {histogram.count}' + ) + lines.append("") + + # Requests in progress + if self._request_in_progress: + lines.append( + f"# HELP {self.app_name}_http_requests_in_progress Current in-progress requests" + ) + lines.append(f"# TYPE {self.app_name}_http_requests_in_progress gauge") + for (method, path), gauge in sorted(self._request_in_progress.items()): + lines.append( + f'{self.app_name}_http_requests_in_progress{{method="{method}",path="{path}"}} {gauge.value}' + ) + lines.append("") + + # Error count + if self._error_count: + lines.append(f"# HELP {self.app_name}_http_errors_total Total HTTP errors") + lines.append(f"# TYPE {self.app_name}_http_errors_total counter") + for (method, path, error_type), counter in sorted(self._error_count.items()): + lines.append( + f'{self.app_name}_http_errors_total{{method="{method}",path="{path}",error_type="{error_type}"}} {counter.value}' + ) + lines.append("") + + return "\n".join(lines) + + +# Global metrics registries keyed by app_name (with thread-safe access) +_metrics_registries: dict[str, MetricsRegistry] = {} +_registry_lock = Lock() + + +def get_metrics_registry(app_name: str = "fastapi_bootstrap") -> MetricsRegistry: + """Get the global metrics registry for an app. + + Thread-safe: uses double-checked locking pattern with explicit lock. + + Args: + app_name: Application name for metric prefixes + + Returns: + The MetricsRegistry instance for the given app_name + """ + global _metrics_registries + # Fast path: check without lock + if app_name in _metrics_registries: + return _metrics_registries[app_name] + + # Slow path: acquire lock and double-check + with _registry_lock: + if app_name not in _metrics_registries: + _metrics_registries[app_name] = MetricsRegistry(app_name) + return _metrics_registries[app_name] + + +def reset_metrics_registry() -> None: + """Reset all global metrics registries (for testing).""" + global _metrics_registries + _metrics_registries = {} + + +class MetricsMiddleware(BaseHTTPMiddleware): + """Middleware that collects HTTP request metrics. + + Collects: + - Request count by method, path, and status code + - Request duration histogram + - In-progress requests gauge + - Error counts by type + + Example: + ```python + from fastapi_bootstrap.metrics import MetricsMiddleware + + app.add_middleware(MetricsMiddleware) + ``` + """ + + def __init__( + self, + app: ASGIApp, + app_name: str = "fastapi_bootstrap", + exclude_paths: list[str] | None = None, + ): + """Initialize the metrics middleware. + + Args: + app: The ASGI application + app_name: Application name for metric prefixes + exclude_paths: Paths to exclude from metrics (e.g., ["/healthz", "/metrics"]) + """ + super().__init__(app) + self.registry = get_metrics_registry(app_name) + self.exclude_paths = set(exclude_paths or ["/healthz", "/metrics", "/docs", "/redoc"]) + + def _get_path_template(self, request: Request) -> str: + """Get the path template for the request (uses route pattern, not actual path). + + This ensures /users/123 and /users/456 are grouped as /users/{id} + """ + # First, try to use the route that Starlette/FastAPI has already matched + route = request.scope.get("route") + if route is not None: + path = getattr(route, "path", None) + if path: + return path + + # Fallback: iterate routes to find matching one (for non-standard scopes) + for route in request.app.routes: + match, _ = route.matches(request.scope) + if match == Match.FULL: + return getattr(route, "path", request.url.path) + return request.url.path + + async def dispatch(self, request: Request, call_next: Callable) -> Response: + """Process the request and collect metrics.""" + path = request.url.path + + # Skip excluded paths + if path in self.exclude_paths: + return await call_next(request) + + method = request.method + path_template = self._get_path_template(request) + + # Track in-progress requests + in_progress = self.registry.get_requests_in_progress(method, path_template) + in_progress.inc() + + start_time = time.perf_counter() + status_code = 500 + error_type = None + + try: + response = await call_next(request) + status_code = response.status_code + return response + except Exception as e: + error_type = type(e).__name__ + raise + finally: + # Record duration + duration = time.perf_counter() - start_time + self.registry.get_request_latency(method, path_template).observe(duration) + + # Record request count + self.registry.get_request_count(method, path_template, status_code).inc() + + # Record errors (avoid double-counting: exception OR 5xx, not both) + if error_type: + self.registry.get_error_count(method, path_template, error_type).inc() + elif status_code >= 500: + self.registry.get_error_count(method, path_template, "http_5xx").inc() + + # Decrement in-progress + in_progress.dec() + + +def get_metrics_router( + path: str = "/metrics", + include_in_schema: bool = False, + tags: list[str] | None = None, +) -> APIRouter: + """Create a router with the metrics endpoint. + + Args: + path: Path for the metrics endpoint + include_in_schema: Include in OpenAPI schema + tags: Tags for the endpoint + + Returns: + FastAPI APIRouter with metrics endpoint + + Example: + ```python + from fastapi_bootstrap.metrics import get_metrics_router + + app.include_router(get_metrics_router()) + # Metrics available at GET /metrics + ``` + """ + router = APIRouter(tags=tags or ["monitoring"]) + + @router.get(path, include_in_schema=include_in_schema) + async def metrics() -> Response: + """Export Prometheus metrics.""" + registry = get_metrics_registry() + return Response( + content=registry.export(), + media_type="text/plain; version=0.0.4; charset=utf-8", + ) + + return router + + +__all__ = [ + "Counter", + "Gauge", + "Histogram", + "MetricsMiddleware", + "MetricsRegistry", + "get_metrics_registry", + "get_metrics_router", + "reset_metrics_registry", +] diff --git a/src/fastapi_bootstrap/middleware/__init__.py b/src/fastapi_bootstrap/middleware/__init__.py new file mode 100644 index 0000000..6f2435e --- /dev/null +++ b/src/fastapi_bootstrap/middleware/__init__.py @@ -0,0 +1,297 @@ +"""Security middleware for FastAPI Bootstrap. + +This module provides security-related middleware including: +- Security headers (HSTS, CSP, X-Frame-Options, etc.) +- Request size limiting +- Request ID injection +""" + +import time +import uuid +from collections.abc import Callable +from typing import Any + +from starlette.middleware.base import BaseHTTPMiddleware +from starlette.requests import Request +from starlette.responses import Response +from starlette.types import ASGIApp + + +class SecurityHeadersMiddleware(BaseHTTPMiddleware): + """Middleware that adds security headers to all responses. + + This middleware adds the following security headers: + - Strict-Transport-Security (HSTS) + - Content-Security-Policy + - X-Frame-Options + - X-Content-Type-Options + - X-XSS-Protection + - Referrer-Policy + - Permissions-Policy + + Example: + ```python + from fastapi_bootstrap.middleware import SecurityHeadersMiddleware + + app.add_middleware( + SecurityHeadersMiddleware, + hsts_max_age=31536000, + content_security_policy="default-src 'self'", + ) + ``` + """ + + def __init__( + self, + app: ASGIApp, + hsts_max_age: int = 31536000, + hsts_include_subdomains: bool = True, + hsts_preload: bool = False, + content_security_policy: str | None = None, + x_frame_options: str = "DENY", + x_content_type_options: str = "nosniff", + x_xss_protection: str | None = None, # Deprecated header, disabled by default + referrer_policy: str = "strict-origin-when-cross-origin", + permissions_policy: str | None = None, + enable_in_dev: bool = False, + stage: str = "dev", + ): + """Initialize the security headers middleware. + + Args: + app: The ASGI application + hsts_max_age: Max age for HSTS in seconds (0 to disable) + hsts_include_subdomains: Include subdomains in HSTS + hsts_preload: Enable HSTS preload + content_security_policy: CSP header value (None to disable) + x_frame_options: X-Frame-Options value (DENY, SAMEORIGIN, or None) + x_content_type_options: X-Content-Type-Options value + x_xss_protection: X-XSS-Protection value. DEPRECATED: This header is + removed from modern browsers (Chrome 78+, Edge 79+, Firefox 70+) + in favor of Content-Security-Policy. Default is None (disabled). + Only set this if you need to support legacy browsers. + referrer_policy: Referrer-Policy value + permissions_policy: Permissions-Policy value (None for default) + enable_in_dev: Enable security headers in development mode + stage: Current application stage (dev, staging, prod) + """ + super().__init__(app) + self.hsts_max_age = hsts_max_age + self.hsts_include_subdomains = hsts_include_subdomains + self.hsts_preload = hsts_preload + self.content_security_policy = content_security_policy + self.x_frame_options = x_frame_options + self.x_content_type_options = x_content_type_options + self.x_xss_protection = x_xss_protection + self.referrer_policy = referrer_policy + self.permissions_policy = permissions_policy or "geolocation=(), microphone=(), camera=()" + self.enable_in_dev = enable_in_dev + self.stage = stage + + async def dispatch(self, request: Request, call_next: Callable) -> Response: + """Process the request and add security headers to response.""" + response = await call_next(request) + + # Skip security headers in dev unless explicitly enabled + if self.stage == "dev" and not self.enable_in_dev: + return response + + # HSTS (only for HTTPS connections) + is_https = ( + request.url.scheme == "https" or request.headers.get("x-forwarded-proto") == "https" + ) + if self.hsts_max_age > 0 and is_https: + hsts_value = f"max-age={self.hsts_max_age}" + if self.hsts_include_subdomains: + hsts_value += "; includeSubDomains" + if self.hsts_preload: + hsts_value += "; preload" + response.headers["Strict-Transport-Security"] = hsts_value + + # Content Security Policy + if self.content_security_policy: + response.headers["Content-Security-Policy"] = self.content_security_policy + + # X-Frame-Options + if self.x_frame_options: + response.headers["X-Frame-Options"] = self.x_frame_options + + # X-Content-Type-Options + if self.x_content_type_options: + response.headers["X-Content-Type-Options"] = self.x_content_type_options + + # X-XSS-Protection (deprecated, only add if explicitly enabled) + if self.x_xss_protection: + response.headers["X-XSS-Protection"] = self.x_xss_protection + + # Referrer-Policy + if self.referrer_policy: + response.headers["Referrer-Policy"] = self.referrer_policy + + # Permissions-Policy + if self.permissions_policy: + response.headers["Permissions-Policy"] = self.permissions_policy + + return response + + +class RequestIDMiddleware(BaseHTTPMiddleware): + """Middleware that ensures every request has a unique ID. + + If the request includes an X-Request-ID header, it's used. + Otherwise, a new UUID is generated. The request ID is added + to both the request state and response headers. + + Example: + ```python + from fastapi_bootstrap.middleware import RequestIDMiddleware + + app.add_middleware(RequestIDMiddleware) + + @app.get("/") + async def root(request: Request): + return {"request_id": request.state.request_id} + ``` + """ + + def __init__( + self, + app: ASGIApp, + header_name: str = "X-Request-ID", + generator: Callable[[], str] | None = None, + ): + """Initialize the request ID middleware. + + Args: + app: The ASGI application + header_name: Header name for request ID + generator: Custom ID generator function (defaults to UUID4) + """ + super().__init__(app) + self.header_name = header_name + self.generator = generator or (lambda: str(uuid.uuid4())) + + async def dispatch(self, request: Request, call_next: Callable) -> Response: + """Process the request and inject request ID.""" + # Get existing request ID or generate new one + request_id = request.headers.get(self.header_name) or self.generator() + + # Store in request state for access in handlers + request.state.request_id = request_id + + # Process request + response = await call_next(request) + + # Add request ID to response headers + response.headers[self.header_name] = request_id + + return response + + +class RequestTimingMiddleware(BaseHTTPMiddleware): + """Middleware that tracks request processing time. + + Adds X-Response-Time header with the time taken to process + the request in milliseconds. + + Example: + ```python + from fastapi_bootstrap.middleware import RequestTimingMiddleware + + app.add_middleware(RequestTimingMiddleware) + # Response will include: X-Response-Time: 123.45ms + ``` + """ + + def __init__( + self, + app: ASGIApp, + header_name: str = "X-Response-Time", + ): + """Initialize the timing middleware. + + Args: + app: The ASGI application + header_name: Header name for response time + """ + super().__init__(app) + self.header_name = header_name + + async def dispatch(self, request: Request, call_next: Callable) -> Response: + """Process the request and measure timing.""" + start_time = time.perf_counter() + + response = await call_next(request) + + process_time = (time.perf_counter() - start_time) * 1000 + response.headers[self.header_name] = f"{process_time:.2f}ms" + + return response + + +class MaxRequestSizeMiddleware: + """Middleware that limits request body size. + + Returns 413 Payload Too Large if request body exceeds the limit. + + Note: + This middleware checks the Content-Length header. For chunked transfer + encoding (no Content-Length), the request will be allowed through. + For complete protection against oversized chunked requests, consider + using a reverse proxy (nginx, traefik) with body size limits. + + Example: + ```python + from fastapi_bootstrap.middleware import MaxRequestSizeMiddleware + + app.add_middleware(MaxRequestSizeMiddleware, max_size=10 * 1024 * 1024) # 10MB + ``` + """ + + def __init__( + self, + app: ASGIApp, + max_size: int = 10 * 1024 * 1024, # 10MB default + ): + """Initialize the max size middleware. + + Args: + app: The ASGI application + max_size: Maximum request body size in bytes + """ + self.app = app + self.max_size = max_size + + async def __call__(self, scope: dict, receive: Callable, send: Callable) -> None: + """Process the request.""" + if scope["type"] != "http": + await self.app(scope, receive, send) + return + + # Check Content-Length header + headers = dict(scope.get("headers", [])) + content_length = headers.get(b"content-length") + + if content_length: + try: + length = int(content_length.decode()) + if length > self.max_size: + response = Response( + content=f"Request body too large. Max size: {self.max_size} bytes", + status_code=413, + ) + await response(scope, receive, send) + return + except ValueError: + pass # Invalid Content-Length, let the request proceed + + await self.app(scope, receive, send) + + +# Export all middleware classes +__all__ = [ + "SecurityHeadersMiddleware", + "RequestIDMiddleware", + "RequestTimingMiddleware", + "MaxRequestSizeMiddleware", +] diff --git a/tests/test_builder.py b/tests/test_builder.py new file mode 100644 index 0000000..bbb2042 --- /dev/null +++ b/tests/test_builder.py @@ -0,0 +1,194 @@ +"""Tests for Builder pattern API.""" + +from fastapi import APIRouter +from fastapi.testclient import TestClient + +from fastapi_bootstrap.builder import FastAPIBootstrap, bootstrap +from fastapi_bootstrap.config import Stage + + +class TestFastAPIBootstrap: + """Tests for FastAPIBootstrap builder.""" + + def test_basic_build(self): + """Basic build should create a working app.""" + router = APIRouter() + + @router.get("/test") + async def test_endpoint(): + return {"status": "ok"} + + app = FastAPIBootstrap().title("Test API").add_router(router).build() + + client = TestClient(app) + response = client.get("/test") + assert response.status_code == 200 + assert response.json() == {"status": "ok"} + + def test_title_and_version(self): + """Title and version should be set correctly.""" + app = FastAPIBootstrap().title("My API").version("2.0.0").build() + + assert app.title == "My API" + assert app.version == "2.0.0" + + def test_stage_configuration(self): + """Stage should affect CORS and other settings.""" + # Dev stage + app_dev = FastAPIBootstrap().stage("dev").build() + assert app_dev is not None + + # Prod stage + app_prod = FastAPIBootstrap().stage("prod").build() + assert app_prod is not None + + # Using enum + app_staging = FastAPIBootstrap().stage(Stage.STAGING).build() + assert app_staging is not None + + def test_prefix_url(self): + """URL prefix should be applied to routes.""" + router = APIRouter() + + @router.get("/items") + async def get_items(): + return {"items": []} + + app = FastAPIBootstrap().prefix("/api/v1").add_router(router).build() + + client = TestClient(app) + response = client.get("/api/v1/items") + assert response.status_code == 200 + + def test_with_metrics(self): + """Metrics endpoint should be available when enabled.""" + app = FastAPIBootstrap().with_metrics(endpoint="/metrics").build() + + client = TestClient(app) + response = client.get("/metrics") + assert response.status_code == 200 + assert "text/plain" in response.headers["content-type"] + + def test_with_health_check(self): + """Health check endpoint should be configurable.""" + app = FastAPIBootstrap().with_health_check(endpoint="/health").build() + + client = TestClient(app) + # The builder sets health_check_api parameter which creates the endpoint + response = client.get("/health") + assert response.status_code == 200 + # Health endpoint returns plain text "healthy" or JSON based on implementation + assert "healthy" in response.text.lower() or response.status_code == 200 + + def test_chaining(self): + """All methods should return self for chaining.""" + builder = FastAPIBootstrap() + + result = ( + builder.title("Test") + .version("1.0.0") + .description("A test API") + .stage("dev") + .prefix("/api") + .with_logging(level="DEBUG") + .with_cors(origins=["http://localhost"]) + .with_security_headers() + .with_metrics() + .with_health_check() + .with_graceful_shutdown(timeout=5) + ) + + assert result is builder + + def test_multiple_routers(self): + """Multiple routers should be combined.""" + router1 = APIRouter() + router2 = APIRouter() + + @router1.get("/users") + async def get_users(): + return {"users": []} + + @router2.get("/posts") + async def get_posts(): + return {"posts": []} + + app = ( + FastAPIBootstrap() + .add_router(router1, prefix="/v1") + .add_router(router2, prefix="/v2") + .build() + ) + + client = TestClient(app) + + response1 = client.get("/v1/users") + assert response1.status_code == 200 + + response2 = client.get("/v2/posts") + assert response2.status_code == 200 + + def test_startup_shutdown_coroutines(self): + """Startup and shutdown coroutines should be registered.""" + startup_called = [] + shutdown_called = [] + + async def on_startup(): + startup_called.append(True) + + async def on_shutdown(): + shutdown_called.append(True) + + app = FastAPIBootstrap().on_startup(on_startup).on_shutdown(on_shutdown).build() + + # Use TestClient as context manager to trigger lifespan events + with TestClient(app): + assert len(startup_called) == 1 + + assert len(shutdown_called) == 1 + + def test_with_request_id(self): + """Request ID middleware should add headers.""" + router = APIRouter() + + @router.get("/test") + async def test_endpoint(): + return {"status": "ok"} + + app = FastAPIBootstrap().with_request_id().add_router(router).build() + + client = TestClient(app) + response = client.get("/test") + + assert "X-Request-ID" in response.headers + + def test_with_request_timing(self): + """Request timing middleware should add headers.""" + router = APIRouter() + + @router.get("/test") + async def test_endpoint(): + return {"status": "ok"} + + app = FastAPIBootstrap().with_request_timing().add_router(router).build() + + client = TestClient(app) + response = client.get("/test") + + assert "X-Response-Time" in response.headers + assert response.headers["X-Response-Time"].endswith("ms") + + +class TestBootstrapFunction: + """Tests for bootstrap() convenience function.""" + + def test_returns_builder(self): + """bootstrap() should return a FastAPIBootstrap instance.""" + builder = bootstrap() + assert isinstance(builder, FastAPIBootstrap) + + def test_creates_app(self): + """bootstrap() should create a working app.""" + app = bootstrap().title("Test").build() + assert app is not None + assert app.title == "Test" diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..392fd15 --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,205 @@ +"""Tests for configuration module.""" + +import os +from unittest.mock import patch + +import pytest + +from fastapi_bootstrap.config import ( + BootstrapSettings, + CORSSettings, + LogFormat, + LoggingSettings, + SecuritySettings, + Stage, + get_settings, + mask_sensitive_data, +) + + +class TestLoggingSettings: + """Tests for LoggingSettings.""" + + def test_default_values(self): + """Default logging settings should be sensible.""" + settings = LoggingSettings() + assert settings.level == "INFO" + assert settings.format == LogFormat.TEXT + assert settings.string_max_length == 5000 + assert settings.truncation_threshold == 2000 + + def test_level_validation(self): + """Log level should be validated and normalized.""" + settings = LoggingSettings(level="debug") + assert settings.level == "DEBUG" + + settings = LoggingSettings(level="WARNING") + assert settings.level == "WARNING" + + def test_invalid_level_raises(self): + """Invalid log level should raise ValueError.""" + with pytest.raises(ValueError): + LoggingSettings(level="INVALID") + + +class TestCORSSettings: + """Tests for CORSSettings.""" + + def test_default_values(self): + """Default CORS settings should be empty origins.""" + settings = CORSSettings() + assert settings.origins == [] + assert settings.allow_credentials is True + assert "GET" in settings.allow_methods + + def test_custom_origins(self): + """Custom origins should be set correctly.""" + settings = CORSSettings(origins=["https://example.com", "https://api.example.com"]) + assert len(settings.origins) == 2 + assert "https://example.com" in settings.origins + + +class TestSecuritySettings: + """Tests for SecuritySettings.""" + + def test_default_sensitive_fields(self): + """Default sensitive fields should include common patterns.""" + settings = SecuritySettings() + assert "password" in settings.mask_sensitive_fields + assert "token" in settings.mask_sensitive_fields + assert "secret" in settings.mask_sensitive_fields + assert "api_key" in settings.mask_sensitive_fields + + def test_security_header_defaults(self): + """Security header defaults should be secure.""" + settings = SecuritySettings() + assert settings.enable_security_headers is True + assert settings.hsts_max_age == 31536000 + assert settings.x_frame_options == "DENY" + + +class TestBootstrapSettings: + """Tests for main BootstrapSettings class.""" + + def test_default_stage(self): + """Default stage should be dev.""" + settings = BootstrapSettings() + assert settings.stage == Stage.DEV + + def test_nested_settings(self): + """Nested settings should be accessible.""" + settings = BootstrapSettings() + assert isinstance(settings.logging, LoggingSettings) + assert isinstance(settings.cors, CORSSettings) + assert isinstance(settings.security, SecuritySettings) + + def test_from_env_basic(self): + """from_env should load from environment variables.""" + with patch.dict( + os.environ, + { + "STAGE": "prod", + "LOG_LEVEL": "WARNING", + "APP_TITLE": "Test API", + }, + clear=False, + ): + # Clear cache + get_settings.cache_clear() + settings = BootstrapSettings.from_env() + assert settings.stage == Stage.PROD + assert settings.logging.level == "WARNING" + assert settings.title == "Test API" + + def test_from_env_cors_origins(self): + """from_env should parse CORS_ORIGINS correctly.""" + with patch.dict( + os.environ, + { + "CORS_ORIGINS": "https://a.com,https://b.com, https://c.com", + }, + clear=False, + ): + # Clear cache for test isolation + get_settings.cache_clear() + settings = BootstrapSettings.from_env() + assert "https://a.com" in settings.cors.origins + assert "https://b.com" in settings.cors.origins + assert "https://c.com" in settings.cors.origins + + def test_get_cors_config_for_dev_stage(self): + """Dev stage should have permissive CORS defaults.""" + settings = BootstrapSettings(stage=Stage.DEV) + cors = settings.get_cors_config_for_stage() + assert "*" in cors.origins + assert "*" in cors.allow_methods + + def test_get_cors_config_for_prod_stage(self): + """Prod stage should have restrictive CORS defaults.""" + settings = BootstrapSettings(stage=Stage.PROD) + cors = settings.get_cors_config_for_stage() + assert cors.origins == [] # Empty by default + assert "*" not in cors.allow_methods + + def test_get_cors_config_with_explicit_origins(self): + """Explicit origins should override stage defaults.""" + settings = BootstrapSettings( + stage=Stage.DEV, cors=CORSSettings(origins=["https://custom.com"]) + ) + cors = settings.get_cors_config_for_stage() + assert cors.origins == ["https://custom.com"] + + +class TestMaskSensitiveData: + """Tests for mask_sensitive_data function.""" + + def test_mask_password(self): + """Password fields should be masked.""" + data = {"username": "john", "password": "secret123"} + result = mask_sensitive_data(data) + assert result["username"] == "john" + assert result["password"] == "***MASKED***" + + def test_mask_nested_sensitive_data(self): + """Sensitive fields in nested structures should be masked.""" + data = { + "user": { + "email": "john@example.com", + "credentials": { + "api_key": "key123", + "secret": "shhh", + }, + } + } + result = mask_sensitive_data(data) + assert result["user"]["email"] == "john@example.com" + assert result["user"]["credentials"]["api_key"] == "***MASKED***" + assert result["user"]["credentials"]["secret"] == "***MASKED***" + + def test_mask_in_list(self): + """Sensitive fields in lists should be masked.""" + data = [{"token": "abc123"}, {"token": "def456"}] + result = mask_sensitive_data(data) + assert result[0]["token"] == "***MASKED***" + assert result[1]["token"] == "***MASKED***" + + def test_case_insensitive_masking(self): + """Masking should be case-insensitive.""" + data = {"PASSWORD": "secret", "Token": "abc", "API_KEY": "key"} + result = mask_sensitive_data(data) + assert result["PASSWORD"] == "***MASKED***" + assert result["Token"] == "***MASKED***" + assert result["API_KEY"] == "***MASKED***" + + def test_custom_sensitive_fields(self): + """Custom sensitive fields should be supported.""" + data = {"custom_secret": "value", "normal": "data"} + result = mask_sensitive_data(data, sensitive_fields=["custom_secret"]) + assert result["custom_secret"] == "***MASKED***" + assert result["normal"] == "data" + + def test_non_dict_data(self): + """Non-dict data should pass through unchanged.""" + assert mask_sensitive_data("string") == "string" + assert mask_sensitive_data(123) == 123 + assert mask_sensitive_data(None) is None diff --git a/tests/test_logging.py b/tests/test_logging.py new file mode 100644 index 0000000..6e65870 --- /dev/null +++ b/tests/test_logging.py @@ -0,0 +1,144 @@ +"""Tests for logging setup and configuration.""" + +import logging + +from fastapi_bootstrap.log.setup import ( + LOG_JSON, + LOG_LEVEL, + LOG_STRING_LENGTH, + truncate_strings_in_structure, +) + + +class TestTruncateStringsInStructure: + """Tests for truncate_strings_in_structure function.""" + + def test_short_string_not_truncated(self): + """Short strings should not be modified.""" + result = truncate_strings_in_structure("short string") + assert result == "short string" + + def test_long_string_truncated(self): + """Strings over 2000 chars should be truncated.""" + long_string = "x" * 2001 + result = truncate_strings_in_structure(long_string) + assert result == "[[truncated]]" + + def test_string_at_threshold(self): + """Strings at exactly 2000 chars should not be truncated.""" + threshold_string = "x" * 2000 + result = truncate_strings_in_structure(threshold_string) + assert result == threshold_string + + def test_nested_dict_truncation(self): + """Long strings in nested dicts should be truncated.""" + data = { + "short": "hello", + "long": "x" * 2001, + "nested": { + "also_long": "y" * 2500, + "normal": "world", + }, + } + result = truncate_strings_in_structure(data) + + assert result["short"] == "hello" + assert result["long"] == "[[truncated]]" + assert result["nested"]["also_long"] == "[[truncated]]" + assert result["nested"]["normal"] == "world" + + def test_nested_list_truncation(self): + """Long strings in nested lists should be truncated.""" + data = [ + "short", + "x" * 2001, + ["y" * 2500, "normal"], + ] + result = truncate_strings_in_structure(data) + + assert result[0] == "short" + assert result[1] == "[[truncated]]" + assert result[2][0] == "[[truncated]]" + assert result[2][1] == "normal" + + def test_mixed_structure(self): + """Mixed nested structures should be handled correctly.""" + data = { + "list": ["a" * 2001, "b"], + "dict": {"long": "c" * 3000}, + "primitive": 123, + "none": None, + "bool": True, + } + result = truncate_strings_in_structure(data) + + assert result["list"][0] == "[[truncated]]" + assert result["list"][1] == "b" + assert result["dict"]["long"] == "[[truncated]]" + assert result["primitive"] == 123 + assert result["none"] is None + assert result["bool"] is True + + def test_empty_structures(self): + """Empty structures should be handled correctly.""" + assert truncate_strings_in_structure({}) == {} + assert truncate_strings_in_structure([]) == [] + assert truncate_strings_in_structure("") == "" + + +class TestEnvironmentConfiguration: + """Tests for environment-based configuration.""" + + def test_default_log_level(self): + """Default log level should be INFO.""" + # This tests the module-level default + assert LOG_LEVEL in ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] + + def test_default_log_json(self): + """Default LOG_JSON should be False.""" + # Default is false + assert isinstance(LOG_JSON, bool) + + def test_default_log_string_length(self): + """Default log string length should be 5000.""" + assert LOG_STRING_LENGTH == 5000 or isinstance(LOG_STRING_LENGTH, int) + + +class TestLogFormatting: + """Tests for log formatting behavior.""" + + def test_truncate_preserves_structure_types(self): + """Truncation should preserve the types of structures.""" + data = { + "list": [1, 2, 3], + "dict": {"a": 1}, + "string": "test", + "number": 42, + "float": 3.14, + } + result = truncate_strings_in_structure(data) + + assert isinstance(result["list"], list) + assert isinstance(result["dict"], dict) + assert isinstance(result["string"], str) + assert isinstance(result["number"], int) + assert isinstance(result["float"], float) + + def test_deeply_nested_structure(self): + """Deeply nested structures should be handled correctly.""" + data = { + "level1": { + "level2": { + "level3": { + "level4": { + "long_string": "z" * 5000, + "short_string": "hello", + } + } + } + } + } + result = truncate_strings_in_structure(data) + + assert result["level1"]["level2"]["level3"]["level4"]["long_string"] == "[[truncated]]" + assert result["level1"]["level2"]["level3"]["level4"]["short_string"] == "hello" diff --git a/tests/test_metrics.py b/tests/test_metrics.py new file mode 100644 index 0000000..fef6f7d --- /dev/null +++ b/tests/test_metrics.py @@ -0,0 +1,222 @@ +"""Tests for metrics module.""" + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from fastapi_bootstrap.metrics import ( + Counter, + Gauge, + Histogram, + MetricsMiddleware, + MetricsRegistry, + get_metrics_registry, + get_metrics_router, + reset_metrics_registry, +) + + +class TestCounter: + """Tests for Counter class.""" + + def test_initial_value(self): + """Counter should start at 0.""" + counter = Counter() + assert counter.value == 0 + + def test_increment(self): + """Counter should increment correctly.""" + counter = Counter() + counter.inc() + assert counter.value == 1 + + def test_increment_by_value(self): + """Counter should increment by specified value.""" + counter = Counter() + counter.inc(5) + assert counter.value == 5 + + def test_multiple_increments(self): + """Multiple increments should accumulate.""" + counter = Counter() + counter.inc() + counter.inc(3) + counter.inc(2) + assert counter.value == 6 + + +class TestGauge: + """Tests for Gauge class.""" + + def test_initial_value(self): + """Gauge should start at 0.""" + gauge = Gauge() + assert gauge.value == 0.0 + + def test_set_value(self): + """Gauge should set value correctly.""" + gauge = Gauge() + gauge.set(42.5) + assert gauge.value == 42.5 + + def test_increment(self): + """Gauge should increment correctly.""" + gauge = Gauge() + gauge.set(10) + gauge.inc(5) + assert gauge.value == 15 + + def test_decrement(self): + """Gauge should decrement correctly.""" + gauge = Gauge() + gauge.set(10) + gauge.dec(3) + assert gauge.value == 7 + + +class TestHistogram: + """Tests for Histogram class.""" + + def test_observe(self): + """Histogram should record observations.""" + histogram = Histogram() + histogram.observe(0.1) + histogram.observe(0.5) + assert histogram.count == 2 + assert histogram.sum == 0.6 + + def test_bucket_counts(self): + """Histogram should count buckets correctly (cumulative).""" + histogram = Histogram() + histogram.observe(0.005) # 1st bucket (0.005) + histogram.observe(0.05) # 4th bucket (0.05) + histogram.observe(1.0) # 8th bucket (1.0) + + buckets = histogram.get_bucket_counts() + # Cumulative counts: each bucket includes all observations <= bucket boundary + assert buckets[0.005] == 1 # Only the 0.005 observation + assert buckets[0.05] == 2 # 0.005 + 0.05 observations + assert buckets[1.0] == 3 # All three observations + + +class TestMetricsRegistry: + """Tests for MetricsRegistry class.""" + + def test_request_count(self): + """Registry should track request counts.""" + registry = MetricsRegistry() + counter = registry.get_request_count("GET", "/api/users", 200) + counter.inc() + counter.inc() + + same_counter = registry.get_request_count("GET", "/api/users", 200) + assert same_counter.value == 2 + + def test_request_latency(self): + """Registry should track request latency.""" + registry = MetricsRegistry() + histogram = registry.get_request_latency("GET", "/api/users") + histogram.observe(0.1) + histogram.observe(0.2) + + same_histogram = registry.get_request_latency("GET", "/api/users") + assert same_histogram.count == 2 + + def test_export_format(self): + """Registry should export in Prometheus format.""" + registry = MetricsRegistry(app_name="test_app") + registry.set_app_info("1.0.0", "3.12.0") + registry.get_request_count("GET", "/api/users", 200).inc() + + output = registry.export() + + assert "# HELP" in output + assert "# TYPE" in output + assert "test_app_http_requests_total" in output + assert 'method="GET"' in output + + +class TestMetricsMiddleware: + """Tests for MetricsMiddleware.""" + + @pytest.fixture(autouse=True) + def reset_registry(self): + """Reset metrics registry before each test.""" + reset_metrics_registry() + yield + reset_metrics_registry() + + def test_middleware_records_requests(self): + """Middleware should record request metrics.""" + app = FastAPI() + app.add_middleware(MetricsMiddleware, app_name="test") + + @app.get("/test") + async def test_endpoint(): + return {"status": "ok"} + + client = TestClient(app) + + # Make some requests + client.get("/test") + client.get("/test") + + # Check metrics + registry = get_metrics_registry("test") + # Path template might be /test + output = registry.export() + assert "http_requests_total" in output + + def test_middleware_excludes_paths(self): + """Middleware should exclude configured paths.""" + app = FastAPI() + app.add_middleware(MetricsMiddleware, exclude_paths=["/health", "/metrics"]) + + @app.get("/health") + async def health(): + return {"status": "healthy"} + + @app.get("/api/data") + async def data(): + return {"data": []} + + client = TestClient(app) + client.get("/health") + client.get("/api/data") + + # Health endpoint should be excluded, api/data should be included + registry = get_metrics_registry() + output = registry.export() + assert "/health" not in output and "/api/data" in output + + +class TestMetricsRouter: + """Tests for metrics router.""" + + @pytest.fixture(autouse=True) + def reset_registry(self): + """Reset metrics registry before each test.""" + reset_metrics_registry() + yield + reset_metrics_registry() + + def test_metrics_endpoint(self): + """Metrics endpoint should return Prometheus format.""" + app = FastAPI() + app.include_router(get_metrics_router()) + + client = TestClient(app) + response = client.get("/metrics") + + assert response.status_code == 200 + assert "text/plain" in response.headers["content-type"] + + def test_custom_metrics_path(self): + """Custom metrics path should work.""" + app = FastAPI() + app.include_router(get_metrics_router(path="/custom-metrics")) + + client = TestClient(app) + response = client.get("/custom-metrics") + + assert response.status_code == 200 diff --git a/tests/test_middleware.py b/tests/test_middleware.py new file mode 100644 index 0000000..550f1d3 --- /dev/null +++ b/tests/test_middleware.py @@ -0,0 +1,266 @@ +"""Tests for security middleware.""" + +from fastapi import FastAPI, Request +from fastapi.testclient import TestClient + +from fastapi_bootstrap.middleware import ( + MaxRequestSizeMiddleware, + RequestIDMiddleware, + RequestTimingMiddleware, + SecurityHeadersMiddleware, +) + + +class TestSecurityHeadersMiddleware: + """Tests for SecurityHeadersMiddleware.""" + + def test_adds_security_headers_in_prod(self): + """Security headers should be added in production (except HSTS on HTTP).""" + app = FastAPI() + app.add_middleware(SecurityHeadersMiddleware, stage="prod") + + @app.get("/test") + async def test_endpoint(): + return {"status": "ok"} + + client = TestClient(app) + response = client.get("/test") + + assert response.status_code == 200 + # HSTS is only added for HTTPS connections, TestClient uses HTTP + assert "Strict-Transport-Security" not in response.headers + assert "X-Frame-Options" in response.headers + assert response.headers["X-Frame-Options"] == "DENY" + assert "X-Content-Type-Options" in response.headers + assert response.headers["X-Content-Type-Options"] == "nosniff" + + def test_adds_hsts_on_https(self): + """HSTS header should be added when X-Forwarded-Proto is https.""" + app = FastAPI() + app.add_middleware(SecurityHeadersMiddleware, stage="prod") + + @app.get("/test") + async def test_endpoint(): + return {"status": "ok"} + + client = TestClient(app) + # Simulate HTTPS via proxy header + response = client.get("/test", headers={"X-Forwarded-Proto": "https"}) + + assert response.status_code == 200 + assert "Strict-Transport-Security" in response.headers + + def test_skips_headers_in_dev_by_default(self): + """Security headers should be skipped in dev by default.""" + app = FastAPI() + app.add_middleware(SecurityHeadersMiddleware, stage="dev", enable_in_dev=False) + + @app.get("/test") + async def test_endpoint(): + return {"status": "ok"} + + client = TestClient(app) + response = client.get("/test") + + assert response.status_code == 200 + assert "Strict-Transport-Security" not in response.headers + + def test_enables_headers_in_dev_when_configured(self): + """Security headers should be enabled in dev when configured.""" + app = FastAPI() + app.add_middleware(SecurityHeadersMiddleware, stage="dev", enable_in_dev=True) + + @app.get("/test") + async def test_endpoint(): + return {"status": "ok"} + + client = TestClient(app) + # Simulate HTTPS via proxy header + response = client.get("/test", headers={"X-Forwarded-Proto": "https"}) + + assert "Strict-Transport-Security" in response.headers + + def test_custom_csp(self): + """Custom CSP should be applied.""" + app = FastAPI() + app.add_middleware( + SecurityHeadersMiddleware, + stage="prod", + content_security_policy="default-src 'self'; script-src 'unsafe-inline'", + ) + + @app.get("/test") + async def test_endpoint(): + return {"status": "ok"} + + client = TestClient(app) + response = client.get("/test") + + assert "Content-Security-Policy" in response.headers + assert "script-src" in response.headers["Content-Security-Policy"] + + def test_hsts_configuration(self): + """HSTS should be configurable.""" + app = FastAPI() + app.add_middleware( + SecurityHeadersMiddleware, + stage="prod", + hsts_max_age=3600, + hsts_include_subdomains=True, + hsts_preload=True, + ) + + @app.get("/test") + async def test_endpoint(): + return {"status": "ok"} + + client = TestClient(app) + # Simulate HTTPS via proxy header + response = client.get("/test", headers={"X-Forwarded-Proto": "https"}) + + hsts = response.headers["Strict-Transport-Security"] + assert "max-age=3600" in hsts + assert "includeSubDomains" in hsts + assert "preload" in hsts + + +class TestRequestIDMiddleware: + """Tests for RequestIDMiddleware.""" + + def test_generates_request_id(self): + """Middleware should generate a request ID.""" + app = FastAPI() + app.add_middleware(RequestIDMiddleware) + + @app.get("/test") + async def test_endpoint(request: Request): + return {"request_id": request.state.request_id} + + client = TestClient(app) + response = client.get("/test") + + assert response.status_code == 200 + assert "X-Request-ID" in response.headers + assert len(response.headers["X-Request-ID"]) == 36 # UUID format + + def test_uses_existing_request_id(self): + """Middleware should use existing request ID from header.""" + app = FastAPI() + app.add_middleware(RequestIDMiddleware) + + @app.get("/test") + async def test_endpoint(request: Request): + return {"request_id": request.state.request_id} + + client = TestClient(app) + response = client.get("/test", headers={"X-Request-ID": "custom-id-123"}) + + assert response.headers["X-Request-ID"] == "custom-id-123" + assert response.json()["request_id"] == "custom-id-123" + + def test_custom_header_name(self): + """Middleware should support custom header name.""" + app = FastAPI() + app.add_middleware(RequestIDMiddleware, header_name="X-Correlation-ID") + + @app.get("/test") + async def test_endpoint(): + return {"status": "ok"} + + client = TestClient(app) + response = client.get("/test") + + assert "X-Correlation-ID" in response.headers + + def test_custom_generator(self): + """Middleware should support custom ID generator.""" + app = FastAPI() + app.add_middleware(RequestIDMiddleware, generator=lambda: "fixed-id") + + @app.get("/test") + async def test_endpoint(): + return {"status": "ok"} + + client = TestClient(app) + response = client.get("/test") + + assert response.headers["X-Request-ID"] == "fixed-id" + + +class TestRequestTimingMiddleware: + """Tests for RequestTimingMiddleware.""" + + def test_adds_timing_header(self): + """Middleware should add timing header.""" + app = FastAPI() + app.add_middleware(RequestTimingMiddleware) + + @app.get("/test") + async def test_endpoint(): + return {"status": "ok"} + + client = TestClient(app) + response = client.get("/test") + + assert "X-Response-Time" in response.headers + # Should be in format like "1.23ms" + timing = response.headers["X-Response-Time"] + assert timing.endswith("ms") + # Parse the number part + time_value = float(timing.replace("ms", "")) + assert time_value >= 0 + + def test_custom_header_name(self): + """Middleware should support custom header name.""" + app = FastAPI() + app.add_middleware(RequestTimingMiddleware, header_name="X-Processing-Time") + + @app.get("/test") + async def test_endpoint(): + return {"status": "ok"} + + client = TestClient(app) + response = client.get("/test") + + assert "X-Processing-Time" in response.headers + + +class TestMaxRequestSizeMiddleware: + """Tests for MaxRequestSizeMiddleware.""" + + def test_allows_small_requests(self): + """Small requests should be allowed.""" + app = FastAPI() + app.add_middleware(MaxRequestSizeMiddleware, max_size=1024) # 1KB + + @app.post("/test") + async def test_endpoint(request: Request): + body = await request.body() + return {"size": len(body)} + + client = TestClient(app) + response = client.post("/test", content="small data") + + assert response.status_code == 200 + + def test_rejects_large_requests(self): + """Large requests should be rejected.""" + app = FastAPI() + app.add_middleware(MaxRequestSizeMiddleware, max_size=100) # 100 bytes + + @app.post("/test") + async def test_endpoint(request: Request): + body = await request.body() + return {"size": len(body)} + + client = TestClient(app) + # Create a request larger than 100 bytes + large_content = "x" * 200 + response = client.post( + "/test", + content=large_content, + headers={"Content-Length": str(len(large_content))}, + ) + + assert response.status_code == 413 + assert "too large" in response.text.lower() diff --git a/tests/test_utils.py b/tests/test_utils.py new file mode 100644 index 0000000..275f133 --- /dev/null +++ b/tests/test_utils.py @@ -0,0 +1,151 @@ +"""Tests for utility functions.""" + +import uuid + +import pytest + +from fastapi_bootstrap.util.etc import str2bool, timeit +from fastapi_bootstrap.util.identifier import generate, get_trace_id + + +class TestStr2Bool: + """Tests for str2bool function.""" + + @pytest.mark.parametrize( + "value,expected", + [ + ("yes", True), + ("YES", True), + ("Yes", True), + ("true", True), + ("TRUE", True), + ("True", True), + ("t", True), + ("T", True), + ("y", True), + ("Y", True), + ("1", True), + ], + ) + def test_truthy_values(self, value: str, expected: bool): + """Truthy string values should return True.""" + assert str2bool(value) == expected + + @pytest.mark.parametrize( + "value,expected", + [ + ("no", False), + ("NO", False), + ("No", False), + ("false", False), + ("FALSE", False), + ("False", False), + ("f", False), + ("F", False), + ("n", False), + ("N", False), + ("0", False), + ], + ) + def test_falsy_values(self, value: str, expected: bool): + """Falsy string values should return False.""" + assert str2bool(value) == expected + + def test_invalid_value_raises(self): + """Invalid values should raise NotImplementedError.""" + with pytest.raises(NotImplementedError): + str2bool("maybe") + + with pytest.raises(NotImplementedError): + str2bool("invalid") + + with pytest.raises(NotImplementedError): + str2bool("") + + +class TestTimeitDecorator: + """Tests for timeit decorator.""" + + def test_timeit_returns_function_result(self): + """Timeit decorator should return the function's result.""" + + @timeit() + def add(a, b): + return a + b + + result = add(2, 3) + assert result == 5 + + def test_timeit_with_prefix(self): + """Timeit with prefix should work correctly.""" + + @timeit(prefix="test_function") + def multiply(a, b): + return a * b + + result = multiply(3, 4) + assert result == 12 + + def test_timeit_preserves_function_name(self): + """Timeit should preserve the original function name.""" + + @timeit() + def my_function(): + return "hello" + + # Note: Due to wraps, __name__ should be preserved + assert my_function.__name__ == "my_function" + + +class TestGenerate: + """Tests for generate UUID function.""" + + def test_generate_returns_string(self): + """Generate should return a string.""" + result = generate() + assert isinstance(result, str) + + def test_generate_returns_valid_uuid(self): + """Generate should return a valid UUID string.""" + result = generate() + # Should not raise an exception + parsed = uuid.UUID(result) + assert str(parsed) == result + + def test_generate_returns_unique_values(self): + """Generate should return unique values on each call.""" + results = [generate() for _ in range(100)] + assert len(set(results)) == 100 + + def test_generate_uuid4_format(self): + """Generate should return UUID4 format.""" + result = generate() + parsed = uuid.UUID(result) + # UUID4 has version 4 + assert parsed.version == 4 + + +class TestGetTraceId: + """Tests for get_trace_id function.""" + + def test_get_trace_id_returns_string(self): + """Get trace ID should return a string.""" + result = get_trace_id() + assert isinstance(result, str) + + def test_get_trace_id_returns_32_chars(self): + """Get trace ID should return a 32-character string.""" + result = get_trace_id() + assert len(result) == 32 + + def test_get_trace_id_is_hexadecimal(self): + """Get trace ID should return a hexadecimal string.""" + result = get_trace_id() + # Should only contain hex characters + assert all(c in "0123456789abcdef" for c in result) + + def test_get_trace_id_without_active_span(self): + """Without active span, should return zero trace ID.""" + result = get_trace_id() + # Without OpenTelemetry context, returns zeros + assert result == "0" * 32 diff --git a/uv.lock b/uv.lock index 0bf7558..1dcce51 100644 --- a/uv.lock +++ b/uv.lock @@ -278,7 +278,7 @@ wheels = [ ] [[package]] -name = "fastapi_bootstrap" +name = "fastapi-bootstrap" version = "0.0.dev0" source = { editable = "." } dependencies = [ @@ -291,6 +291,10 @@ dependencies = [ ] [package.optional-dependencies] +all = [ + { name = "httpx" }, + { name = "python-jose", extra = ["cryptography"] }, +] auth = [ { name = "httpx" }, { name = "python-jose", extra = ["cryptography"] }, @@ -309,25 +313,27 @@ dev = [ [package.metadata] requires-dist = [ - { name = "fastapi", specifier = ">=0.100.0" }, - { name = "httpx", marker = "extra == 'auth'", specifier = ">=0.24.0" }, - { name = "httpx", marker = "extra == 'dev'", specifier = ">=0.24.0" }, - { name = "loguru", specifier = ">=0.7.0" }, - { name = "mypy", marker = "extra == 'dev'", specifier = ">=1.0.0" }, - { name = "opentelemetry-api", specifier = ">=1.20.0" }, - { name = "pydantic", specifier = ">=2.0.0" }, - { name = "pytest", marker = "extra == 'dev'", specifier = ">=7.0.0" }, - { name = "pytest-asyncio", marker = "extra == 'dev'", specifier = ">=0.21.0" }, - { name = "pytest-cov", marker = "extra == 'dev'", specifier = ">=4.0.0" }, - { name = "pytest-mock", marker = "extra == 'dev'", specifier = ">=3.10.0" }, + { name = "fastapi", specifier = ">=0.109.0" }, + { name = "httpx", marker = "extra == 'all'", specifier = ">=0.26.0" }, + { name = "httpx", marker = "extra == 'auth'", specifier = ">=0.26.0" }, + { name = "httpx", marker = "extra == 'dev'", specifier = ">=0.26.0" }, + { name = "loguru", specifier = ">=0.7.2" }, + { name = "mypy", marker = "extra == 'dev'", specifier = ">=1.8.0" }, + { name = "opentelemetry-api", specifier = ">=1.22.0" }, + { name = "pydantic", specifier = ">=2.5.0" }, + { name = "pytest", marker = "extra == 'dev'", specifier = ">=8.0.0" }, + { name = "pytest-asyncio", marker = "extra == 'dev'", specifier = ">=0.23.0" }, + { name = "pytest-cov", marker = "extra == 'dev'", specifier = ">=4.1.0" }, + { name = "pytest-mock", marker = "extra == 'dev'", specifier = ">=3.12.0" }, + { name = "python-jose", extras = ["cryptography"], marker = "extra == 'all'", specifier = ">=3.3.0" }, { name = "python-jose", extras = ["cryptography"], marker = "extra == 'auth'", specifier = ">=3.3.0" }, { name = "python-jose", extras = ["cryptography"], marker = "extra == 'dev'", specifier = ">=3.3.0" }, - { name = "pyyaml", specifier = ">=6.0" }, - { name = "ruff", marker = "extra == 'dev'", specifier = ">=0.1.0" }, - { name = "types-pyyaml", marker = "extra == 'dev'", specifier = ">=6.0.0" }, - { name = "uvicorn", specifier = ">=0.20.0" }, + { name = "pyyaml", specifier = ">=6.0.1" }, + { name = "ruff", marker = "extra == 'dev'", specifier = ">=0.2.0" }, + { name = "types-pyyaml", marker = "extra == 'dev'", specifier = ">=6.0.12" }, + { name = "uvicorn", specifier = ">=0.27.0" }, ] -provides-extras = ["auth", "dev"] +provides-extras = ["auth", "metrics", "all", "dev"] [[package]] name = "h11"