Would be nice to be able to route request using header's Accept field (or generic header's field)

[mcosta74]

First Check

• I added a very descriptive title to this issue.
• I used the GitHub search to find a similar issue and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to FastAPI but to Pydantic.
• I already checked if it is not related to FastAPI but to Swagger UI.
• I already checked if it is not related to FastAPI but to ReDoc.

Commit to Help

• I commit to help with one of those options 👆

Example Code

N.A.

Description

In some cases would be nice to specify header's field as routing rules.

One important example is to support API versioning based on Header's Accept field

Wanted Solution

Ability to specify some header's fields in the .get(), .post() .... decorators

Wanted Code

from fastapi import FastAPI

app = FastAPI()


@app.get("/", accept="application/json;version=1.0")
async def root():
    return {"message": "Hello World v1.0"}

@app.get("/", accept="application/json;version=1.1")
async def root():
    return {"message": "Hello World v1.1"}

Alternatives

from fastapi import FastAPI

app = FastAPI()

@app.get("/", headers={"accept": "application/json;version=1.0"})
async def root():
return {"message": "Hello World v1.0"}

@app.get("/", headers={"accept": "application/json;version=1.1"})
async def root():
return {"message": "Hello World v1.1"}

Operating System

macOS

Operating System Details

No response

FastAPI Version

python -c "import fastapi; print(fastapi.version)"

Python Version

Python 3.9.7

Additional Context

No response

[ycd]

You still have access to request object, couldn't you use that to retrieve content-type and implement your logic as if there are two different endpoints?

[mcosta74]

You still have access to request object, couldn't you use that to retrieve content-type and implement your logic as if there are two different endpoints?

well, it's not so easy.
Usually new version of the API means breaking changes in the request body and/or response body.

Handle it in a single endpoint will end-up in a "messy" code.

For instance how to define different input body depending of the version?

[ghandic]

You could handle the version in the route? Eg /api/v1/person /api/v2/person

[mcosta74]

You could handle the version in the route? Eg /api/v1/person /api/v2/person

well, I know that routing using URLs is available; the feature request is about adding routing using Accept header

[ghandic]

Could you link to any other implementations where this feature is available? It seems quite niche to need to do this.

[mcosta74]

Could you link to any other implementations where this feature is available? It seems quite niche to need to do this.

One example is the Go gorilla/mux toolkit. It allows to route also using header values

[ghandic]

From what I can see that repo doesn't implement openapi specifications - I haven't spent too much time digging but if you can link to how it would be supported from openapi then it would definitely be helpful in continuing the discussion

[ghandic]

Two references I have found on the hotly debated topic

belgif/rest-guide#1

OAI/OpenAPI-Specification#146

[mcosta74]

From what I can see that repo doesn't implement openapi specifications

Does OpenAPI describe also HTTP headers? (no sarcasm here, I'm really asking)

The reason I'm asking is because the Django Rest Framework Documentation mentions that Accept header versioning is considered a sort of "best practice"

[ghandic]

It supports the headers, but I'm not sure if it supports distinguishing routes to the same resource and having a different header that could return a different response model

Eg v1 returns {status: str}
V2 returns {status: bool}

You could do a oneOf the models but then it's not explicit

[ghandic]

Another open issue referencing it

OpenAPITools/openapi-generator#3569

[ghandic]

Could you have two FastAPI applications each using a different global version with a proxy in front that diverts traffic based on headers?

[zMendes]

I liked the proposal and I will to try to work on this feature.

[zamonia500]

I needs this feature too! I hope fastapi support not only accept but also content-type.

This is my usecase.
I'd like to upload some data on database and the data could be passed to the fastapi as a json string or plain file.

@router.post(
    "/configs",
    response_model=ConfigOut,
    status_code=status.HTTP_201_CREATED,
)
def upload_config_multipart(config: UploadFile = File(...)) -> Any:
    # read file and upload config to database

@router.post(
    "/configs",
    response_model=ConfigOut,
    status_code=status.HTTP_201_CREATED,
)
def upload_config_json(config: ConfigJson) -> Any:
    # read the json payload and upload config to database

[bluebrown]

Open API has consumes and produces. For example:

paths:
  /items:
    get:
      consumes:
        - text/plain
      # produces is the specification of what is supported in the Accept header
      # it will also be available in the dropdown in swagger ui
      produces:
        - application/json
        - text/csv

I am looking for a way to implement this with FastAPI, but it's not straight forward, it seems.

Ok, it was not too bad TBH. I got it working like this.

@warranty_router.get(
    "/",
    status_code=status.HTTP_200_OK,
    response_model=Sequence[Item],
    responses={
        status.HTTP_200_OK: {
            "content": {
                "application/json": {},
                "text/csv": {},
            }
        },
    },
)
async def get_warranty_requests(response: Response) :
  
    accept = request.headers.get("Accept")

    if not accept or accept == "application/json":
        ...
        response.headers["Vary"] = "Accept"
        return records

    if accept == "text/csv":
        ...
        return Response(
            data, media_type="text/csv", headers={
                "Vary": "Accept",
                "Content-Disposition": "attachment; filename=warranty_requests.csv",
            }
        )

    else:
        raise HTTPException(
            status_code=status.HTTP_406_NOT_ACCEPTABLE, detail="Accept header must be 'application/json' or 'text/csv'"
        )

[HHK1]

You could use the specification for vendor types in the Accept header, and have headers like: Accept: application/vnd.myapp.v1.1+json

Then you need to do 3 things:

• Read that version inside an ASGI middleware
• Decorate your routes to add the version to them
• Override the route matches function to find the matching version

(basically what @dmontagu indicates here: #200 (comment))

Middleware looks like this

from typing import Union, cast

from asgiref.typing import (
    ASGI3Application,
    ASGIReceiveCallable,
    ASGISendCallable,
    HTTPScope,
    Scope,
    WebSocketScope,
)
from black import re


class AcceptHeaderVersionMiddleware:
    """
    Use this middleware to parse the Accept Header if present and get an API version
    from the vendor tree. See https://www.rfc-editor.org/rfc/rfc6838#section-3.2

    If incoming http or websocket request contains an Accept header with the following
    value: `"accept/vnd.vendor_prefix.v42+json"`, the scope of the ASGI application
    will then contain an `api_version` of 42.

    If the http or websocket request does not contain an Accept header, or if the accept
    header value does not use a proper format, the scope of the ASGI application will
    then contain an `api_version` that defaults to the provided `latest_version`
    """

    def __init__(
        self, app: ASGI3Application, vendor_prefix: str, latest_version: str
    ) -> None:
        self.app = app
        self.latest_version = latest_version
        self.accept_regex = rf"^application/vnd\.{vendor_prefix}\.v([0-9]+)\+.*"

    async def __call__(
        self, scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable
    ) -> None:
        if scope["type"] in ("http", "websocket"):
            scope = cast(Union[HTTPScope, WebSocketScope], scope)
            headers = dict(scope["headers"])
            scope["latest_version"] = self.latest_version  # type: ignore[index]
            scope["requested_version"] = self.latest_version  # type: ignore[index]

            if b"accept" in headers:
                accept_header = headers[b"accept"].decode("latin1")
                match = re.search(self.accept_regex, accept_header)
                if match is not None:
                    api_version = match.group(1)
                    if api_version is not None:
                        scope["requested_version"] = api_version  # type: ignore[index]

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

Router + Route subclasses:

from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Type, Union

from fastapi import APIRouter, params
from fastapi.datastructures import Default
from fastapi.types import DecoratedCallable

from starlette.responses import JSONResponse, Response, PlainTextResponse
from starlette.routing import BaseRoute, Match
from starlette.types import ASGIApp, Receive, Scope, Send
from starlette.exceptions import HTTPException
from fastapi.routing import APIRoute


class VersionedAPIRoute(APIRoute):
    @property
    def endpoint_version(self) -> str:
        return str(self.endpoint.__api_version__)  # type:ignore

    def is_version_matching(self, scope: Scope) -> bool:
        requested_version = scope["requested_version"]
        is_latest = self.endpoint_version == "latest"

        return (
            is_latest and requested_version == scope["latest_version"]
        ) or self.endpoint_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
        else:
            return Match.PARTIAL, child_scope

    async def handle(self, scope: Scope, receive: Receive, send: Send) -> None:
        if not self.is_version_matching(scope):
            if "app" in scope:
                raise HTTPException(
                    406,
                    f"Requested version {scope['requested_version']} does not exist. "
                    f"Latest available version is {scope['latest_version']}.",
                )
            else:
                response = PlainTextResponse("Not Acceptable", status_code=406)
            await response(scope, receive, send)
        await super().handle(scope, receive, send)


class VersionedAPIRouter(APIRouter):
    def __init__(
        self,
        *,
        prefix: str = "",
        tags: Optional[List[str]] = 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,
        routes: Optional[List[BaseRoute]] = None,
        redirect_slashes: bool = True,
        default: Optional[ASGIApp] = None,
        dependency_overrides_provider: Optional[Any] = None,
        route_class: Type[VersionedAPIRoute] = VersionedAPIRoute,
        on_startup: Optional[Sequence[Callable[[], Any]]] = None,
        on_shutdown: Optional[Sequence[Callable[[], Any]]] = None,
        deprecated: Optional[bool] = None,
        include_in_schema: bool = True,
    ) -> None:
        super().__init__(
            prefix=prefix,
            tags=tags,
            dependencies=dependencies,
            default_response_class=default_response_class,
            responses=responses,
            callbacks=callbacks,
            routes=routes,
            redirect_slashes=redirect_slashes,
            default=default,
            dependency_overrides_provider=dependency_overrides_provider,
            route_class=route_class,
            on_startup=on_startup,
            on_shutdown=on_shutdown,
            deprecated=deprecated,
            include_in_schema=include_in_schema,
        )

    def version(
        self, api_version: str
    ) -> Callable[[DecoratedCallable], DecoratedCallable]:
        def decorator(func: DecoratedCallable) -> DecoratedCallable:
            func.__api_version__ = api_version  # type:ignore
            return func

        return decorator

Then you can plug everything together:

router = VersionedAPIRouter()

GREETINGS_V1 = "Hello"
GREETINGS_V2 = "It's me"
GREETINGS_V3 = "I was wondering if after all these years you'd like to meet"
GREETINGS_LATEST = "Hello, can you hear me?"

router = VersionedAPIRouter()

@router.get("/hello")
@router.version("1")
async def hello_v1() -> Dict:
    return {"greetings": GREETINGS_V1}

@router.get("/hello")
@router.version("2")
async def hello_v2() -> Dict:
    return {"greetings": GREETINGS_V2}

@router.get("/hello")
@router.version("3")
async def hello_v3() -> Dict:
    return {"greetings": GREETINGS_V3}

@router.get("/hello")
@router.version("latest")
async def hello_latest() -> Dict:
    return {"greetings": GREETINGS_LATEST}


app = FastAPI(title="Versioned app")
app.add_middleware(
    AcceptHeaderVersionMiddleware, vendor_prefix="mytestapp", latest_version="4"
)
app.include_router(router)

[ghandic]

How does this look on the swagger page? I wasn't aware that openapi supported multi version or for the routes to have different params for the same route

[HHK1]

How does this look on the swagger page? I wasn't aware that openapi supported multi version or for the routes to have different params for the same route

Haven't tested that yet, I'll take a look. I was thinking that since it's using the accept header for versioning, we could use the media type option of the open API spec: https://swagger.io/docs/specification/media-types/. That could work, at least for the response format.

Regarding request params, I don't know exactly how that would be described in swagger.

[plankthom]

I contributed PR #4727 that merges the open-api descriptions for routes that overlap on path and method (as in the examples above).

The current version of fast api just keeps the last operation description for each path and method.