Header versioning for FastAPI using arbitrary header.
import uvicorn

from examples.basic.no_version.routes import router
from examples.basic.v1.routes import router as router_v1
from examples.basic.v2.routes import router as router_v2
from fastapi_header_versioning import HeaderRoutingFastAPI
from fastapi_header_versioning.openapi import doc_generation

app = HeaderRoutingFastAPI(version_header="x-version", title="Versioned app")
app.include_router(router_v1)
app.include_router(router_v2)
app.include_router(router)
app = doc_generation(app)

uvicorn.Config(
    app=app,
    proxy_headers=True,
    access_log=False,
)

if __name__ == "__main__":
    uvicorn.run(
        app=app,
        port=9999,
        reload=False,
    )

examples/basic/app.py
tikon93/fastapi-header-versioning
from fastapi import Header


def get_version(x_version: str = Header(None)) -> str:
    return x_version

examples/basic/commons.py
tikon93/fastapi-header-versioning
from fastapi.routing import APIRouter

router = APIRouter()


@router.get("/hello")
async def greeting():
    return {"greeting": "Hi! It's not versioned route."}

examples/basic/no_version/routes.py
tikon93/fastapi-header-versioning
import uuid

from fastapi import Depends, Response, status

from examples.basic.commons import get_version
from fastapi_header_versioning import HeaderVersionedAPIRouter

from .schemas import GetResponse, Item, ItemResponse

router = HeaderVersionedAPIRouter()


async def get_parameters(skip: int = 0, limit: int = 100):
    return {"skip": skip, "limit": limit}


@router.get("/item/{item_id}", response_model=GetResponse)
@router.version("1")
async def get_item(
    item_id: uuid.UUID,
    query_parameter: str,
    dependency_parameters: dict = Depends(get_parameters),
    version: str = Depends(get_version),
):
    return {
        "query_parameter": query_parameter,
        "id": item_id,
        "dependency_parameters": dependency_parameters,
        "version": version,
    }


@router.post("/item", response_model=ItemResponse)
@router.version("1")
async def create_item(item: Item):
    created_item = item.dict()
    created_item["id"] = uuid.uuid4()
    return created_item


@router.patch("/item/{item_id}", response_model=ItemResponse)
@router.version("1")
async def update_item(item_id: uuid.UUID, item: Item):
    return {"id": item_id, "name": "base_name", "description": "base_description"}.update(
        item.dict(exclude_none=True),
    )


@router.put("/item/{item_id}", response_model=ItemResponse)
@router.version("1")
async def replace_item(item_id: uuid.UUID, item: Item):
    replaced_item = item.dict()
    replaced_item["id"] = item_id

    return replaced_item


@router.delete("/item/{item_id}")
@router.version("1")
async def delete_item(item_id: uuid.UUID):
    return Response(status_code=status.HTTP_204_NO_CONTENT)

examples/basic/v1/routes.py
tikon93/fastapi-header-versioning
import uuid

from pydantic import BaseModel


class Item(BaseModel):
    name: str
    description: str


class ItemResponse(Item):
    id: uuid.UUID


class ItemPatch(BaseModel):
    name: str | None = None
    description: str | None = None


class GetResponse(BaseModel):
    dependency_parameters: dict
    query_parameter: str
    id: uuid.UUID
    version: str

examples/basic/v1/schemas.py
tikon93/fastapi-header-versioning
import uuid

from fastapi import Depends, Response, status

from fastapi_header_versioning import HeaderVersionedAPIRouter

from .schemas import GetResponse, Item, ItemResponse

router = HeaderVersionedAPIRouter()


async def get_parameters(offset: int = 0, limit: int = 100):
    # parameters set differs from v1
    return {"offset": offset, "limit": limit}


@router.get("/items")
@router.version("2")
async def get_items():
    return {"items": "items"}


@router.get("/item/{item_id}", response_model=GetResponse)
@router.version("2")
async def get_item(
    item_id: uuid.UUID,
    new_query_parameter: str,
    dependency_parameters: dict = Depends(get_parameters),
):
    return {
        "new_query_parameter": new_query_parameter,
        "id": item_id,
        "dependency_parameters": dependency_parameters,
    }


@router.post("/item", response_model=ItemResponse)
@router.version("2")
async def create_item(item: Item):
    created_item = item.dict()
    created_item["id"] = uuid.uuid4()
    return created_item


@router.patch("/item/{item_id}", response_model=ItemResponse)
@router.version("2")
async def update_item(item_id: uuid.UUID, item: Item):
    return {"id": item_id, "new_field_for_name": "base_name", "description": "base_description"}.update(
        item.dict(exclude_none=True),
    )


@router.put("/item/{item_id}", response_model=ItemResponse)
@router.version("2")
async def replace_item(item_id: uuid.UUID, item: Item):
    replaced_item = item.dict()
    replaced_item["id"] = item_id
    return replaced_item


@router.delete("/item/{item_id}")
@router.version("2")
async def delete_item(item_id: uuid.UUID):
    return Response(status_code=status.HTTP_204_NO_CONTENT)

examples/basic/v2/routes.py
tikon93/fastapi-header-versioning
import uuid

from pydantic import BaseModel


class Item(BaseModel):
    new_field_for_name: str
    description: str


class ItemResponse(Item):
    id: uuid.UUID


class ItemPatch(BaseModel):
    new_field_for_name: str | None = None
    description: str | None = None


class GetResponse(BaseModel):
    dependency_parameters: dict
    new_query_parameter: str
    id: uuid.UUID

examples/basic/v2/schemas.py
tikon93/fastapi-header-versioning
import uvicorn

from examples.router_level_versions.no_version.routes import router
from examples.router_level_versions.v1.routes import router as router_v1
from examples.router_level_versions.v2.routes import router as router_v2
from fastapi_header_versioning import HeaderRoutingFastAPI
from fastapi_header_versioning.openapi import doc_generation
from fastapi_header_versioning.routing import HeaderVersionedAPIRouter

app = HeaderRoutingFastAPI(version_header="x-version", title="Versioned app")
router_override = HeaderVersionedAPIRouter(default_version="foo")

router_override.include_router(router)


app.include_router(router_override)
app.include_router(router_v1)
app.include_router(router_v2)
app = doc_generation(app)

uvicorn.Config(
    app=app,
    proxy_headers=True,
    access_log=False,
)

if __name__ == "__main__":
    uvicorn.run(
        app=app,
        port=9999,
        reload=False,
    )

examples/router_level_versions/app.py
tikon93/fastapi-header-versioning
from fastapi import Header


def get_version(x_version: str = Header(None)) -> str:
    return x_version

examples/router_level_versions/commons.py
tikon93/fastapi-header-versioning
from fastapi.routing import APIRouter

router = APIRouter()


@router.get("/hello")
async def greeting():
    return {"greeting": "Hi! It's not versioned route."}

examples/router_level_versions/no_version/routes.py
tikon93/fastapi-header-versioning
import uuid

from fastapi import Depends, Response, status

from fastapi_header_versioning import HeaderVersionedAPIRouter

from ..commons import get_version
from .schemas import GetResponse, Item, ItemResponse

router = HeaderVersionedAPIRouter(default_version="1")


async def get_parameters(skip: int = 0, limit: int = 100):
    return {"skip": skip, "limit": limit}


@router.get("/item/{item_id}", response_model=GetResponse)
async def get_item(
    item_id: uuid.UUID,
    query_parameter: str,
    dependency_parameters: dict = Depends(get_parameters),
    version: str = Depends(get_version),
):
    return {
        "query_parameter": query_parameter,
        "id": item_id,
        "dependency_parameters": dependency_parameters,
        "version": version,
    }


@router.post("/item", response_model=ItemResponse)
async def create_item(item: Item):
    created_item = item.dict()
    created_item["id"] = uuid.uuid4()
    return created_item


@router.patch("/item/{item_id}", response_model=ItemResponse)
async def update_item(item_id: uuid.UUID, item: Item):
    return {"id": item_id, "name": "base_name", "description": "base_description"}.update(
        item.dict(exclude_none=True),
    )


@router.put("/item/{item_id}", response_model=ItemResponse)
async def replace_item(item_id: uuid.UUID, item: Item):
    replaced_item = item.dict()
    replaced_item["id"] = item_id

    return replaced_item


@router.delete("/item/{item_id}")
async def delete_item(item_id: uuid.UUID):
    return Response(status_code=status.HTTP_204_NO_CONTENT)

examples/router_level_versions/v1/routes.py
tikon93/fastapi-header-versioning
import uuid

from pydantic import BaseModel


class Item(BaseModel):
    name: str
    description: str


class ItemResponse(Item):
    id: uuid.UUID


class ItemPatch(BaseModel):
    name: str | None = None
    description: str | None = None


class GetResponse(BaseModel):
    dependency_parameters: dict
    query_parameter: str
    id: uuid.UUID
    version: str

examples/router_level_versions/v1/schemas.py
tikon93/fastapi-header-versioning
import uuid

from fastapi import Depends, Response, status

from fastapi_header_versioning import HeaderVersionedAPIRouter

from .schemas import GetResponse, Item, ItemResponse

router = HeaderVersionedAPIRouter(default_version="2")


async def get_parameters(offset: int = 0, limit: int = 100):
    # parameters set differs from v1
    return {"offset": offset, "limit": limit}


@router.get("/item/{item_id}", response_model=GetResponse)
async def get_item(
    item_id: uuid.UUID,
    new_query_parameter: str,
    dependency_parameters: dict = Depends(get_parameters),
):
    return {
        "new_query_parameter": new_query_parameter,
        "id": item_id,
        "dependency_parameters": dependency_parameters,
    }


@router.post("/item", response_model=ItemResponse)
async def create_item(item: Item):
    created_item = item.dict()
    created_item["id"] = uuid.uuid4()
    return created_item


@router.patch("/item/{item_id}", response_model=ItemResponse)
async def update_item(item_id: uuid.UUID, item: Item):
    return {"id": item_id, "new_field_for_name": "base_name", "description": "base_description"}.update(
        item.dict(exclude_none=True),
    )


@router.put("/item/{item_id}", response_model=ItemResponse)
async def replace_item(item_id: uuid.UUID, item: Item):
    replaced_item = item.dict()
    replaced_item["id"] = item_id
    return replaced_item


@router.delete("/item/{item_id}")
async def delete_item(item_id: uuid.UUID):
    return Response(status_code=status.HTTP_204_NO_CONTENT)

examples/router_level_versions/v2/routes.py
tikon93/fastapi-header-versioning
import uuid

from pydantic import BaseModel


class Item(BaseModel):
    new_field_for_name: str
    description: str


class ItemResponse(Item):
    id: uuid.UUID


class ItemPatch(BaseModel):
    new_field_for_name: str | None = None
    description: str | None = None


class GetResponse(BaseModel):
    dependency_parameters: dict
    new_query_parameter: str
    id: uuid.UUID

examples/router_level_versions/v2/schemas.py
tikon93/fastapi-header-versioning
import uvicorn

from examples.unversioned_routers_wrapping.no_version.routes import router
from examples.unversioned_routers_wrapping.v1.routes import router as router_v1
from examples.unversioned_routers_wrapping.v2.routes import router as router_v2
from fastapi_header_versioning import HeaderRoutingFastAPI, HeaderVersionedAPIRouter
from fastapi_header_versioning.openapi import doc_generation

app = HeaderRoutingFastAPI(version_header="x-version", title="Versioned app")

root_router = HeaderVersionedAPIRouter()
root_router.include_router(router_v1, version="1")
root_router.include_router(router_v2, version="2")
root_router.include_router(router)

app.include_router(root_router)
app = doc_generation(app)

uvicorn.Config(
    app=app,
    proxy_headers=True,
    access_log=False,
)

if __name__ == "__main__":
    uvicorn.run(
        app=app,
        port=9999,
        reload=False,
    )

examples/unversioned_routers_wrapping/app.py
tikon93/fastapi-header-versioning
from fastapi import Header


def get_version(x_version: str = Header(None)) -> str:
    return x_version

examples/unversioned_routers_wrapping/commons.py
tikon93/fastapi-header-versioning
from fastapi.routing import APIRouter

router = APIRouter()


@router.get("/hello")
async def greeting():
    return {"greeting": "Hi! It's not versioned route."}

examples/unversioned_routers_wrapping/no_version/routes.py
tikon93/fastapi-header-versioning
import uuid

from fastapi import APIRouter, Depends, Response, status

from examples.basic.commons import get_version

from .schemas import GetResponse, Item, ItemResponse

router = APIRouter()


async def get_parameters(skip: int = 0, limit: int = 100):
    return {"skip": skip, "limit": limit}


@router.get("/item/{item_id}", response_model=GetResponse)
async def get_item(
    item_id: uuid.UUID,
    query_parameter: str,
    dependency_parameters: dict = Depends(get_parameters),
    version: str = Depends(get_version),
):
    return {
        "query_parameter": query_parameter,
        "id": item_id,
        "dependency_parameters": dependency_parameters,
        "version": version,
    }


@router.post("/item", response_model=ItemResponse)
async def create_item(item: Item):
    created_item = item.dict()
    created_item["id"] = uuid.uuid4()
    return created_item


@router.patch("/item/{item_id}", response_model=ItemResponse)
async def update_item(item_id: uuid.UUID, item: Item):
    return {"id": item_id, "name": "base_name", "description": "base_description"}.update(
        item.dict(exclude_none=True),
    )


@router.put("/item/{item_id}", response_model=ItemResponse)
async def replace_item(item_id: uuid.UUID, item: Item):
    replaced_item = item.dict()
    replaced_item["id"] = item_id

    return replaced_item


@router.delete("/item/{item_id}")
async def delete_item(item_id: uuid.UUID):
    return Response(status_code=status.HTTP_204_NO_CONTENT)

examples/unversioned_routers_wrapping/v1/routes.py
tikon93/fastapi-header-versioning
import uuid

from pydantic import BaseModel


class Item(BaseModel):
    name: str
    description: str


class ItemResponse(Item):
    id: uuid.UUID


class ItemPatch(BaseModel):
    name: str | None = None
    description: str | None = None


class GetResponse(BaseModel):
    dependency_parameters: dict
    query_parameter: str
    id: uuid.UUID
    version: str

examples/unversioned_routers_wrapping/v1/schemas.py
tikon93/fastapi-header-versioning
import uuid

from fastapi import APIRouter, Depends, Response, status

from .schemas import GetResponse, Item, ItemResponse

router = APIRouter()


async def get_parameters(offset: int = 0, limit: int = 100):
    # parameters set differs from v1
    return {"offset": offset, "limit": limit}


@router.get("/item/{item_id}", response_model=GetResponse)
async def get_item(
    item_id: uuid.UUID,
    new_query_parameter: str,
    dependency_parameters: dict = Depends(get_parameters),
):
    return {
        "new_query_parameter": new_query_parameter,
        "id": item_id,
        "dependency_parameters": dependency_parameters,
    }


@router.post("/item", response_model=ItemResponse)
async def create_item(item: Item):
    created_item = item.dict()
    created_item["id"] = uuid.uuid4()
    return created_item


@router.patch("/item/{item_id}", response_model=ItemResponse)
async def update_item(item_id: uuid.UUID, item: Item):
    return {"id": item_id, "new_field_for_name": "base_name", "description": "base_description"}.update(
        item.dict(exclude_none=True),
    )


@router.put("/item/{item_id}", response_model=ItemResponse)
async def replace_item(item_id: uuid.UUID, item: Item):
    replaced_item = item.dict()
    replaced_item["id"] = item_id
    return replaced_item


@router.delete("/item/{item_id}")
async def delete_item(item_id: uuid.UUID):
    return Response(status_code=status.HTTP_204_NO_CONTENT)

examples/unversioned_routers_wrapping/v2/routes.py
tikon93/fastapi-header-versioning
import uuid

from pydantic import BaseModel


class Item(BaseModel):
    new_field_for_name: str
    description: str


class ItemResponse(Item):
    id: uuid.UUID


class ItemPatch(BaseModel):
    new_field_for_name: str | None = None
    description: str | None = None


class GetResponse(BaseModel):
    dependency_parameters: dict
    new_query_parameter: str
    id: uuid.UUID

examples/unversioned_routers_wrapping/v2/schemas.py
tikon93/fastapi-header-versioning
from .fastapi import HeaderRoutingFastAPI
from .routing import HeaderVersionedAPIRoute, HeaderVersionedAPIRouter

__all__ = [
    "HeaderRoutingFastAPI",
    "HeaderVersionedAPIRoute",
    "HeaderVersionedAPIRouter",
]

fastapi_header_versioning/__init__.py
tikon93/fastapi-header-versioning
from collections.abc import Callable, Sequence
from typing import Any, Optional, Union

from fastapi import Depends, FastAPI
from fastapi.applications import AppType
from fastapi.datastructures import Default
from fastapi.routing import APIRoute
from fastapi.utils import generate_unique_id
from starlette.responses import JSONResponse, Response
from starlette.routing import BaseRoute
from starlette.types import Lifespan, Receive, Scope, Send

from .routing import HeaderVersionedAPIRouter


class CustomHeaderVersionMiddleware:
    def __init__(
        self,
        app: FastAPI,
        version_header: str,
    ) -> None:
        self.app = app
        self.version_header = version_header.encode()

    async def __call__(
        self,
        scope: Scope,
        receive: Receive,
        send: Send,
    ) -> None:
        if scope["type"] in ("http", "websocket"):
            headers: dict[bytes, bytes] = dict(scope["headers"])
            scope["requested_version"] = None
            if self.version_header in headers:
                scope["requested_version"] = headers[self.version_header].decode()

        return await self.app(scope, receive, send)


class HeaderRoutingFastAPI(FastAPI):
    def __init__(
        self,
        version_header: str,
        routes: Optional[list[BaseRoute]] = None,
        dependencies: Optional[Sequence[Depends]] = None,
        default_response_class: type[Response] = Default(JSONResponse),
        on_startup: Optional[Sequence[Callable[[], Any]]] = None,
        on_shutdown: Optional[Sequence[Callable[[], Any]]] = None,
        lifespan: Optional[Lifespan[AppType]] = None,
        responses: Optional[dict[Union[int, str], dict[str, Any]]] = None,
        callbacks: Optional[list[BaseRoute]] = None,
        deprecated: Optional[bool] = None,
        include_in_schema: bool = True,
        generate_unique_id_function: Callable[[APIRoute], str] = Default(
            generate_unique_id,
        ),
        *args: Any,
        **kwargs: Any,
    ):
        super().__init__(
            *args,
            routes=routes,
            dependencies=dependencies,
            on_startup=on_startup,
            on_shutdown=on_shutdown,
            lifespan=lifespan,
            default_response_class=default_response_class,
            callbacks=callbacks,
            deprecated=deprecated,
            include_in_schema=include_in_schema,
            responses=responses,
            generate_unique_id_function=generate_unique_id_function,
            **kwargs,
        )
        self.router = HeaderVersionedAPIRouter(
            routes=routes,
            dependency_overrides_provider=self,
            on_startup=on_startup,
            on_shutdown=on_shutdown,
            lifespan=lifespan,
            default_response_class=default_response_class,
            dependencies=dependencies,
            callbacks=callbacks,
            deprecated=deprecated,
            include_in_schema=include_in_schema,
            responses=responses,
            generate_unique_id_function=generate_unique_id_function,
        )
        self.add_middleware(
            CustomHeaderVersionMiddleware,
            version_header=version_header,
        )

fastapi_header_versioning/fastapi.py
tikon93/fastapi-header-versioning
from collections import defaultdict

from fastapi import FastAPI
from fastapi.routing import APIRoute
from starlette.routing import BaseRoute

from .fastapi import HeaderRoutingFastAPI
from .routing import HeaderVersionedAPIRoute


def get_version_from_route(route: BaseRoute) -> str | None:
    if isinstance(route, HeaderVersionedAPIRoute):
        return route.api_version or None

    return None


def doc_generation(
    app: HeaderRoutingFastAPI,
) -> HeaderRoutingFastAPI:
    parent_app = app
    version_route_mapping: dict[str | None, list[BaseRoute]] = defaultdict(list)

    for route in app.routes:
        version = get_version_from_route(route)  # pyright: ignore[reportGeneralTypeIssues]
        version_route_mapping[version].append(route)  # pyright: ignore[reportGeneralTypeIssues]

    versions = version_route_mapping.keys()
    for version in versions:
        unique_routes = {}
        version_description = version if version is not None else "Not versioned"
        versioned_app = FastAPI(
            title=app.title,
            description=version_description + " " + app.description,
        )
        for route in version_route_mapping[version]:
            if isinstance(route, APIRoute):
                for method in route.methods:
                    unique_routes[route.path + "|" + method] = route

            # TODO: support websocket routes

        versioned_app.router.routes.extend(unique_routes.values())

        prefix = f"/version_{version}"
        if version is None:
            prefix = "/no_version"
        parent_app.mount(prefix, versioned_app)

    return parent_app

fastapi_header_versioning/openapi.py
tikon93/fastapi-header-versioning
from collections.abc import Callable, Sequence
from enum import Enum
from functools import cache
from typing import (
    Any,
    Optional,
    TypeVar,
    Union,
)

from fastapi import APIRouter, params
from fastapi.datastructures import Default
from fastapi.routing import APIRoute
from fastapi.types import DecoratedCallable
from fastapi.utils import (
    generate_unique_id,
)
from starlette.datastructures import URL
from starlette.exceptions import HTTPException
from starlette.responses import JSONResponse, PlainTextResponse, RedirectResponse, Response
from starlette.routing import (
    BaseRoute,
    Match,
)
from starlette.types import Receive, Scope, Send

_T = TypeVar("_T")


def same_definition_as_in(t: _T) -> Callable[[Callable], _T]:
    def decorator(f: Callable) -> _T:
        return f  # pyright: ignore[reportGeneralTypeIssues]

    return decorator


class HeaderVersionedAPIRoute(APIRoute):
    api_version = None

    def is_version_matching(self, scope: Scope) -> bool:
        requested_version = scope["requested_version"]
        return self.api_version == requested_version

    def matches(self, scope: Scope) -> tuple[Match, Scope]:
        match, child_scope = super().matches(scope)

        if match == Match.NONE or match == Match.PARTIAL:
            return match, child_scope

        if self.is_version_matching(scope):
            return Match.FULL, child_scope

        return Match.NONE, child_scope


@cache
def specific_version_api_route(
    version: str,
    route_class: type[APIRoute] = APIRoute,
) -> type[APIRoute]:
    class SpecificVersionAPIRoute(HeaderVersionedAPIRoute, route_class):
        api_version = version

    return SpecificVersionAPIRoute


async def handle_non_existing_version(scope: Scope, receive: Receive, send: Send) -> None:
    if "app" in scope:
        raise HTTPException(
            406,
            f"Requested version {scope['requested_version']} does not exist. ",
        )

    response = PlainTextResponse("Not Acceptable", status_code=406)  # pragma: no cover
    await response(scope, receive, send)  # pragma: no cover


class HeaderVersionedAPIRouter(APIRouter):
    def __init__(
        self,
        default_version: str | None = None,
        *args: Any,
        **kwargs: Any,
    ) -> None:
        self.default_version: str | None = default_version
        self._context_version: str | None = None
        self.registered_versions: set[str | None] = set()
        self.registered_versions.add(self.default_version)
        super().__init__(*args, **kwargs)

    def version(
        self,
        api_version: str,
    ) -> Callable[[DecoratedCallable], DecoratedCallable]:
        self.registered_versions.add(api_version)

        def decorator(func: DecoratedCallable) -> DecoratedCallable:
            func.__endpoint_api_version__ = api_version
            return func

        return decorator

    @same_definition_as_in(APIRouter.add_api_route)
    def add_api_route(
        self,
        path: str,
        endpoint: Callable[..., Any],
        *,
        route_class_override: type[APIRoute] | None = None,
        **kwargs: Any,
    ):
        if route_class_override:
            # called from include_router or similar functions - we are re-generating routes

            # don't override route_class for already versioned routes
            if not issubclass(route_class_override, HeaderVersionedAPIRoute) and self._context_version:
                # need to wrap original route class with HeaderVersionedAPIRoute
                # currently including routes from unversioned router with some externally defined version
                route_class_override = specific_version_api_route(
                    self._context_version,
                    route_class_override,
                )
        else:
            # called from decorator-based routes declaration. extract __endpoint_api_version__ if set and generate
            # proper route
            if endpoint_version := getattr(endpoint, "__endpoint_api_version__", None):
                route_class_override = specific_version_api_route(endpoint_version, APIRoute)
            else:
                # wrap with default version
                route_class_override = specific_version_api_route(self.default_version, APIRoute)

        super().add_api_route(path, endpoint, route_class_override=route_class_override, **kwargs)

    def include_router(
        self,
        router: "APIRouter",
        *,
        prefix: str = "",
        version: str | None = None,
        tags: Optional[list[Union[str, Enum]]] = None,
        dependencies: Optional[Sequence[params.Depends]] = None,
        default_response_class: type[Response] = Default(JSONResponse),
        responses: Optional[dict[Union[int, str], dict[str, Any]]] = None,
        callbacks: Optional[list[BaseRoute]] = None,
        deprecated: Optional[bool] = None,
        include_in_schema: bool = True,
        generate_unique_id_function: Callable[[APIRoute], str] = Default(
            generate_unique_id,
        ),
    ) -> None:
        """
        if 'version' provided - include all the unversioned routes with this version. May be used to wrap existing
        routers with desired version.
        """
        self._context_version = version
        self.registered_versions.add(version)
        super().include_router(
            router=router,
            prefix=prefix,
            tags=tags,
            dependencies=dependencies,
            default_response_class=default_response_class,
            responses=responses,
            callbacks=callbacks,
            deprecated=deprecated,
            include_in_schema=include_in_schema,
            generate_unique_id_function=generate_unique_id_function,
        )
        if isinstance(router, HeaderVersionedAPIRouter):
            self.registered_versions.update(router.registered_versions)

        self._context_version = None

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:  # noqa: C901
        """
        Mostly a duplicate of FastAPI implementation, but with ability to handle partially matched versions.
        A lot of no-covers as there are a lot of edge cases handled exactly as fastapi does
        """
        assert scope["type"] in ("http", "websocket", "lifespan")  # noqa: S101

        if "router" not in scope:  # pragma: no cover
            scope["router"] = self

        if scope["type"] == "lifespan":  # pragma: no cover
            await self.lifespan(scope, receive, send)
            return

        partial = None
        partial_scope = None
        requested_version = scope.get("requested_version")

        if requested_version is not None and requested_version not in self.registered_versions:
            # there are no such route with exactly the same version - just use the closest, but last version.
            # eg - request passed with version header value "10.10.1", and app has only "9.0.0" and "11.0.0".
            # we'll fallback to "9.0.0". It makes sense if there are many different services with independent
            # release cycles. Thus, one service may release 100 different API versions and another - just 2. Clients
            # will be able to use same header for requests to both services, not caring a lot about which versions are
            # supported in each service.
            suitable_versions = []
            for version in self.registered_versions:
                if version is not None and version < requested_version:
                    suitable_versions.append(version)

            if not suitable_versions:
                # this implementation will trigger 406 even on not versioned route if provided version is not registered
                # however, it covers more real-world scenarios. proper distinguishing between 404 in case of not
                # versioned route and 406 with not found version requires deep dive into starlette's Mount (used for
                # doc generation) implementation which seems to be weird in case of further match processing
                await handle_non_existing_version(scope, receive, send)
                # it's not really executed as we'll return from function above, but for code readability it's better
                # to have it
                return  # pragma: no cover
            version_to_use = max(suitable_versions)
            scope["requested_version"] = version_to_use
            # maybe we can generate more narrow set of routes at this point, but it's possible to optimize it later

        for route in self.routes:
            # Determine if any route matches the incoming scope,
            # and hand over to the matching route if found.
            match, child_scope = route.matches(scope)
            if match == Match.FULL:
                scope.update(child_scope)
                await route.handle(scope, receive, send)
                return  # pragma: no cover

            if match == Match.PARTIAL and partial is None:
                partial = route
                partial_scope = child_scope

        if partial is not None:
            # Handle partial matches. These are cases where an endpoint is
            # able to handle the request, but is not a preferred option.
            # We use this in particular to deal with "405 Method Not Allowed".
            scope.update(partial_scope)  # pyright: ignore[reportGeneralTypeIssues]
            await partial.handle(scope, receive, send)
            return  # pragma: no cover

        if scope["type"] == "http" and self.redirect_slashes and scope["path"] != "/":
            redirect_scope = dict(scope)
            if scope["path"].endswith("/"):
                redirect_scope["path"] = redirect_scope["path"].rstrip("/")
            else:
                redirect_scope["path"] = redirect_scope["path"] + "/"

            for route in self.routes:
                match, child_scope = route.matches(redirect_scope)
                if match != Match.NONE:
                    redirect_url = URL(scope=redirect_scope)
                    response = RedirectResponse(url=str(redirect_url))
                    await response(scope, receive, send)
                    return

        await self.default(scope, receive, send)

fastapi_header_versioning/routing.py
tikon93/fastapi-header-versioning
import uuid

import pytest
from fastapi.testclient import TestClient

from examples.basic.app import app


@pytest.fixture()
def client() -> TestClient:
    return TestClient(
        app,
    )


@pytest.fixture()
def item_id() -> str:
    return str(uuid.uuid4())


async def test__versioned_route__exact_version_matching__should_use_correct_version(client: TestClient, item_id: str):
    result = client.get(f"/item/{item_id}?query_parameter=foo", headers={"x-version": "1"})
    assert result.status_code == 200
    assert result.json() == {
        "query_parameter": "foo",
        "id": item_id,
        "dependency_parameters": {
            "skip": 0,
            "limit": 100,
        },
        "version": "1",
    }

    result = client.get(f"/item/{item_id}?new_query_parameter=foo", headers={"x-version": "2"})
    assert result.status_code == 200
    assert result.json() == {
        "new_query_parameter": "foo",
        "id": item_id,
        "dependency_parameters": {
            "offset": 0,
            "limit": 100,
        },
    }


@pytest.mark.parametrize("headers", [{"x-version": "1"}, {"x-version": "3"}, {}])
async def test__route_without_version__any_version_header__should_use_correct_route(client: TestClient, headers):
    result = client.get("/hello", headers=headers)
    assert result.status_code == 200
    assert result.json() == {"greeting": "Hi! It's not versioned route."}


async def test__versioned_route__not_exact_version__should_use_correct_route(client: TestClient, item_id: str):
    result = client.get(f"/item/{item_id}?query_parameter=foo", headers={"x-version": "1.5"})
    assert result.status_code == 200
    assert result.json() == {
        "query_parameter": "foo",
        "id": item_id,
        "dependency_parameters": {
            "skip": 0,
            "limit": 100,
        },
        "version": "1.5",
    }

    result = client.get(f"/item/{item_id}?new_query_parameter=foo", headers={"x-version": "3"})
    assert result.status_code == 200
    assert result.json() == {
        "new_query_parameter": "foo",
        "id": item_id,
        "dependency_parameters": {
            "offset": 0,
            "limit": 100,
        },
    }


async def test__versioned_route__impossible_to_pick_version__should_return_406(client: TestClient, item_id):
    result = client.get(f"/item/{item_id}?query_parameter=foo", headers={"x-version": "0"})
    assert result.status_code == 406


async def test__versioned_route__not_existing_path__should_return_404(client: TestClient):
    result = client.get(f"/item/{item_id}/foo?new_query_parameter=foo", headers={"x-version": "2"})
    assert result.status_code == 404


async def test__route_present_in_one_version__other_version_return_404(client: TestClient):
    result = client.get("/items", headers={"x-version": "2"})
    assert result.status_code == 200
    result = client.get("/items", headers={"x-version": "1"})
    assert result.status_code == 404


@pytest.mark.parametrize("path", ["/hello", "/item/foo"])
async def test__any_route__not_existing_method__should_return_405(client: TestClient, path: str):
    result = client.head(path, headers={"x-version": "2"})
    assert result.status_code == 405


@pytest.mark.parametrize("path", ["/hello/", "/items/"])
async def test__any_route_redirect_slashes__should_perform_redirect(client, path: str):
    result = client.get(path, headers={"x-version": "2"})
    assert result.status_code == 200

tests/examples/test_basic.py
tikon93/fastapi-header-versioning
import uuid

import pytest
from fastapi.testclient import TestClient

from examples.router_level_versions.app import app


@pytest.fixture()
def client() -> TestClient:
    return TestClient(
        app,
    )


@pytest.fixture()
def item_id() -> str:
    return str(uuid.uuid4())


async def test__versioned_route__exact_version_matching__should_use_correct_version(client: TestClient, item_id: str):
    result = client.get(f"/item/{item_id}?query_parameter=foo", headers={"x-version": "1"})
    assert result.status_code == 200
    assert result.json() == {
        "query_parameter": "foo",
        "id": item_id,
        "dependency_parameters": {
            "skip": 0,
            "limit": 100,
        },
        "version": "1",
    }

    result = client.get(f"/item/{item_id}?new_query_parameter=foo", headers={"x-version": "2"})
    assert result.status_code == 200
    assert result.json() == {
        "new_query_parameter": "foo",
        "id": item_id,
        "dependency_parameters": {
            "offset": 0,
            "limit": 100,
        },
    }


@pytest.mark.parametrize("headers", [{"x-version": "1"}, {"x-version": "3"}, {}])
async def test__route_without_version__any_version_header__should_use_correct_route(client: TestClient, headers):
    result = client.get("/hello", headers=headers)
    assert result.status_code == 200
    assert result.json() == {"greeting": "Hi! It's not versioned route."}


async def test__versioned_route__not_exact_version__should_use_correct_route(client: TestClient, item_id: str):
    result = client.get(f"/item/{item_id}?query_parameter=foo", headers={"x-version": "1.5"})
    assert result.status_code == 200
    assert result.json() == {
        "query_parameter": "foo",
        "id": item_id,
        "dependency_parameters": {
            "skip": 0,
            "limit": 100,
        },
        "version": "1.5",
    }

    result = client.get(f"/item/{item_id}?new_query_parameter=foo", headers={"x-version": "3"})
    assert result.status_code == 200
    assert result.json() == {
        "new_query_parameter": "foo",
        "id": item_id,
        "dependency_parameters": {
            "offset": 0,
            "limit": 100,
        },
    }


async def test__versioned_route__impossible_to_pick_version__should_return_406(client: TestClient, item_id):
    result = client.get(f"/item/{item_id}?query_parameter=foo", headers={"x-version": "0"})
    assert result.status_code == 406


async def test__versioned_route__not_existing_path__should_return_404(client: TestClient):
    result = client.get(f"/item/{item_id}/foo?new_query_parameter=foo", headers={"x-version": "2"})
    assert result.status_code == 404


@pytest.mark.parametrize("path", ["/hello", "/item/foo"])
async def test__any_route__not_existing_method__should_return_405(client: TestClient, path: str):
    result = client.head(path, headers={"x-version": "2"})
    assert result.status_code == 405

tests/examples/test_router_level_versions.py
tikon93/fastapi-header-versioning
import uuid

import pytest
from fastapi.testclient import TestClient

from examples.unversioned_routers_wrapping.app import app


@pytest.fixture()
def client() -> TestClient:
    return TestClient(
        app,
    )


@pytest.fixture()
def item_id() -> str:
    return str(uuid.uuid4())


async def test__versioned_route__exact_version_matching__should_use_correct_version(client: TestClient, item_id: str):
    result = client.get(f"/item/{item_id}?query_parameter=foo", headers={"x-version": "1"})
    assert result.status_code == 200
    assert result.json() == {
        "query_parameter": "foo",
        "id": item_id,
        "dependency_parameters": {
            "skip": 0,
            "limit": 100,
        },
        "version": "1",
    }

    result = client.get(f"/item/{item_id}?new_query_parameter=foo", headers={"x-version": "2"})
    assert result.status_code == 200
    assert result.json() == {
        "new_query_parameter": "foo",
        "id": item_id,
        "dependency_parameters": {
            "offset": 0,
            "limit": 100,
        },
    }


@pytest.mark.parametrize("headers", [{"x-version": "1"}, {"x-version": "3"}, {}])
async def test__route_without_version__any_version_header__should_use_correct_route(client: TestClient, headers):
    result = client.get("/hello", headers=headers)
    assert result.status_code == 200
    assert result.json() == {"greeting": "Hi! It's not versioned route."}


async def test__versioned_route__not_exact_version__should_use_correct_route(client: TestClient, item_id: str):
    result = client.get(f"/item/{item_id}?query_parameter=foo", headers={"x-version": "1.5"})
    assert result.status_code == 200
    assert result.json() == {
        "query_parameter": "foo",
        "id": item_id,
        "dependency_parameters": {
            "skip": 0,
            "limit": 100,
        },
        "version": "1.5",
    }

    result = client.get(f"/item/{item_id}?new_query_parameter=foo", headers={"x-version": "3"})
    assert result.status_code == 200
    assert result.json() == {
        "new_query_parameter": "foo",
        "id": item_id,
        "dependency_parameters": {
            "offset": 0,
            "limit": 100,
        },
    }


async def test__versioned_route__impossible_to_pick_version__should_return_406(client: TestClient, item_id):
    result = client.get(f"/item/{item_id}?query_parameter=foo", headers={"x-version": "0"})
    assert result.status_code == 406


async def test__versioned_route__not_existing_path__should_return_404(client: TestClient):
    result = client.get(f"/item/{item_id}/foo?new_query_parameter=foo", headers={"x-version": "2"})
    assert result.status_code == 404


@pytest.mark.parametrize("path", ["/hello", "/item/foo"])
async def test__any_route__not_existing_method__should_return_405(client: TestClient, path: str):
    result = client.head(path, headers={"x-version": "2"})
    assert result.status_code == 405

tests/examples/test_unversioned_routers_wrapping.py
tikon93/fastapi-header-versioning
async def test__example__get_different_versions__return_correct_response():
    pass

tests/test_example.py
tikon93/fastapi-header-versioning