High-performance Python web framework.
Built from scratch on msgspec and a custom ASGI layer. No Starlette, no Pydantic (by default), no compromises on speed — and a feature set no other Python framework ships in the box.
from hawkapi import HawkAPI
app = HawkAPI()
@app.get("/")
async def hello():
return {"message": "Hello, World!"}hawkapi dev app:app| Performance | Production rigor | Unique features |
|---|---|---|
| 6/6 throughput wins in the competitive suite | 0 known CVEs, weekly Bandit + Semgrep + pip-audit + Gitleaks + CodeQL | gRPC, GraphQL, OpenAPI mounts in one app |
| 6/6 p99 latency wins | STRIDE threat model + OWASP API Top 10 compliance | hawkapi doctor lints 18 production-readiness rules |
| Static-response cache, trivial-route fast path, mypyc-compiled router, uvloop by default | CSRF / Session / TrustedProxy / RateLimit / Bulkhead / CircuitBreaker built in | Free-threaded Python 3.13 wheels; FastAPI → HawkAPI migration codemod |
| Scenario | HawkAPI | BlackSheep | Starlette | Sanic | Litestar | FastAPI |
|---|---|---|---|---|---|---|
body_validation |
19,218 🏆 | 11,760 | 15,563 | 10,248 | 8,231 | 6,538 |
json |
34,494 🏆 | 31,715 | 27,567 | 12,241 | 13,576 | 11,365 |
path_param |
31,684 🏆 | 28,256 | 24,418 | 11,855 | 12,410 | 8,997 |
plaintext |
53,926 🏆 | 42,510 | 34,336 | 13,301 | 14,198 | 12,217 |
query_params |
19,997 🏆 | 17,363 | 16,346 | 9,992 | 11,826 | 8,533 |
routing_stress |
34,074 🏆 | 30,305 | 9,176 | 11,979 | 13,420 | 6,976 |
| Scenario | HawkAPI | BlackSheep | Starlette | Litestar | Sanic | FastAPI |
|---|---|---|---|---|---|---|
body_validation |
4.22 🏆 | 6.99 | 5.76 | 10.74 | 15.26 | 20.26 |
json |
2.44 🏆 | 2.50 | 2.82 | 5.51 | 6.47 | 6.54 |
path_param |
2.51 🏆 | 2.95 | 3.22 | 6.23 | 6.71 | 8.77 |
plaintext |
1.56 🏆 | 1.90 | 2.37 | 5.17 | 5.90 | 6.47 |
query_params |
3.85 🏆 | 4.35 | 4.50 | 6.43 | 7.83 | 10.03 |
routing_stress |
2.29 🏆 | 2.61 | 8.48 | 5.51 | 12.00 | 12.12 |
Requests/second on a shared ubuntu-latest runner with Granian (1 worker, ASGI),
wrk 4 threads × 64 connections × 10 seconds. Fresh numbers auto-regenerate every
Monday via the Competitive benchmarks workflow.
Reproduction: benchmarks/competitive/README.md.
pip install hawkapiWith extras:
pip install hawkapi[uvicorn] # Uvicorn ASGI server
pip install hawkapi[granian] # Granian ASGI server
pip install hawkapi[pydantic] # Optional Pydantic v2 support
pip install hawkapi[otel] # OpenTelemetry tracing
pip install hawkapi[metrics] # Prometheus metrics
pip install hawkapi[logging] # Structured logging (structlog)
pip install hawkapi[grpc] # gRPC support (grpcio + grpcio-reflection)
pip install hawkapi[all] # EverythingRequirements: Python 3.12+ (3.13 fully supported; experimental free-threaded 3.13t wheels are shipped — see Free-threaded Python) and msgspec >= 0.19.0.
import msgspec
from typing import Annotated
from hawkapi import HawkAPI
app = HawkAPI()
class CreateUser(msgspec.Struct):
name: str
email: str
age: Annotated[int, msgspec.Meta(ge=0, le=150)]
class UserResponse(msgspec.Struct):
id: int
name: str
email: str
@app.post("/users", status_code=201)
async def create_user(body: CreateUser) -> UserResponse:
return UserResponse(id=1, name=body.name, email=body.email)Type annotations drive validation, serialization, and OpenAPI schema generation. Invalid requests return RFC 9457 Problem Details:
{
"type": "https://hawkapi.ashimov.com/errors/validation",
"title": "Validation Error",
"status": 400,
"detail": "1 validation error",
"errors": [
{"field": "age", "message": "Expected `int` >= 0", "value": -5}
]
}Radix tree router with typed path parameters (str, int, float, uuid), query parameter coercion, and ~500ns lookups:
import uuid
@app.get("/users/{user_id:int}")
async def get_user(user_id: int):
return {"id": user_id}
@app.get("/items/{item_id:uuid}")
async def get_item(item_id: uuid.UUID):
return {"id": str(item_id)}
@app.get("/search")
async def search(q: str, limit: int = 10):
return {"query": q, "limit": limit}Both def and async def handlers work. Sync handlers run in a threadpool automatically.
Fine-tune parameter sources with Body, Query, Header, Cookie, and Path:
from typing import Annotated
from hawkapi import Query, Header
@app.get("/items")
async def list_items(
q: Annotated[str, Query(alias="search")],
x_token: Annotated[str, Header()],
limit: Annotated[int, Query(ge=1, le=100)] = 20,
):
return {"query": q, "token": x_token, "limit": limit}Full-featured DI container with three lifecycles — works in routes, background tasks, CLI commands, and tests:
from hawkapi import HawkAPI, Container, Depends
container = Container()
container.singleton(Database, factory=lambda: Database(url=DB_URL))
container.scoped(Session, factory=lambda db=Depends(Database): db.session())
app = HawkAPI(container=container)
@app.get("/users/{user_id}")
async def get_user(user_id: int, session: Session):
return await session.get(User, user_id)| Lifecycle | Behavior |
|---|---|
singleton |
Created once, shared globally |
scoped |
Created once per request |
transient |
Created fresh every injection |
Generator dependencies with yield manage resource lifecycles — cleanup code runs after the handler completes, even on exceptions:
async def get_db():
db = await create_connection()
try:
yield db
finally:
await db.close()
@app.get("/users")
async def list_users(db: Annotated[Connection, Depends(get_db)]):
return await db.fetch_all("SELECT * FROM users")Side-effect dependencies (auth guards, audit writers) that run before the handler — return values are discarded; HTTPException short-circuits the request:
from hawkapi import Depends
async def audit(request): ...
@app.get("/admin", dependencies=[Depends(audit)])
async def admin(): ...Router-level is also supported: Router(dependencies=[Depends(audit)]).
Inspect the container at runtime or export a Mermaid dependency graph:
from hawkapi.di.introspection import container_graph, to_mermaid
print(container_graph(container)) # JSON-serializable dict of all providers
print(to_mermaid(container)) # Mermaid graph TD diagramOpenAPI 3.1 schema auto-generated from type annotations:
| URL | UI |
|---|---|
/docs |
Swagger UI |
/redoc |
ReDoc |
/scalar |
Scalar |
/openapi.json |
Raw JSON schema |
All security schemes appear in the Authorize button automatically. Disable with app = HawkAPI(docs_url=None, openapi_url=None).
Filter and validate responses — hide internal fields from API output:
class UserFull(msgspec.Struct):
id: int
name: str
email: str
password_hash: str # Internal field
class UserOut(msgspec.Struct):
id: int
name: str
email: str
@app.get("/users/{user_id}", response_model=UserOut)
async def get_user(user_id: int):
# password_hash is automatically filtered out
return await db.get_user(user_id)Works with both msgspec Structs and Pydantic models.
Auto-inference from return annotation — response_model is inferred automatically when not set explicitly:
@app.get("/users/{id}")
async def get_user(id: int) -> UserOut: # response_model inferred automatically
return await db.get(id)Exclusion flags — strip None, unset, or default-value fields from serialized output:
@app.get("/users/{id}", response_model_exclude_none=True, response_model_exclude_unset=True)
async def get_user(id: int) -> UserOut: ...Flags: response_model_exclude_none, response_model_exclude_unset, response_model_exclude_defaults.
Built-in offset and cursor pagination helpers:
from hawkapi import Page, PaginationParams, CursorPage, CursorParams
@app.get("/users")
async def list_users(params: PaginationParams) -> Page[UserOut]:
users, total = await db.get_users(offset=params.offset, limit=params.limit)
return Page(items=users, total=total, page=params.page, size=params.limit)
@app.get("/feed")
async def feed(params: CursorParams) -> CursorPage[PostOut]:
posts, next_cursor = await db.get_feed(after=params.after, limit=params.limit)
return CursorPage(items=posts, next_cursor=next_cursor)from hawkapi import Middleware, Request, Response
class AuthMiddleware(Middleware):
async def before_request(self, request: Request) -> Request | Response:
token = request.headers.get("authorization")
if not token:
return Response(status_code=401)
request.state.user = verify_token(token)
return request| Middleware | Description |
|---|---|
CORSMiddleware |
Cross-Origin Resource Sharing |
GZipMiddleware |
Response compression (streaming-aware) |
TimingMiddleware |
Server-Timing header |
TrustedHostMiddleware |
Host header validation |
TrustedProxyMiddleware |
X-Forwarded-For/Proto/Host from trusted CIDRs |
SecurityHeadersMiddleware |
X-Content-Type-Options, X-Frame-Options, etc. |
RequestIDMiddleware |
X-Request-ID header (generates UUID4 if missing) |
HTTPSRedirectMiddleware |
Redirect HTTP to HTTPS |
RateLimitMiddleware |
Token bucket rate limiting (429 + Retry-After) |
RedisRateLimitMiddleware |
Redis-backed token bucket rate limiting (distributed) |
RequestLimitsMiddleware |
Max query string / header size enforcement |
CircuitBreakerMiddleware |
Three-state circuit breaker (per-path tracking) |
RedisCircuitBreakerMiddleware |
Distributed circuit breaker (Redis-backed) |
AdaptiveConcurrencyMiddleware |
Adaptive in-flight limit based on latency |
CSRFMiddleware |
Double-submit cookie CSRF protection |
SessionMiddleware |
Signed cookie-based session management |
ErrorHandlerMiddleware |
Structured error handling pipeline |
PrometheusMiddleware |
Prometheus-compatible /metrics endpoint |
StructuredLoggingMiddleware |
JSON request/response logs |
DebugMiddleware |
/_debug/routes and /_debug/stats endpoints |
ObservabilityMiddleware |
All-in-one tracing, structured logs, metrics |
Middleware is registered globally via app.add_middleware(). Each entry is stored as a MiddlewareEntry dataclass holding the middleware class and its kwargs, then compiled into an onion-model pipeline at startup.
See also: Bulkhead — named async concurrency isolator living in hawkapi.middleware.bulkhead.
from hawkapi import HTTPBearer, Depends
auth = HTTPBearer()
@app.get("/protected")
async def protected(credentials=Depends(auth)):
return {"token": credentials.credentials}Built-in schemes: HTTPBearer, HTTPBasic, APIKeyHeader, APIKeyQuery, APIKeyCookie, OAuth2PasswordBearer.
Comparing credentials safely.
HTTPBasic/HTTPBeareronly extract credentials; comparison against your stored secret is your responsibility. Always use a constant-time helper to avoid timing attacks:import secrets if not secrets.compare_digest(creds.password, stored_hash): raise HTTPException(401, detail="Invalid credentials")
from hawkapi.security import PermissionPolicy
app.permission_policy = PermissionPolicy(
resolver=get_user_permissions,
mode="all", # "all" = require all listed, "any" = require at least one
)
@app.get("/admin", permissions=["admin:read"])
async def admin_panel():
return {"secret": "data"}Fine-grained scope enforcement with OpenAPI operation.security reflection:
from hawkapi import OAuth2PasswordBearer, Security, SecurityScopes
oauth2 = OAuth2PasswordBearer(
tokenUrl="token",
scopes={"items:read": "Read items", "items:write": "Write items"},
)
async def current_user(scopes: SecurityScopes, token: str = Security(oauth2)): ...
@app.get("/items", dependencies=[Security(current_user, scopes=["items:read"])])
async def read_items(): ...Route-level scopes are aggregated into the OpenAPI operation.security field automatically.
from hawkapi import (
JSONResponse, HTMLResponse, PlainTextResponse,
RedirectResponse, StreamingResponse, FileResponse,
EventSourceResponse, ServerSentEvent,
)
# Server-Sent Events
async def events():
yield ServerSentEvent(data="connected", event="open")
yield ServerSentEvent(data='{"temp": 22.5}', event="reading")
return EventSourceResponse(events())HawkAPI supports automatic content negotiation via the Accept header. Clients requesting application/msgpack or application/x-msgpack receive MessagePack-encoded responses instead of JSON:
from hawkapi.serialization.negotiation import negotiate_content_type, encode_for_content_type
# Automatic: clients send Accept: application/msgpack
# curl -H "Accept: application/msgpack" http://localhost:8000/data
# Manual usage in custom responses:
content_type = negotiate_content_type(request.headers.get("accept"))
body = encode_for_content_type(data, content_type)Both JSON and MessagePack encoders share the same enc_hook fallback for datetime, UUID, set, and bytes types.
from hawkapi import WebSocket
@app.websocket("/ws")
async def websocket_handler(ws: WebSocket):
await ws.accept()
async for message in ws:
await ws.send_text(f"Echo: {message}")Raise HTTP errors from anywhere with custom status, detail, and headers:
from hawkapi import HTTPException
@app.get("/items/{item_id:int}")
async def get_item(item_id: int):
item = await db.get(item_id)
if item is None:
raise HTTPException(404, detail="Item not found")
return item
@app.get("/admin")
async def admin():
raise HTTPException(
401,
detail="Token expired",
headers={"WWW-Authenticate": "Bearer"},
)Use hawkapi.status for named HTTP and WebSocket constants (FastAPI parity):
from hawkapi import status
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)Both HTTP (status.HTTP_200_OK, status.HTTP_422_UNPROCESSABLE_ENTITY, …) and WebSocket (status.WS_1008_POLICY_VIOLATION, …) constants are available.
@app.exception_handler(ValueError)
async def handle_value_error(request, exc):
return Response(
content=b'{"error": "bad value"}',
status_code=400,
content_type="application/json",
)from hawkapi import BackgroundTasks
@app.post("/notify")
async def notify(tasks: BackgroundTasks):
tasks.add_task(send_email, to="user@example.com", subject="Hello")
return {"status": "queued"}@app.on_startup
async def startup():
await db.connect()
@app.on_shutdown
async def shutdown():
await db.disconnect()app = HawkAPI(max_body_size=1024 * 1024) # 1 MB (default: 10 MB)
# Returns 413 Payload Too Large when exceededStream the request body in chunks without buffering the entire payload in memory. Useful for large file uploads:
@app.post("/upload")
async def upload(request: Request):
total = 0
async for chunk in request.stream():
total += len(chunk)
await process_chunk(chunk)
return {"bytes_received": total}Respects max_body_size and raises RequestEntityTooLarge if exceeded. Once streamed, calling request.body() raises RuntimeError; if body() was called first, stream() yields the cached body as a single chunk.
from hawkapi import Router
api = Router(prefix="/api/v1", tags=["api"])
@api.get("/health")
async def health():
return {"status": "ok"}
app.include_router(api)Class-based controllers:
from hawkapi import Controller, get, post
class UserController(Controller):
prefix = "/users"
tags = ["users"]
@get("/")
async def list_users(self):
return []
@post("/")
async def create_user(self, body: CreateUser):
return {"id": 1}
app.include_controller(UserController)Apply middleware to individual routes instead of the entire app. Pass a middleware= list of middleware classes (or (class, kwargs) tuples) to any route decorator:
from hawkapi.middleware.rate_limit import RateLimitMiddleware
@app.get("/public", middleware=[RateLimitMiddleware])
async def public():
return {"data": "rate-limited route"}
@app.post("/upload", middleware=[(RateLimitMiddleware, {"requests_per_second": 2.0})])
async def upload(body: UploadBody):
return {"status": "ok"}from hawkapi import StaticFiles
app.mount("/static", StaticFiles(directory="static"))
app.mount("/site", StaticFiles(directory="public", html=True))Runtime feature flags with zero external dependencies. Providers are swappable without restarting the app:
from hawkapi import HawkAPI, Depends
from hawkapi.flags import StaticFlagProvider, Flags, get_flags, requires_flag
app = HawkAPI(flags=StaticFlagProvider({"new_checkout": True}))
@app.get("/checkout")
async def checkout(flags: Flags = Depends(get_flags)) -> dict:
if await flags.bool("new_checkout", default=False):
return await new_flow()
return await old_flow()
@app.get("/beta/reports")
@requires_flag("beta.reports")
async def beta_reports() -> dict: ...Providers: StaticFlagProvider, EnvFlagProvider, FileFlagProvider (JSON/TOML/YAML with mtime hot-reload). Use when you need to gate features per environment or roll out incrementally without a deployment. See docs/guide/feature-flags.md.
Thin GraphQL-over-HTTP adapter with GraphiQL UI served to browsers. Zero runtime deps on the default path — graphql-core and strawberry are imported lazily only when the adapter is used:
from hawkapi.graphql.adapters import from_graphql_core
from graphql import build_schema
schema = build_schema("type Query { hello: String }")
app.mount_graphql("/graphql", executor=from_graphql_core(schema))Adapters: from_graphql_core, from_strawberry. Use when you need a GraphQL API alongside REST without adding a separate framework. See docs/guide/graphql.md.
Thin gRPC integration with ASGI lifespan-tied lifecycle — the gRPC server starts and stops with the app:
from myproto import greeter_pb2_grpc
class Greeter(greeter_pb2_grpc.GreeterServicer):
async def SayHello(self, request, context):
app = context.hawkapi_app # access app state
...
app.mount_grpc(
Greeter(),
add_to_server=greeter_pb2_grpc.add_GreeterServicer_to_server,
port=50051,
reflection=True,
reflection_service_names=["greeter.Greeter"],
)Built-in HawkAPIObservabilityInterceptor adds structured logs and Prometheus metrics to every RPC. TLS via ssl_credentials=. Install: pip install hawkapi[grpc]. Use when you need gRPC and HTTP APIs from a single process. See docs/guide/grpc.md.
Named async concurrency isolator — caps simultaneous in-flight requests per named pool to prevent one slow resource from exhausting the entire process:
from hawkapi.middleware.bulkhead import bulkhead
@app.get("/export")
@bulkhead("exports", max_concurrent=4)
async def export(): ...Context-manager form also supported. Backends: LocalBulkheadBackend (default, asyncio.Semaphore per name) and RedisBulkheadBackend (distributed, hash + lease-TTL). Optional Prometheus metrics. Use when you need hard concurrency caps per operation type. See docs/guide/bulkhead.md.
HawkAPI ships experimental cp313t-cp313t wheels built via cibuildwheel for the no-GIL CPython 3.13 free-threaded build (PEP 703):
from hawkapi._threading import FREE_THREADED, maybe_thread_lock, maybe_async_lock
# FREE_THREADED is True only on a no-GIL 3.13t interpreter
lock = maybe_thread_lock() # real threading.Lock on 3.13t, no-op elsewhere
alock = maybe_async_lock() # real asyncio.Lock on 3.13t, no-op elsewheremaybe_thread_lock() and maybe_async_lock() let library code be PEP 703-aware without branching on every call site. Use when running HawkAPI under python3.13t for CPU-bound workloads that benefit from true parallelism. See docs/guide/free-threaded.md.
Built-in Kubernetes-ready health endpoints:
app = HawkAPI(readyz_url="/readyz", livez_url="/livez")
@app.readiness_check("database")
async def check_db():
return await db.ping()/livez always returns 200. /readyz runs all registered checks and returns 200 or 503 with aggregated results.
from hawkapi.middleware.circuit_breaker import CircuitBreakerMiddleware
app.add_middleware(
CircuitBreakerMiddleware,
failure_threshold=5,
recovery_timeout=30.0,
half_open_max_calls=2,
)Three-state circuit breaker (CLOSED → OPEN → HALF_OPEN) with per-path tracking. Returns 503 with application/problem+json when the circuit is open.
from hawkapi.middleware.trusted_proxy import TrustedProxyMiddleware
app.add_middleware(
TrustedProxyMiddleware,
trusted_proxies=["10.0.0.0/8", "172.16.0.0/12"],
)Extracts real client IP, scheme, and host from X-Forwarded-For, X-Forwarded-Proto, X-Forwarded-Host — only from trusted CIDR ranges.
from hawkapi.middleware.request_limits import RequestLimitsMiddleware
app.add_middleware(
RequestLimitsMiddleware,
max_query_length=2048,
max_headers_count=100,
max_header_size=8192,
)Rejects oversized requests at the ASGI scope level before body parsing. Returns 414 (query) or 431 (headers).
Double-submit cookie CSRF protection. Safe methods (GET, HEAD, OPTIONS) pass through and receive a signed CSRF token cookie. Unsafe methods require the token in an X-CSRF-Token header or csrf_token form field:
from hawkapi.middleware.csrf import CSRFMiddleware
app.add_middleware(
CSRFMiddleware,
secret="your-secret-key",
cookie_name="csrftoken",
header_name="x-csrf-token",
cookie_secure=True,
cookie_samesite="lax",
)Returns 403 with application/problem+json when the token is missing or mismatched. Tokens are HMAC-SHA256 signed and verified with hmac.compare_digest for timing safety.
Signed cookie-based sessions using HMAC-SHA256. Session data is serialized with msgspec, base64url-encoded, and stored in a cookie with expiry checking:
from hawkapi.middleware.session import SessionMiddleware
app.add_middleware(
SessionMiddleware,
secret_key="your-secret-key",
session_cookie="session",
max_age=14 * 24 * 3600, # 14 days
)
@app.get("/dashboard")
async def dashboard(request: Request):
request.scope["session"]["views"] = request.scope["session"].get("views", 0) + 1
return {"views": request.scope["session"]["views"]}The cookie is only set when session data changes (dirty checking via snapshot comparison).
Distributed rate limiting backed by Redis using an atomic Lua-scripted token bucket. Survives restarts and works across multiple processes:
from hawkapi.middleware.rate_limit_redis import RedisRateLimitMiddleware
app.add_middleware(
RedisRateLimitMiddleware,
redis_url="redis://localhost:6379",
requests_per_second=10.0,
burst=20,
key_prefix="hawkapi:rl:",
)Falls back to in-memory rate limiting automatically when Redis is unavailable. Supports custom key functions for per-user or per-API-key limits.
Mark routes as deprecated with RFC 8594 Sunset headers:
@app.get("/v1/users", deprecated=True, sunset="2025-06-01", deprecation_link="https://docs.example.com/migration")
async def old_users():
return []Adds Deprecation: true, Sunset, and Link headers to responses automatically.
app = HawkAPI(observability=True)Every request gets: request ID, structured JSON logs, metrics, and OpenTelemetry spans (if installed).
HawkAPI implements W3C Trace Context propagation. Incoming traceparent and tracestate headers are parsed, validated, and propagated to responses. When OpenTelemetry is installed, it delegates to the OTel propagation API; otherwise it uses a built-in manual parser:
from hawkapi.observability.tracing import extract_context, inject_context
# Extract trace context from incoming request headers
ctx = extract_context(scope["headers"])
# ctx = {"trace_id": "4bf9...", "span_id": "00f0...", "trace_flags": "01", "tracestate": ""}
# Inject trace context into outgoing response headers
headers = inject_context(headers, ctx["trace_id"], ctx["span_id"])If no valid traceparent is present, new trace and span IDs are generated automatically.
app = HawkAPI(serverless=True)Disables documentation routes to minimize cold start. Combined with lazy imports, heavy modules load only on first use.
@app.get("/users", version="v1")
async def list_users_v1():
return [{"id": 1, "name": "Alice"}]
@app.get("/users", version="v2")
async def list_users_v2():
return [{"id": 1, "name": "Alice", "email": "alice@example.com"}]
# GET /v1/users -> v1 handler
# GET /v2/users -> v2 handlerUse VersionRouter to scope an entire router to a version:
from hawkapi.routing import VersionRouter
v2 = VersionRouter("v2", prefix="/api")
@v2.get("/users")
async def list_users(): # -> /v2/api/users
return []
app.include_router(v2)Compare two OpenAPI specs and detect breaking changes:
from hawkapi.openapi import detect_breaking_changes, format_report
old_spec = app.openapi(api_version="v1")
# ... make changes ...
new_spec = app.openapi(api_version="v1")
changes = detect_breaking_changes(old_spec, new_spec)
print(format_report(changes))Detects: path removed, method removed, required parameter added, parameter removed, parameter type changed, response field removed, status code changed.
from hawkapi.openapi.linter import lint, format_lint_report
issues = lint(app.openapi())
print(format_lint_report(issues))Built-in rules: operation-id-required, operation-summary-required, response-description-required. Custom rules are simple functions.
from hawkapi.openapi import detect_breaking_changes
from hawkapi.openapi.changelog import generate_changelog
changes = detect_breaking_changes(old_spec, new_spec)
print(generate_changelog(changes))Auto-generate test cases from your OpenAPI schema:
from hawkapi.testing.contract import generate_contract_tests
tests = generate_contract_tests(app)
for t in tests:
response = client.request(t.method, t.path)
assert response.status_code == t.expected_statusExtend HawkAPI behavior with plugins. The Plugin base class provides six hooks:
| Hook | When it fires |
|---|---|
on_route_registered(route) |
A route is registered; return the (possibly modified) route |
on_schema_generated(spec) |
OpenAPI schema is generated; return the enriched spec |
on_startup() |
Application startup |
on_shutdown() |
Application shutdown |
on_exception(request, exc) |
Unhandled exception, before the exception handler chain |
on_middleware_added(middleware_class, kwargs) |
A middleware is added to the application |
from hawkapi.plugins import Plugin
class AuditPlugin(Plugin):
def on_route_registered(self, route):
print(f"Route registered: {route.path}")
return route
def on_schema_generated(self, spec):
spec["info"]["x-audited"] = True
return spec
def on_startup(self):
print("App starting up")
def on_exception(self, request, exc):
log_to_sentry(exc)
app.add_plugin(AuditPlugin())# Development server with auto-reload
hawkapi dev app:app
hawkapi dev app:app --host 0.0.0.0 --port 3000
# Detect API breaking changes
hawkapi diff old_app:app new_app:app
# Lint OpenAPI spec
hawkapi check app:app
# Generate API changelog
hawkapi changelog old_app:app new_app:app
# Scaffold a new project
hawkapi new myproject
hawkapi new myproject --docker
# Initialize config files in current directory
hawkapi init
# Generate a TypeScript or Python client SDK from OpenAPI
hawkapi gen-client --app app:app --lang typescript --out ./client
hawkapi gen-client --app app:app --lang python --out ./client
# Migrate a FastAPI codebase to HawkAPI (in-place rewrite)
hawkapi migrate path/to/fastapi_project
hawkapi migrate path/to/fastapi_project --dry-run # preview diffs only
hawkapi migrate path/to/fastapi_project --convert-models # also rewrite Pydantic BaseModel → msgspec.Struct
hawkapi migrate path/to/fastapi_project --output ./migrated # write to separate directory
# Health-check a running app (18 rules across 5 categories)
hawkapi doctor app:app
hawkapi doctor app:app --format=json
hawkapi doctor app:app --severity=warn # only warnings and errorshawkapi init creates .env and .env.example files with commented-out HawkAPI configuration templates. Existing files are skipped.
hawkapi gen-client generates a zero-dependency client: native fetch for TypeScript, msgspec-backed for Python. Source can be a live app (--app module:attr) or a saved spec file (--spec openapi.json).
hawkapi migrate uses AST rewriting (libcst) to replace FastAPI imports, decorators, and patterns with their HawkAPI equivalents. See docs/guide/migration-from-fastapi.md.
hawkapi doctor lints a live app for security misconfigurations, missing observability middleware, performance anti-patterns, and outdated dependencies. Exits 0 (clean), 1 (warnings), or 2 (errors). See docs/guide/doctor.md.
from hawkapi import Settings, env_field
class AppSettings(Settings):
db_url: str = env_field("DATABASE_URL")
debug: bool = env_field("DEBUG", default=False)
port: int = env_field("PORT", default=8000)
allowed_hosts: list = env_field("ALLOWED_HOSTS", default=["*"])
settings = AppSettings.load(profile="production")Supports .env files and environment profiles (.env.development, .env.production).
Sync TestClient for pytest — no async needed:
from hawkapi.testing import TestClient, override
client = TestClient(app)
def test_hello():
response = client.get("/")
assert response.status_code == 200
assert response.json()["message"] == "Hello, World!"
def test_with_mock_db():
with override(app, Database, mock_db):
response = client.get("/users/1")
assert response.status_code == 200TestClient maintains a persistent cookie jar across requests. Set-Cookie headers from responses are automatically stored and sent on subsequent requests:
client = TestClient(app)
client.cookies["session"] = "abc123" # Pre-set cookies
response = client.get("/dashboard") # Cookie sent automatically
# Response Set-Cookie headers update the jar for future requestsTestResponse provides convenience properties for assertions:
response = client.get("/users")
assert response.is_success # 2xx status
assert not response.is_redirect # 3xx status
response.raise_for_status() # Raises HTTPStatusError for 4xx/5xx
assert response.text == "OK" # Body as UTF-8 string
assert "Content-Type" in response.headers # Case-insensitive header lookupResponse headers use CaseInsensitiveDict — lookups like response.headers["content-type"] and response.headers["Content-Type"] both work.
Note: The numbers below are local dev measurements on Apple M3 Pro. The canonical competitive comparison (Linux CI, head-to-head against FastAPI/Litestar/BlackSheep/Starlette/Sanic) is the table at the top of this README.
Tested on Apple M3 Pro, Python 3.13, msgspec 0.20. ASGI-level benchmarks (no HTTP server overhead — pure framework performance).
| Scenario | Latency | Throughput |
|---|---|---|
Simple JSON (GET /ping) |
35 us | ~28,500 req/s |
Path param (GET /users/42) |
39 us | ~25,600 req/s |
Body decode (POST /items) |
40 us | ~25,000 req/s |
| Large response (100 items) | 57 us | ~17,500 req/s |
| Payload | Ops/sec | vs stdlib json |
|---|---|---|
| Small dict (56 bytes) | 13.0M | 12.2x |
| 100-item list (8.1 KB) | 189K | 6.0x |
| 1000-item list (198 KB) | 8.7K | 6.1x |
Radix tree with 48 registered routes:
| Metric | Value |
|---|---|
| Lookups/sec | ~2,000,000 |
| Per lookup | ~486 ns |
Run benchmarks yourself:
python benchmarks/bench_request_response.py
python benchmarks/bench_serialization.py
python benchmarks/bench_routing.py
python benchmarks/bench_vs_fastapi.pygit clone https://github.com/Hawk-API/HawkAPI.git
cd HawkAPI
pip install -e ".[dev]"
# Run tests (1211 tests)
pytest
# Lint and type check
ruff check src/ tests/
pyright src/Add this badge to your project's README — the version auto-refreshes from PyPI:
Markdown:
[](https://github.com/Hawk-API/HawkAPI)reStructuredText:
.. image:: https://img.shields.io/pypi/v/hawkapi?label=Hawk&color=D97706
:target: https://github.com/Hawk-API/HawkAPI
:alt: HawkUsing style=for-the-badge instead of flat gives a larger, more prominent variant. The raw SVG icon lives at docs/assets/hawk-icon.svg.
The core stays small; everything else lives in optional hawkapi-* packages on PyPI. Each follows the same init_xxx(app, ...) + Depends(get_xxx) pattern.
| Package | Purpose |
|---|---|
hawkapi-sentry |
Sentry SDK — exceptions, traces, request context |
hawkapi-otel |
OpenTelemetry — traces, metrics, logs |
hawkapi-auth |
JWT (access + refresh), argon2id passwords, DI guards |
hawkapi-sqlalchemy |
Async SQLAlchemy 2.0, multi-DB routing, Alembic helper, fixtures |
hawkapi-cache |
Response caching with TTL + tag invalidation; in-memory + Redis |
hawkapi-storage |
Local / S3 / GCS / Azure with streaming + pre-signed URLs |
hawkapi-mail |
SMTP / SES / SendGrid / Mailgun / Resend, templates, outbox, webhooks |
hawkapi-celery |
Celery — async tasks, beat, context propagation, healthchecks |
hawkapi-websockets |
Connection manager, rooms, Redis pub/sub backplane, heartbeat |
hawkapi-mcp |
Model Context Protocol — expose routes as MCP tools for LLM agents |
hawkapi-admin |
Auto-generated CRUD admin UI for hawkapi-sqlalchemy models |
See docs/guide/plugins.md for the full reference and the upcoming-plugin roadmap.
MIT License. See LICENSE for details.