aiohttp-openapi-helper 0.2.0
Lightweight OpenAPI 3 helper for aiohttp
License: MIT
Requires Python: >=3.8

# aiohttp-openapi-helper

[![PyPI version](https://badge.fury.io/py/aiohttp-openapi-helper.svg)](https://badge.fury.io/py/aiohttp-openapi-helper)
[![PyPI downloads](https://img.shields.io/pypi/dm/aiohttp-openapi-helper.svg)](https://pypi.org/project/aiohttp-openapi-helper/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)
[![Python Versions](https://img.shields.io/pypi/pyversions/aiohttp-openapi-helper.svg)](https://pypi.org/project/aiohttp-openapi-helper/)
[![Build Status](https://img.shields.io/github/actions/workflow/status/alexdev/aiohttp-openapi-helper/test.yml?branch=main)](https://github.com/alexdev/aiohttp-openapi-helper/actions)
[![Coverage](https://img.shields.io/codecov/c/github/alexdev/aiohttp-openapi-helper)](https://codecov.io/gh/alexdev/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

| Feature | aiohttp-openapi-helper | aiohttp-apispec | aiohttp-swagger3 |
|---------|------------------------|-----------------|------------------|
| Decorator-based routes | ✅ | ❌ | ✅ |
| Auto spec generation | ✅ | ✅ | ✅ |
| Request validation | ✅ | ❌ | ✅ |
| Response validation | ✅ | ❌ | ✅ |
| Lightweight | ✅ | ⚠️ | ❌ |
| Type hints | ✅ | ⚠️ | ✅ |

## Installation

Install via pip:

```bash
pip install aiohttp-openapi-helper
```

## Quick Start
### Basic Usage

```python
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
```python
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:
```bash
pytest tests/ -v
```
Run with coverage:
```bash
pytest tests/ --cov=aiohttp_openapi_helper --cov-report=html
```

# Examples
Complete Example
```python
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)
```