Custom Routers

This guide explains how to create a custom router for integrating FastOpenAPI with a new web framework.

When to Create a Custom Router

Create a custom router when:

• You want to integrate FastOpenAPI with an unsupported framework
• You need custom behavior that differs from existing routers
• You're building a framework and want FastOpenAPI support

Architecture Overview

FastOpenAPI uses a three-tier composition-based adapter pattern:

BaseRouter (route registration, OpenAPI generation)
    ↓ inherits
BaseAdapter (request pipeline via composition)
    ↓ inherits
FrameworkRouter (framework-specific glue)

Your custom router needs to:

1. Inherit from BaseAdapter
2. Implement a request data extractor
3. Implement framework response building
4. Register routes with your framework

Step 1: Create Request Data Extractor

The extractor converts framework requests into FastOpenAPI's RequestData:

from typing import Any
from fastopenapi.core.types import FileUpload, RequestData
from fastopenapi.routers.extractors import BaseRequestDataExtractor

class MyFrameworkExtractor(BaseRequestDataExtractor):
    """Extract request data from MyFramework requests"""

    @classmethod
    def _get_path_params(cls, request: Any) -> dict:
        """Extract path parameters from URL"""
        # Example: request.match_info for aiohttp, request.path_params for starlette
        return getattr(request, "path_params", {})

    @classmethod
    def _get_query_params(cls, request: Any) -> dict:
        """Extract query string parameters"""
        query_params = {}
        for key in request.query:
            values = request.query.getall(key)
            query_params[key] = values[0] if len(values) == 1 else values
        return query_params

    @classmethod
    def _get_headers(cls, request: Any) -> dict:
        """Extract HTTP headers"""
        return dict(request.headers)

    @classmethod
    def _get_cookies(cls, request: Any) -> dict:
        """Extract cookies"""
        return dict(request.cookies)

    @classmethod
    def _get_body(cls, request: Any) -> dict | list | None:
        """Extract JSON body"""
        content_type = request.content_type or ""
        if "application/json" in content_type:
            return request.json()
        return {}

    @classmethod
    def _get_form_data(cls, request: Any) -> dict:
        """Extract form data"""
        return dict(request.form) if hasattr(request, "form") else {}

    @classmethod
    def _get_files(cls, request: Any) -> dict[str, FileUpload | list[FileUpload]]:
        """Extract uploaded files"""
        files = {}
        for key, file_obj in request.files.items():
            files[key] = FileUpload(
                filename=file_obj.filename,
                content_type=file_obj.content_type,
                size=file_obj.size,
                file=file_obj.file,
            )
        return files

For async frameworks, inherit from BaseAsyncRequestDataExtractor instead.

Step 2: Create the Router

from collections.abc import Callable
from fastopenapi.response.serializer import WireResponse
from fastopenapi.routers.base import BaseAdapter
from fastopenapi.routers.common import RequestEnvelope
from fastopenapi.openapi.ui import render_swagger_ui, render_redoc_ui

# Import your framework
from myframework import App, Response as FrameworkResponse


class MyFrameworkRouter(BaseAdapter):
    """Router for MyFramework"""

    # Path conversion pattern: {param} -> framework format
    # Examples:
    #   Flask: (r"{(\w+)}", r"<\1>")      -> /users/<user_id>
    #   Starlette: (r"{(\w+)}", r"{\1}")  -> /users/{user_id}
    PATH_CONVERSIONS = (r"{(\w+)}", r"<\1>")

    # Reject async endpoints at registration time (sync-only router).
    # BaseAdapter.add_route raises TypeError when this is set and an
    # async def endpoint is registered. Leave as None for async routers.
    ASYNC_ENDPOINT_ERROR = "cannot be used with MyFramework (sync router)."

    # Set the extractor class
    extractor_cls = MyFrameworkExtractor

    def add_route(self, path: str, method: str, endpoint: Callable):
        """Register route with the framework"""
        # First, call parent to store route info (this also enforces
        # ASYNC_ENDPOINT_ERROR for async endpoints)
        super().add_route(path, method, endpoint)

        if self.app is not None:
            # Convert path format for framework
            framework_path = self._convert_path_for_framework(path)

            # Create view function wrapper
            def view_func(**path_params):
                # Create request envelope
                env = RequestEnvelope(
                    request=get_current_request(),  # Framework-specific
                    path_params=path_params
                )

                # Delegate to BaseAdapter's request handler
                return self.handle_request(endpoint, env)

            # Register with framework
            self.app.add_route(framework_path, view_func, methods=[method])

    def build_framework_response(self, response: WireResponse) -> FrameworkResponse:
        """Wrap the finalized wire triple into a framework response.

        `WireResponse` already carries the encoded body (`bytes | None`),
        the status, and the complete headers (Content-Type included) — the
        adapter must not re-serialize or guess a content type.
        """
        return FrameworkResponse(
            body=response.body if response.body is not None else b"",
            status=response.status,
            headers=response.headers,
        )

    def is_framework_response(self, response) -> bool:
        """Check if response is already a framework response"""
        return isinstance(response, FrameworkResponse)

    def _register_docs_endpoints(self):
        """Register OpenAPI documentation endpoints"""
        app = self.app

        @app.route(self.openapi_url, methods=["GET"])
        def openapi_json():
            return FrameworkResponse(
                body=self.openapi,
                content_type="application/json"
            )

        @app.route(self.docs_url, methods=["GET"])
        def swagger_ui():
            html = render_swagger_ui(self.openapi_url)
            return FrameworkResponse(body=html, content_type="text/html")

        @app.route(self.redoc_url, methods=["GET"])
        def redoc():
            html = render_redoc_ui(self.openapi_url)
            return FrameworkResponse(body=html, content_type="text/html")

Step 3: Async Router (Optional)

For async frameworks:

from fastopenapi.routers.extractors import BaseAsyncRequestDataExtractor


class MyAsyncExtractor(BaseAsyncRequestDataExtractor):
    """Async extractor for MyFramework"""

    @classmethod
    async def _get_body(cls, request) -> dict | list | None:
        """Async body extraction"""
        if request.content_type == "application/json":
            return await request.json()
        return {}

    @classmethod
    async def _get_files(cls, request) -> dict[str, FileUpload | list[FileUpload]]:
        """Async file extraction"""
        files = {}
        async for field in request.multipart():
            if field.filename:
                content = await field.read()
                files[field.name] = FileUpload(
                    filename=field.filename,
                    content_type=field.content_type,
                    size=len(content),
                    file=content,
                )
        return files


class MyAsyncRouter(BaseAdapter):
    """Async router for MyFramework"""

    extractor_async_cls = MyAsyncExtractor

    def add_route(self, path: str, method: str, endpoint: Callable):
        """Register async route"""
        super().add_route(path, method, endpoint)

        if self.app is not None:
            framework_path = self._convert_path_for_framework(path)

            async def view_func(request):
                env = RequestEnvelope(
                    request=request,
                    path_params=request.match_info
                )
                # Use async handler
                return await self.handle_request_async(endpoint, env)

            self.app.router.add_route(method, framework_path, view_func)

    # ... implement other methods

Complete Example: Sync Router

Here's a complete example based on Flask's pattern:

from collections.abc import Callable
from typing import Any

from fastopenapi.core.types import FileUpload
from fastopenapi.response.serializer import WireResponse
from fastopenapi.routers.base import BaseAdapter
from fastopenapi.routers.common import RequestEnvelope
from fastopenapi.routers.extractors import BaseRequestDataExtractor
from fastopenapi.openapi.ui import render_swagger_ui, render_redoc_ui

# Framework imports
from myframework import App, Request, Response as MFResponse


class MyFrameworkExtractor(BaseRequestDataExtractor):
    @classmethod
    def _get_path_params(cls, request: Any) -> dict:
        return getattr(request, "path_params", {})

    @classmethod
    def _get_query_params(cls, request: Any) -> dict:
        return dict(request.args)

    @classmethod
    def _get_headers(cls, request: Any) -> dict:
        return dict(request.headers)

    @classmethod
    def _get_cookies(cls, request: Any) -> dict:
        return dict(request.cookies)

    @classmethod
    def _get_body(cls, request: Any) -> dict | list | None:
        if request.is_json:
            return request.json
        return {}

    @classmethod
    def _get_form_data(cls, request: Any) -> dict:
        return dict(request.form)

    @classmethod
    def _get_files(cls, request: Any) -> dict[str, FileUpload | list[FileUpload]]:
        files = {}
        for key, file_obj in request.files.items():
            files[key] = FileUpload(
                filename=file_obj.filename,
                content_type=file_obj.content_type,
                size=None,
                file=file_obj,
            )
        return files


class MyFrameworkRouter(BaseAdapter):
    """Custom router for MyFramework"""

    PATH_CONVERSIONS = (r"{(\w+)}", r"<\1>")
    # Sync-only router: async endpoints are rejected at registration time
    ASYNC_ENDPOINT_ERROR = "cannot be used with MyFrameworkRouter (sync)."
    extractor_cls = MyFrameworkExtractor

    def add_route(self, path: str, method: str, endpoint: Callable):
        super().add_route(path, method, endpoint)

        if self.app is not None:
            framework_path = self._convert_path_for_framework(path)

            def view_func(**path_params):
                from myframework import current_request
                env = RequestEnvelope(
                    request=current_request,
                    path_params=path_params
                )
                return self.handle_request(endpoint, env)

            endpoint_name = f"{endpoint.__name__}:{method}:{framework_path}"
            self.app.add_url_rule(
                framework_path,
                endpoint_name,
                view_func,
                methods=[method]
            )

    def build_framework_response(self, response: WireResponse) -> MFResponse:
        # response.body is already-encoded bytes; headers include Content-Type
        mf_response = MFResponse(response.body if response.body is not None else b"")
        mf_response.status_code = response.status
        for key, value in response.headers.items():
            mf_response.headers[key] = value
        return mf_response

    def is_framework_response(self, response) -> bool:
        return isinstance(response, MFResponse)

    def _register_docs_endpoints(self):
        @self.app.route(self.openapi_url)
        def openapi_json():
            return MFResponse(self.openapi, mimetype="application/json")

        @self.app.route(self.docs_url)
        def swagger_docs():
            html = render_swagger_ui(self.openapi_url)
            return MFResponse(html, mimetype="text/html")

        @self.app.route(self.redoc_url)
        def redoc_docs():
            html = render_redoc_ui(self.openapi_url)
            return MFResponse(html, mimetype="text/html")

Usage

from myframework import App
from pydantic import BaseModel

app = App()
router = MyFrameworkRouter(
    app=app,
    title="My API",
    version="1.0.0"
)

class Item(BaseModel):
    name: str
    price: float

@router.get("/items/{item_id}")
def get_item(item_id: int):
    return {"id": item_id, "name": "Test"}

@router.post("/items", response_model=Item, status_code=201)
def create_item(item: Item):
    return item

Testing Your Router

def test_custom_router():
    app = App()
    router = MyFrameworkRouter(app=app)

    @router.get("/test")
    def test_endpoint():
        return {"status": "ok"}

    client = app.test_client()
    response = client.get("/test")
    assert response.status_code == 200
    assert response.json == {"status": "ok"}

Key Classes Reference

RouteInfo

Container for route information (from fastopenapi.core.router):

class RouteInfo:
    path: str         # Route path, e.g., "/users/{user_id}"
    method: str       # HTTP method (GET, POST, etc.)
    endpoint: Callable  # The endpoint function
    meta: dict        # Route metadata (response_model, status_code, tags, etc.)

RequestEnvelope

Wrapper for framework request (from fastopenapi.routers.common):

@dataclass(slots=True, frozen=True)
class RequestEnvelope:
    path_params: dict[str, str]  # Path parameters extracted by framework
    request: Any | None          # Framework's request object

Response

FastOpenAPI's response container (from fastopenapi.core.types). This is what endpoints return and what ResponseBuilder produces before serialization:

class Response:
    content: Any           # Response body (Python object, not yet encoded)
    status_code: int       # HTTP status code
    headers: dict[str, str] # Response headers

WireResponse

The finalized, wire-level response your adapter actually receives in build_framework_response (from fastopenapi.response.serializer). The body is already encoded and the headers already include Content-Type — the adapter only wraps it, it does not serialize:

@dataclass
class WireResponse:
    body: bytes | None      # Already-encoded body (None for 204/304)
    status: int             # HTTP status code
    headers: dict[str, str] # Complete headers, Content-Type included

Best Practices

1. Inherit from BaseAdapter - Don't reinvent the request handling pipeline
2. Use composition - Let BaseAdapter handle parameter resolution and validation
3. Handle async properly - Use handle_request_async() for async endpoints
4. Convert paths correctly - Set PATH_CONVERSIONS for your framework's format
5. Test thoroughly - Test all parameter types, validation, and error handling

Submitting to FastOpenAPI

If you create a router for a popular framework:

1. Follow the existing router patterns (see fastopenapi/routers/flask/)
2. Add comprehensive tests (see tests/routers/flask/)
3. Add framework documentation (see docs/frameworks/)
4. Submit a pull request

Next Steps

• Architecture - Understand FastOpenAPI internals
• Testing - Test your custom router
• OpenAPI Customization - Customize schema generation