| from collections.abc import Iterable, Mapping
|
| from enum import Enum
|
| from typing import Annotated, Any, Callable, Optional, Union
|
|
|
| from fastapi._compat import with_info_plain_validator_function
|
| from fastapi.logger import logger
|
| from pydantic import (
|
| AnyUrl,
|
| BaseModel,
|
| Field,
|
| GetJsonSchemaHandler,
|
| )
|
| from typing_extensions import Literal, TypedDict
|
| from typing_extensions import deprecated as typing_deprecated
|
|
|
| try:
|
| import email_validator
|
|
|
| assert email_validator
|
| from pydantic import EmailStr
|
| except ImportError:
|
|
|
| class EmailStr(str):
|
| @classmethod
|
| def __get_validators__(cls) -> Iterable[Callable[..., Any]]:
|
| yield cls.validate
|
|
|
| @classmethod
|
| def validate(cls, v: Any) -> str:
|
| logger.warning(
|
| "email-validator not installed, email fields will be treated as str.\n"
|
| "To install, run: pip install email-validator"
|
| )
|
| return str(v)
|
|
|
| @classmethod
|
| def _validate(cls, __input_value: Any, _: Any) -> str:
|
| logger.warning(
|
| "email-validator not installed, email fields will be treated as str.\n"
|
| "To install, run: pip install email-validator"
|
| )
|
| return str(__input_value)
|
|
|
| @classmethod
|
| def __get_pydantic_json_schema__(
|
| cls, core_schema: Mapping[str, Any], handler: GetJsonSchemaHandler
|
| ) -> dict[str, Any]:
|
| return {"type": "string", "format": "email"}
|
|
|
| @classmethod
|
| def __get_pydantic_core_schema__(
|
| cls, source: type[Any], handler: Callable[[Any], Mapping[str, Any]]
|
| ) -> Mapping[str, Any]:
|
| return with_info_plain_validator_function(cls._validate)
|
|
|
|
|
| class BaseModelWithConfig(BaseModel):
|
| model_config = {"extra": "allow"}
|
|
|
|
|
| class Contact(BaseModelWithConfig):
|
| name: Optional[str] = None
|
| url: Optional[AnyUrl] = None
|
| email: Optional[EmailStr] = None
|
|
|
|
|
| class License(BaseModelWithConfig):
|
| name: str
|
| identifier: Optional[str] = None
|
| url: Optional[AnyUrl] = None
|
|
|
|
|
| class Info(BaseModelWithConfig):
|
| title: str
|
| summary: Optional[str] = None
|
| description: Optional[str] = None
|
| termsOfService: Optional[str] = None
|
| contact: Optional[Contact] = None
|
| license: Optional[License] = None
|
| version: str
|
|
|
|
|
| class ServerVariable(BaseModelWithConfig):
|
| enum: Annotated[Optional[list[str]], Field(min_length=1)] = None
|
| default: str
|
| description: Optional[str] = None
|
|
|
|
|
| class Server(BaseModelWithConfig):
|
| url: Union[AnyUrl, str]
|
| description: Optional[str] = None
|
| variables: Optional[dict[str, ServerVariable]] = None
|
|
|
|
|
| class Reference(BaseModel):
|
| ref: str = Field(alias="$ref")
|
|
|
|
|
| class Discriminator(BaseModel):
|
| propertyName: str
|
| mapping: Optional[dict[str, str]] = None
|
|
|
|
|
| class XML(BaseModelWithConfig):
|
| name: Optional[str] = None
|
| namespace: Optional[str] = None
|
| prefix: Optional[str] = None
|
| attribute: Optional[bool] = None
|
| wrapped: Optional[bool] = None
|
|
|
|
|
| class ExternalDocumentation(BaseModelWithConfig):
|
| description: Optional[str] = None
|
| url: AnyUrl
|
|
|
|
|
|
|
| SchemaType = Literal[
|
| "array", "boolean", "integer", "null", "number", "object", "string"
|
| ]
|
|
|
|
|
| class Schema(BaseModelWithConfig):
|
|
|
|
|
| schema_: Optional[str] = Field(default=None, alias="$schema")
|
| vocabulary: Optional[str] = Field(default=None, alias="$vocabulary")
|
| id: Optional[str] = Field(default=None, alias="$id")
|
| anchor: Optional[str] = Field(default=None, alias="$anchor")
|
| dynamicAnchor: Optional[str] = Field(default=None, alias="$dynamicAnchor")
|
| ref: Optional[str] = Field(default=None, alias="$ref")
|
| dynamicRef: Optional[str] = Field(default=None, alias="$dynamicRef")
|
| defs: Optional[dict[str, "SchemaOrBool"]] = Field(default=None, alias="$defs")
|
| comment: Optional[str] = Field(default=None, alias="$comment")
|
|
|
|
|
| allOf: Optional[list["SchemaOrBool"]] = None
|
| anyOf: Optional[list["SchemaOrBool"]] = None
|
| oneOf: Optional[list["SchemaOrBool"]] = None
|
| not_: Optional["SchemaOrBool"] = Field(default=None, alias="not")
|
| if_: Optional["SchemaOrBool"] = Field(default=None, alias="if")
|
| then: Optional["SchemaOrBool"] = None
|
| else_: Optional["SchemaOrBool"] = Field(default=None, alias="else")
|
| dependentSchemas: Optional[dict[str, "SchemaOrBool"]] = None
|
| prefixItems: Optional[list["SchemaOrBool"]] = None
|
| items: Optional["SchemaOrBool"] = None
|
| contains: Optional["SchemaOrBool"] = None
|
| properties: Optional[dict[str, "SchemaOrBool"]] = None
|
| patternProperties: Optional[dict[str, "SchemaOrBool"]] = None
|
| additionalProperties: Optional["SchemaOrBool"] = None
|
| propertyNames: Optional["SchemaOrBool"] = None
|
| unevaluatedItems: Optional["SchemaOrBool"] = None
|
| unevaluatedProperties: Optional["SchemaOrBool"] = None
|
|
|
|
|
| type: Optional[Union[SchemaType, list[SchemaType]]] = None
|
| enum: Optional[list[Any]] = None
|
| const: Optional[Any] = None
|
| multipleOf: Optional[float] = Field(default=None, gt=0)
|
| maximum: Optional[float] = None
|
| exclusiveMaximum: Optional[float] = None
|
| minimum: Optional[float] = None
|
| exclusiveMinimum: Optional[float] = None
|
| maxLength: Optional[int] = Field(default=None, ge=0)
|
| minLength: Optional[int] = Field(default=None, ge=0)
|
| pattern: Optional[str] = None
|
| maxItems: Optional[int] = Field(default=None, ge=0)
|
| minItems: Optional[int] = Field(default=None, ge=0)
|
| uniqueItems: Optional[bool] = None
|
| maxContains: Optional[int] = Field(default=None, ge=0)
|
| minContains: Optional[int] = Field(default=None, ge=0)
|
| maxProperties: Optional[int] = Field(default=None, ge=0)
|
| minProperties: Optional[int] = Field(default=None, ge=0)
|
| required: Optional[list[str]] = None
|
| dependentRequired: Optional[dict[str, set[str]]] = None
|
|
|
|
|
| format: Optional[str] = None
|
|
|
|
|
| contentEncoding: Optional[str] = None
|
| contentMediaType: Optional[str] = None
|
| contentSchema: Optional["SchemaOrBool"] = None
|
|
|
|
|
| title: Optional[str] = None
|
| description: Optional[str] = None
|
| default: Optional[Any] = None
|
| deprecated: Optional[bool] = None
|
| readOnly: Optional[bool] = None
|
| writeOnly: Optional[bool] = None
|
| examples: Optional[list[Any]] = None
|
|
|
|
|
| discriminator: Optional[Discriminator] = None
|
| xml: Optional[XML] = None
|
| externalDocs: Optional[ExternalDocumentation] = None
|
| example: Annotated[
|
| Optional[Any],
|
| typing_deprecated(
|
| "Deprecated in OpenAPI 3.1.0 that now uses JSON Schema 2020-12, "
|
| "although still supported. Use examples instead."
|
| ),
|
| ] = None
|
|
|
|
|
|
|
|
|
| SchemaOrBool = Union[Schema, bool]
|
|
|
|
|
| class Example(TypedDict, total=False):
|
| summary: Optional[str]
|
| description: Optional[str]
|
| value: Optional[Any]
|
| externalValue: Optional[AnyUrl]
|
|
|
| __pydantic_config__ = {"extra": "allow"}
|
|
|
|
|
| class ParameterInType(Enum):
|
| query = "query"
|
| header = "header"
|
| path = "path"
|
| cookie = "cookie"
|
|
|
|
|
| class Encoding(BaseModelWithConfig):
|
| contentType: Optional[str] = None
|
| headers: Optional[dict[str, Union["Header", Reference]]] = None
|
| style: Optional[str] = None
|
| explode: Optional[bool] = None
|
| allowReserved: Optional[bool] = None
|
|
|
|
|
| class MediaType(BaseModelWithConfig):
|
| schema_: Optional[Union[Schema, Reference]] = Field(default=None, alias="schema")
|
| example: Optional[Any] = None
|
| examples: Optional[dict[str, Union[Example, Reference]]] = None
|
| encoding: Optional[dict[str, Encoding]] = None
|
|
|
|
|
| class ParameterBase(BaseModelWithConfig):
|
| description: Optional[str] = None
|
| required: Optional[bool] = None
|
| deprecated: Optional[bool] = None
|
|
|
| style: Optional[str] = None
|
| explode: Optional[bool] = None
|
| allowReserved: Optional[bool] = None
|
| schema_: Optional[Union[Schema, Reference]] = Field(default=None, alias="schema")
|
| example: Optional[Any] = None
|
| examples: Optional[dict[str, Union[Example, Reference]]] = None
|
|
|
| content: Optional[dict[str, MediaType]] = None
|
|
|
|
|
| class Parameter(ParameterBase):
|
| name: str
|
| in_: ParameterInType = Field(alias="in")
|
|
|
|
|
| class Header(ParameterBase):
|
| pass
|
|
|
|
|
| class RequestBody(BaseModelWithConfig):
|
| description: Optional[str] = None
|
| content: dict[str, MediaType]
|
| required: Optional[bool] = None
|
|
|
|
|
| class Link(BaseModelWithConfig):
|
| operationRef: Optional[str] = None
|
| operationId: Optional[str] = None
|
| parameters: Optional[dict[str, Union[Any, str]]] = None
|
| requestBody: Optional[Union[Any, str]] = None
|
| description: Optional[str] = None
|
| server: Optional[Server] = None
|
|
|
|
|
| class Response(BaseModelWithConfig):
|
| description: str
|
| headers: Optional[dict[str, Union[Header, Reference]]] = None
|
| content: Optional[dict[str, MediaType]] = None
|
| links: Optional[dict[str, Union[Link, Reference]]] = None
|
|
|
|
|
| class Operation(BaseModelWithConfig):
|
| tags: Optional[list[str]] = None
|
| summary: Optional[str] = None
|
| description: Optional[str] = None
|
| externalDocs: Optional[ExternalDocumentation] = None
|
| operationId: Optional[str] = None
|
| parameters: Optional[list[Union[Parameter, Reference]]] = None
|
| requestBody: Optional[Union[RequestBody, Reference]] = None
|
|
|
| responses: Optional[dict[str, Union[Response, Any]]] = None
|
| callbacks: Optional[dict[str, Union[dict[str, "PathItem"], Reference]]] = None
|
| deprecated: Optional[bool] = None
|
| security: Optional[list[dict[str, list[str]]]] = None
|
| servers: Optional[list[Server]] = None
|
|
|
|
|
| class PathItem(BaseModelWithConfig):
|
| ref: Optional[str] = Field(default=None, alias="$ref")
|
| summary: Optional[str] = None
|
| description: Optional[str] = None
|
| get: Optional[Operation] = None
|
| put: Optional[Operation] = None
|
| post: Optional[Operation] = None
|
| delete: Optional[Operation] = None
|
| options: Optional[Operation] = None
|
| head: Optional[Operation] = None
|
| patch: Optional[Operation] = None
|
| trace: Optional[Operation] = None
|
| servers: Optional[list[Server]] = None
|
| parameters: Optional[list[Union[Parameter, Reference]]] = None
|
|
|
|
|
| class SecuritySchemeType(Enum):
|
| apiKey = "apiKey"
|
| http = "http"
|
| oauth2 = "oauth2"
|
| openIdConnect = "openIdConnect"
|
|
|
|
|
| class SecurityBase(BaseModelWithConfig):
|
| type_: SecuritySchemeType = Field(alias="type")
|
| description: Optional[str] = None
|
|
|
|
|
| class APIKeyIn(Enum):
|
| query = "query"
|
| header = "header"
|
| cookie = "cookie"
|
|
|
|
|
| class APIKey(SecurityBase):
|
| type_: SecuritySchemeType = Field(default=SecuritySchemeType.apiKey, alias="type")
|
| in_: APIKeyIn = Field(alias="in")
|
| name: str
|
|
|
|
|
| class HTTPBase(SecurityBase):
|
| type_: SecuritySchemeType = Field(default=SecuritySchemeType.http, alias="type")
|
| scheme: str
|
|
|
|
|
| class HTTPBearer(HTTPBase):
|
| scheme: Literal["bearer"] = "bearer"
|
| bearerFormat: Optional[str] = None
|
|
|
|
|
| class OAuthFlow(BaseModelWithConfig):
|
| refreshUrl: Optional[str] = None
|
| scopes: dict[str, str] = {}
|
|
|
|
|
| class OAuthFlowImplicit(OAuthFlow):
|
| authorizationUrl: str
|
|
|
|
|
| class OAuthFlowPassword(OAuthFlow):
|
| tokenUrl: str
|
|
|
|
|
| class OAuthFlowClientCredentials(OAuthFlow):
|
| tokenUrl: str
|
|
|
|
|
| class OAuthFlowAuthorizationCode(OAuthFlow):
|
| authorizationUrl: str
|
| tokenUrl: str
|
|
|
|
|
| class OAuthFlows(BaseModelWithConfig):
|
| implicit: Optional[OAuthFlowImplicit] = None
|
| password: Optional[OAuthFlowPassword] = None
|
| clientCredentials: Optional[OAuthFlowClientCredentials] = None
|
| authorizationCode: Optional[OAuthFlowAuthorizationCode] = None
|
|
|
|
|
| class OAuth2(SecurityBase):
|
| type_: SecuritySchemeType = Field(default=SecuritySchemeType.oauth2, alias="type")
|
| flows: OAuthFlows
|
|
|
|
|
| class OpenIdConnect(SecurityBase):
|
| type_: SecuritySchemeType = Field(
|
| default=SecuritySchemeType.openIdConnect, alias="type"
|
| )
|
| openIdConnectUrl: str
|
|
|
|
|
| SecurityScheme = Union[APIKey, HTTPBase, OAuth2, OpenIdConnect, HTTPBearer]
|
|
|
|
|
| class Components(BaseModelWithConfig):
|
| schemas: Optional[dict[str, Union[Schema, Reference]]] = None
|
| responses: Optional[dict[str, Union[Response, Reference]]] = None
|
| parameters: Optional[dict[str, Union[Parameter, Reference]]] = None
|
| examples: Optional[dict[str, Union[Example, Reference]]] = None
|
| requestBodies: Optional[dict[str, Union[RequestBody, Reference]]] = None
|
| headers: Optional[dict[str, Union[Header, Reference]]] = None
|
| securitySchemes: Optional[dict[str, Union[SecurityScheme, Reference]]] = None
|
| links: Optional[dict[str, Union[Link, Reference]]] = None
|
|
|
| callbacks: Optional[dict[str, Union[dict[str, PathItem], Reference, Any]]] = None
|
| pathItems: Optional[dict[str, Union[PathItem, Reference]]] = None
|
|
|
|
|
| class Tag(BaseModelWithConfig):
|
| name: str
|
| description: Optional[str] = None
|
| externalDocs: Optional[ExternalDocumentation] = None
|
|
|
|
|
| class OpenAPI(BaseModelWithConfig):
|
| openapi: str
|
| info: Info
|
| jsonSchemaDialect: Optional[str] = None
|
| servers: Optional[list[Server]] = None
|
|
|
| paths: Optional[dict[str, Union[PathItem, Any]]] = None
|
| webhooks: Optional[dict[str, Union[PathItem, Reference]]] = None
|
| components: Optional[Components] = None
|
| security: Optional[list[dict[str, list[str]]]] = None
|
| tags: Optional[list[Tag]] = None
|
| externalDocs: Optional[ExternalDocumentation] = None
|
|
|
|
|
| Schema.model_rebuild()
|
| Operation.model_rebuild()
|
| Encoding.model_rebuild()
|
|
|