Skip to content

Generate AWS API Gateway-compatible Swagger Schemae #5437

@rmehlinger

Description

@rmehlinger

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

from typing import List
from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi
from typing import Any, Generic, Optional, Tuple, Type, TypeVar
from pydantic import BaseModel, validator
from pydantic.generics import GenericModel

app = FastAPI(
    title="AMP Data Platform",
    description="TBW",
    version="0.1",
    docs_url="/docs",
    openapi_url="/openapi.json",
    redoc_url=None,
)


DataT = TypeVar("DataT")


class Error(BaseModel):
    code: int
    message: str


class Response(GenericModel, Generic[DataT]):
    data: Optional[DataT]
    error: Optional[Error]

    @validator("error", always=True)
    def check_consistency(cls, v, values):
        if v is not None and values["data"] is not None:
            raise ValueError("must not provide both data and error")
        if v is None and values.get("data") is None:
            raise ValueError("must provide data or error")
        return v


class User(BaseModel):
    name: str
    email: str
    id: int


@app.get("/", response_model=Response[List[str]])
def root():
    return {"data": "Hello World"}


@app.get("/users", response_model=Response[List[User]])
def users():
    return {"data": []}


@app.get("/users/{user_id}", response_model=Response[User])
def user(user_id: int):
    return {
        "data": {
            "id": user_id,
            "email": "richardm@dropbox.com",
            "name": "Richard Mehlinger",
        }
    }

def generate_open_api_spec():
    # see https://github.com/tiangolo/fastapi/issues/1173
    open_api_content = get_openapi(
        title=app.title,
        version=app.version,
        openapi_version=app.openapi_version,
        description=app.description,
        routes=app.routes,
        # openapi_prefix=app.openapi_prefix,
    )

    # see https://fastapi.tiangolo.com/advanced/generate-clients/#preprocess-the-openapi-specification-for-the-client-generator

    for path_data in open_api_content["paths"].values():
        for operation in path_data.values():
            tags = operation.get("tags", [])
            if len(tags):
                tag = operation.get("tags")[0]
                to_remove = f"{tag}-"
                operation_id = operation["operationId"]
                new_operation_id = operation_id[len(to_remove) :]
                operation["operationId"] = new_operation_id

    return open_api_content


# Generated Schema Follows (from calling generate_open_api_spec)
"""
{
  "openapi": "3.0.2",
  "info": {
    "title": "My App",
    "description": "TBW",
    "version": "0.1"
  },
  "paths": {
    "/": {
      "get": {
        "summary": "Root",
        "operationId": "root__get",
        "responses": {
          "200": {
            "description": "Successful Response",
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/Response_List_str__"
                }
              }
            }
          }
        }
      }
    },
    "/users": {
      "get": {
        "summary": "Users",
        "operationId": "users_users_get",
        "responses": {
          "200": {
            "description": "Successful Response",
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/Response_List_my_app.main.User__"
                }
              }
            }
          }
        }
      }
    },
    "/users/{user_id}": {
      "get": {
        "summary": "User",
        "operationId": "user_users__user_id__get",
        "parameters": [
          {
            "required": true,
            "schema": {
              "title": "User Id",
              "type": "integer"
            },
            "name": "user_id",
            "in": "path"
          }
        ],
        "responses": {
          "200": {
            "description": "Successful Response",
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/Response_User_"
                }
              }
            }
          },
          "422": {
            "description": "Validation Error",
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/HTTPValidationError"
                }
              }
            }
          }
        }
      }
    }
  },
  "components": {
    "schemas": {
      "Error": {
        "title": "Error",
        "required": [
          "code",
          "message"
        ],
        "type": "object",
        "properties": {
          "code": {
            "title": "Code",
            "type": "integer"
          },
          "message": {
            "title": "Message",
            "type": "string"
          }
        }
      },
      "HTTPValidationError": {
        "title": "HTTPValidationError",
        "type": "object",
        "properties": {
          "detail": {
            "title": "Detail",
            "type": "array",
            "items": {
              "$ref": "#/components/schemas/ValidationError"
            }
          }
        }
      },
      "Response_List_my_app.main.User__": {
        "title": "Response[List[my_app.main.User]]",
        "type": "object",
        "properties": {
          "data": {
            "title": "Data",
            "type": "array",
            "items": {
              "$ref": "#/components/schemas/User"
            }
          },
          "error": {
            "$ref": "#/components/schemas/Error"
          }
        }
      },
      "Response_List_str__": {
        "title": "Response[List[str]]",
        "type": "object",
        "properties": {
          "data": {
            "title": "Data",
            "type": "array",
            "items": {
              "type": "string"
            }
          },
          "error": {
            "$ref": "#/components/schemas/Error"
          }
        }
      },
      "Response_User_": {
        "title": "Response[User]",
        "type": "object",
        "properties": {
          "data": {
            "$ref": "#/components/schemas/User"
          },
          "error": {
            "$ref": "#/components/schemas/Error"
          }
        }
      },
      "User": {
        "title": "User",
        "required": [
          "name",
          "email",
          "id"
        ],
        "type": "object",
        "properties": {
          "name": {
            "title": "Name",
            "type": "string"
          },
          "email": {
            "title": "Email",
            "type": "string"
          },
          "id": {
            "title": "Id",
            "type": "integer"
          }
        }
      },
      "ValidationError": {
        "title": "ValidationError",
        "required": [
          "loc",
          "msg",
          "type"
        ],
        "type": "object",
        "properties": {
          "loc": {
            "title": "Location",
            "type": "array",
            "items": {
              "anyOf": [
                {
                  "type": "string"
                },
                {
                  "type": "integer"
                }
              ]
            }
          },
          "msg": {
            "title": "Message",
            "type": "string"
          },
          "type": {
            "title": "Error Type",
            "type": "string"
          }
        }
      }
    }
  }
}
"""

Description

One of the major selling points of FastAPI is its ability to generate OpenAPI specs, which are consumable by AWS API Gateway. However, when attempting to import the generated schema above into AWS API Gateway, we receive the following errors:
image

Essentially, API Gateway strictly bans non-alphanumeric characters from schema names--even underscores. However, by default the schema names generated by Pydantic do include both module path names and underscores, meaning that API Gateway rejects them.

The rather non-obvious solution was to override the Response class's __concrete_name__ method:

    # Necessary because AWS bars any non-alphanumeric characters from schema names, including underscores.
    # This takes the default model name and does the following:
    #   removes the path to this file, which will otherwise show up in all of our names
    #   TitleCases any text
    #   strips non-alphanumeric characters
    #   moves "Response" from the beginning to the end of the name.
    non_alphanumeric = re.compile("[^a-zA-Z0-9]")

    @classmethod
    def __concrete_name__(cls: Type[Any], params: Tuple[Type[Any], ...]) -> str:
        return (
            re.sub(
                non_alphanumeric,
                "",
                super()
                .__concrete_name__(params)
                .title(),
            ).replace("Response", "", 1)
            + "Response"
        )

This ensures that we do not generate any schema names with non-alphanumeric characters.

Wanted Solution

This is partially a Pydantic issue. FastAPI uses names generated by pydantic.schema.get_model_name_map, and the offending model names are generated by the normalize_name and get_long_model_name helpers (See line 422 of fastapi.openapi.utils). However, there is nothing stopping FastAPI from renormalizing the model names generated by Pydantic.

I would propose the following steps:

  1. Add a section to the documentation on integrating with AWS API Gateway with instructions for how to override __concrete_name__.
  2. Add an optional flag to generate AWS API Gateway compatible schema names. Note that in so doing care will need to be taken to avoid generating collisions.
  3. File an issue on Pydantic to expose native support for the above, so that FastAPI does not need to essentially rewrite the schema names.

Wanted Code

def get_openapi(
    *,
    title: str,
    version: str,
    openapi_version: str = "3.0.2",
    description: Optional[str] = None,
    routes: Sequence[BaseRoute],
    tags: Optional[List[Dict[str, Any]]] = None,
    servers: Optional[List[Dict[str, Union[str, Any]]]] = None,
    terms_of_service: Optional[str] = None,
    contact: Optional[Dict[str, Union[str, Any]]] = None,
    license_info: Optional[Dict[str, Union[str, Any]]] = None,
    *schema_name_transform: Callable[[str], str] # one option, allow users to specify a name transform when generating a schema.
    *flags: List[Str] # one option, would include an "aws-api-gateway" flag
) -> Dict[str, Any]:


### Alternatives

Another option could be for FastAPI to provide its own extensions of `BaseModel` and `GenericModel` with overwritten `__concrete_name__` methods. However, this seems a bit off to me. We shouldn't be renaming our models based on the requirements of a consumer; after all, what if we have multiple consumers with different naming requirements? Instead we should be formatting our Open API output to match the requirements of our consumers.

### Operating System

macOS

### Operating System Details

_No response_

### FastAPI Version

0.85.0

### Python Version

3.8.9

### Additional Context

_No response_

Metadata

Metadata

Assignees

No one assigned

    Type

    No type

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions