aiohttp-openapi-helper

Lightweight OpenAPI 3.0 helper for aiohttp applications. Simplifies API development with decorator-based routes, automatic spec generation, and built-in validation.

Features

✨ Decorator-based Routes: Define OpenAPI specs directly on route handlers
🚀 Auto Spec Generation: Automatically generate OpenAPI 3.0 specification
✅ Request Validation: Validate incoming requests against schemas
📦 Response Validation: Validate outgoing responses (optional)
🔧 Flexible Middleware: Easy-to-configure validation middleware
📝 Type Hints: Full type annotations for better IDE support
⚡ Lightweight: Minimal dependencies, no heavy frameworks
📚 Well Documented: Comprehensive examples and API reference

Comparison

Installation

Install via pip:

pip install aiohttp-openapi-helper

Quick Start

Basic Usage

from aiohttp import web
from aiohttp_openapi_helper import openapi_route, generate_openapi_spec, OpenAPIMiddleware

# Create aiohttp application
app = web.Application()

# Add OpenAPI middleware for validation
app.middlewares.append(OpenAPIMiddleware().middleware)

# Define your routes with OpenAPI decorators
@openapi_route(
    app,
    "/users/{user_id}",
    method="GET",
    summary="Get user by ID",
    description="Retrieve user information by user ID",
    parameters=[
        {"name": "user_id", "in": "path", "required": True, "schema": {"type": "integer"}},
        {"name": "include_details", "in": "query", "schema": {"type": "boolean"}}
    ],
    responses={
        "200": {
            "description": "Successful response",
            "content": {
                "application/json": {
                    "schema": {
                        "type": "object",
                        "properties": {
                            "id": {"type": "integer"},
                            "name": {"type": "string"},
                            "email": {"type": "string"}
                        }
                    }
                }
            }
        },
        "404": {"description": "User not found"}
    }
)
async def get_user(request):
    user_id = int(request.match_info["user_id"])
    include_details = request.query.get("include_details") == "true"
    
    # Your business logic here
    user = {"id": user_id, "name": "John Doe", "email": "john@example.com"}
    
    return web.json_response(user)


@openapi_route(
    app,
    "/users",
    method="POST",
    summary="Create a new user",
    description="Create a new user in the system",
    request_body={
        "required": True,
        "content": {
            "application/json": {
                "schema": {
                    "type": "object",
                    "required": ["name", "email"],
                    "properties": {
                        "name": {"type": "string"},
                        "email": {"type": "string", "format": "email"},
                        "age": {"type": "integer"}
                    }
                }
            }
        }
    },
    responses={
        "201": {
            "description": "User created successfully",
            "content": {
                "application/json": {
                    "schema": {
                        "type": "object",
                        "properties": {
                            "id": {"type": "integer"},
                            "name": {"type": "string"},
                            "email": {"type": "string"}
                        }
                    }
                }
            }
        },
        "400": {"description": "Invalid input"}
    }
)
async def create_user(request):
    data = await request.json()
    
    # Validation is automatic by middleware
    # Your business logic here
    new_user = {
        "id": 123,
        "name": data["name"],
        "email": data["email"]
    }
    
    return web.json_response(new_user, status=201)


# Generate and serve OpenAPI spec
spec = generate_openapi_spec(
    title="User API",
    version="1.0.0",
    description="API for user management",
    routes=app.router.routes()
)

# Add route to serve the spec
app.router.add_get("/openapi.json", lambda r: web.json_response(spec))
app.router.add_get("/openapi.yaml", lambda r: web.Response(
    text=web.yaml.dump(spec),
    content_type="application/yaml"
))

# Run the application
if __name__ == "__main__":
    web.run_app(app, port=8080)

Advanced Configuration

from aiohttp import web
from aiohttp_openapi_helper import (
    openapi_route,
    generate_openapi_spec,
    OpenAPIMiddleware,
    security_scheme
)

app = web.Application()

# Configure middleware with custom options
middleware = OpenAPIMiddleware(
    validate_request=True,
    validate_response=True,
    raise_on_error=False,  # Return 400 instead of raising exception
    log_errors=True
)
app.middlewares.append(middleware.middleware)

# Define security schemes
api_key_scheme = security_scheme(
    type="apiKey",
    name="X-API-Key",
    in_="header"
)

bearer_scheme = security_scheme(
    type="http",
    scheme="bearer",
    bearer_format="JWT"
)

# Use security schemes in routes
@openapi_route(
    app,
    "/protected",
    method="GET",
    summary="Protected endpoint",
    security=[{"ApiKeyAuth": []}],
    responses={
        "200": {"description": "Successful response"},
        "401": {"description": "Unauthorized"}
    }
)
async def protected_route(request):
    return web.json_response({"message": "Access granted"})


# Generate spec with security schemes
spec = generate_openapi_spec(
    title="Secure API",
    version="1.0.0",
    security_schemes={
        "ApiKeyAuth": api_key_scheme,
        "BearerAuth": bearer_scheme
    },
    routes=app.router.routes()
)

API Reference

@openapi_route(app, path, method, **kwargs) Decorator to define OpenAPI-compliant routes.

Parameters:

app: aiohttp Application instance
path: Route path (e.g., "/users/{user_id}")
method: HTTP method ("GET", "POST", "PUT", "DELETE", etc.)
summary: Brief summary of the endpoint
description: Detailed description
tags: List of tags for grouping endpoints
parameters: List of parameter specifications
request_body: Request body schema (for POST/PUT/PATCH)
responses: Response schemas keyed by status code
security: Security requirements
deprecated: Mark as deprecated (bool)

generate_openapi_spec(title, version, **kwargs) Generate OpenAPI 3.0 specification from registered routes. Parameters:

title: API title
version: API version
description: API description
routes: aiohttp routes collection
security_schemes: Dictionary of security schemes
servers: List of server objects
tags: List of tag definitions
contact: Contact information
license: License information

Returns: Dictionary containing OpenAPI specification OpenAPIMiddleware(validate_request=True, validate_response=False, raise_on_error=True, log_errors=False) Middleware for request/response validation. Parameters:

validate_request: Enable request validation
validate_response: Enable response validation
raise_on_error: Raise exception on validation error (otherwise return 400)
log_errors: Log validation errors to console

security_scheme(type, **kwargs) Helper to create security scheme definitions. Parameters:

type: "apiKey", "http", "oauth2", or "openIdConnect"
name: Header/query/cookie name (for apiKey)
in_: "header", "query", or "cookie" (for apiKey)
scheme: HTTP scheme (for http type)
bearer_format: Bearer token format (for bearer scheme)
flows: OAuth2 flows configuration
open_id_connect_url: OpenID Connect URL

Testing

Run tests with pytest:

pytest tests/ -v

Run with coverage:

pytest tests/ --cov=aiohttp_openapi_helper --cov-report=html

Examples

Complete Example

from aiohttp import web
from aiohttp_openapi_helper import (
    openapi_route,
    generate_openapi_spec,
    OpenAPIMiddleware,
    security_scheme
)

async def init_app():
    app = web.Application()
    
    # Add middleware
    app.middlewares.append(OpenAPIMiddleware(
        validate_request=True,
        validate_response=False
    ).middleware)
    
    # Define routes
    @openapi_route(
        app,
        "/health",
        method="GET",
        summary="Health check",
        tags=["System"],
        responses={
            "200": {
                "description": "Service is healthy",
                "content": {
                    "application/json": {
                        "schema": {
                            "type": "object",
                            "properties": {
                                "status": {"type": "string"},
                                "version": {"type": "string"}
                            }
                        }
                    }
                }
            }
        }
    )
    async def health_check(request):
        return web.json_response({
            "status": "healthy",
            "version": "1.0.0"
        })
    
    @openapi_route(
        app,
        "/items",
        method="GET",
        summary="List all items",
        tags=["Items"],
        parameters=[
            {"name": "limit", "in": "query", "schema": {"type": "integer", "default": 10}},
            {"name": "offset", "in": "query", "schema": {"type": "integer", "default": 0}}
        ],
        responses={
            "200": {
                "description": "List of items",
                "content": {
                    "application/json": {
                        "schema": {
                            "type": "array",
                            "items": {
                                "type": "object",
                                "properties": {
                                    "id": {"type": "integer"},
                                    "name": {"type": "string"}
                                }
                            }
                        }
                    }
                }
            }
        }
    )
    async def list_items(request):
        limit = int(request.query.get("limit", 10))
        offset = int(request.query.get("offset", 0))
        
        items = [{"id": i, "name": f"Item {i}"} for i in range(offset, offset + limit)]
        
        return web.json_response(items)
    
    # Generate and serve spec
    spec = generate_openapi_spec(
        title="Demo API",
        version="1.0.0",
        description="Demonstration of aiohttp-openapi-helper",
        routes=app.router.routes(),
        tags=[
            {"name": "System", "description": "System endpoints"},
            {"name": "Items", "description": "Item management endpoints"}
        ]
    )
    
    app.router.add_get("/openapi.json", lambda r: web.json_response(spec))
    
    return app

if __name__ == "__main__":
    web.run_app(init_app(), port=8080)