| from collections.abc import Callable, Iterable, Mapping |
| from enum import Enum |
| from typing import Annotated, Any, Literal, 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 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: str | None = None |
| url: AnyUrl | None = None |
| email: EmailStr | None = None |
|
|
|
|
| class License(BaseModelWithConfig): |
| name: str |
| identifier: str | None = None |
| url: AnyUrl | None = None |
|
|
|
|
| class Info(BaseModelWithConfig): |
| title: str |
| summary: str | None = None |
| description: str | None = None |
| termsOfService: str | None = None |
| contact: Contact | None = None |
| license: License | None = None |
| version: str |
|
|
|
|
| class ServerVariable(BaseModelWithConfig): |
| enum: Annotated[list[str] | None, Field(min_length=1)] = None |
| default: str |
| description: str | None = None |
|
|
|
|
| class Server(BaseModelWithConfig): |
| url: AnyUrl | str |
| description: str | None = None |
| variables: dict[str, ServerVariable] | None = None |
|
|
|
|
| class Reference(BaseModel): |
| ref: str = Field(alias="$ref") |
|
|
|
|
| class Discriminator(BaseModel): |
| propertyName: str |
| mapping: dict[str, str] | None = None |
|
|
|
|
| class XML(BaseModelWithConfig): |
| name: str | None = None |
| namespace: str | None = None |
| prefix: str | None = None |
| attribute: bool | None = None |
| wrapped: bool | None = None |
|
|
|
|
| class ExternalDocumentation(BaseModelWithConfig): |
| description: str | None = None |
| url: AnyUrl |
|
|
|
|
| |
| SchemaType = Literal[ |
| "array", "boolean", "integer", "null", "number", "object", "string" |
| ] |
|
|
|
|
| class Schema(BaseModelWithConfig): |
| |
| |
| schema_: str | None = Field(default=None, alias="$schema") |
| vocabulary: str | None = Field(default=None, alias="$vocabulary") |
| id: str | None = Field(default=None, alias="$id") |
| anchor: str | None = Field(default=None, alias="$anchor") |
| dynamicAnchor: str | None = Field(default=None, alias="$dynamicAnchor") |
| ref: str | None = Field(default=None, alias="$ref") |
| dynamicRef: str | None = Field(default=None, alias="$dynamicRef") |
| defs: dict[str, "SchemaOrBool"] | None = Field(default=None, alias="$defs") |
| comment: str | None = Field(default=None, alias="$comment") |
| |
| |
| allOf: list["SchemaOrBool"] | None = None |
| anyOf: list["SchemaOrBool"] | None = None |
| oneOf: list["SchemaOrBool"] | None = 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: dict[str, "SchemaOrBool"] | None = None |
| prefixItems: list["SchemaOrBool"] | None = None |
| items: Optional["SchemaOrBool"] = None |
| contains: Optional["SchemaOrBool"] = None |
| properties: dict[str, "SchemaOrBool"] | None = None |
| patternProperties: dict[str, "SchemaOrBool"] | None = None |
| additionalProperties: Optional["SchemaOrBool"] = None |
| propertyNames: Optional["SchemaOrBool"] = None |
| unevaluatedItems: Optional["SchemaOrBool"] = None |
| unevaluatedProperties: Optional["SchemaOrBool"] = None |
| |
| |
| type: SchemaType | list[SchemaType] | None = None |
| enum: list[Any] | None = None |
| const: Any | None = None |
| multipleOf: float | None = Field(default=None, gt=0) |
| maximum: float | None = None |
| exclusiveMaximum: float | None = None |
| minimum: float | None = None |
| exclusiveMinimum: float | None = None |
| maxLength: int | None = Field(default=None, ge=0) |
| minLength: int | None = Field(default=None, ge=0) |
| pattern: str | None = None |
| maxItems: int | None = Field(default=None, ge=0) |
| minItems: int | None = Field(default=None, ge=0) |
| uniqueItems: bool | None = None |
| maxContains: int | None = Field(default=None, ge=0) |
| minContains: int | None = Field(default=None, ge=0) |
| maxProperties: int | None = Field(default=None, ge=0) |
| minProperties: int | None = Field(default=None, ge=0) |
| required: list[str] | None = None |
| dependentRequired: dict[str, set[str]] | None = None |
| |
| |
| format: str | None = None |
| |
| |
| contentEncoding: str | None = None |
| contentMediaType: str | None = None |
| contentSchema: Optional["SchemaOrBool"] = None |
| |
| |
| title: str | None = None |
| description: str | None = None |
| default: Any | None = None |
| deprecated: bool | None = None |
| readOnly: bool | None = None |
| writeOnly: bool | None = None |
| examples: list[Any] | None = None |
| |
| |
| discriminator: Discriminator | None = None |
| xml: XML | None = None |
| externalDocs: ExternalDocumentation | None = None |
| example: Annotated[ |
| Any | None, |
| typing_deprecated( |
| "Deprecated in OpenAPI 3.1.0 that now uses JSON Schema 2020-12, " |
| "although still supported. Use examples instead." |
| ), |
| ] = None |
|
|
|
|
| |
| |
| SchemaOrBool = Schema | bool |
|
|
|
|
| class Example(TypedDict, total=False): |
| summary: str | None |
| description: str | None |
| value: Any | None |
| externalValue: AnyUrl | None |
|
|
| __pydantic_config__ = {"extra": "allow"} |
|
|
|
|
| class ParameterInType(Enum): |
| query = "query" |
| header = "header" |
| path = "path" |
| cookie = "cookie" |
|
|
|
|
| class Encoding(BaseModelWithConfig): |
| contentType: str | None = None |
| headers: dict[str, Union["Header", Reference]] | None = None |
| style: str | None = None |
| explode: bool | None = None |
| allowReserved: bool | None = None |
|
|
|
|
| class MediaType(BaseModelWithConfig): |
| schema_: Schema | Reference | None = Field(default=None, alias="schema") |
| example: Any | None = None |
| examples: dict[str, Example | Reference] | None = None |
| encoding: dict[str, Encoding] | None = None |
|
|
|
|
| class ParameterBase(BaseModelWithConfig): |
| description: str | None = None |
| required: bool | None = None |
| deprecated: bool | None = None |
| |
| style: str | None = None |
| explode: bool | None = None |
| allowReserved: bool | None = None |
| schema_: Schema | Reference | None = Field(default=None, alias="schema") |
| example: Any | None = None |
| examples: dict[str, Example | Reference] | None = None |
| |
| content: dict[str, MediaType] | None = None |
|
|
|
|
| class Parameter(ParameterBase): |
| name: str |
| in_: ParameterInType = Field(alias="in") |
|
|
|
|
| class Header(ParameterBase): |
| pass |
|
|
|
|
| class RequestBody(BaseModelWithConfig): |
| description: str | None = None |
| content: dict[str, MediaType] |
| required: bool | None = None |
|
|
|
|
| class Link(BaseModelWithConfig): |
| operationRef: str | None = None |
| operationId: str | None = None |
| parameters: dict[str, Any | str] | None = None |
| requestBody: Any | str | None = None |
| description: str | None = None |
| server: Server | None = None |
|
|
|
|
| class Response(BaseModelWithConfig): |
| description: str |
| headers: dict[str, Header | Reference] | None = None |
| content: dict[str, MediaType] | None = None |
| links: dict[str, Link | Reference] | None = None |
|
|
|
|
| class Operation(BaseModelWithConfig): |
| tags: list[str] | None = None |
| summary: str | None = None |
| description: str | None = None |
| externalDocs: ExternalDocumentation | None = None |
| operationId: str | None = None |
| parameters: list[Parameter | Reference] | None = None |
| requestBody: RequestBody | Reference | None = None |
| |
| responses: dict[str, Response | Any] | None = None |
| callbacks: dict[str, dict[str, "PathItem"] | Reference] | None = None |
| deprecated: bool | None = None |
| security: list[dict[str, list[str]]] | None = None |
| servers: list[Server] | None = None |
|
|
|
|
| class PathItem(BaseModelWithConfig): |
| ref: str | None = Field(default=None, alias="$ref") |
| summary: str | None = None |
| description: str | None = None |
| get: Operation | None = None |
| put: Operation | None = None |
| post: Operation | None = None |
| delete: Operation | None = None |
| options: Operation | None = None |
| head: Operation | None = None |
| patch: Operation | None = None |
| trace: Operation | None = None |
| servers: list[Server] | None = None |
| parameters: list[Parameter | Reference] | None = None |
|
|
|
|
| class SecuritySchemeType(Enum): |
| apiKey = "apiKey" |
| http = "http" |
| oauth2 = "oauth2" |
| openIdConnect = "openIdConnect" |
|
|
|
|
| class SecurityBase(BaseModelWithConfig): |
| type_: SecuritySchemeType = Field(alias="type") |
| description: str | None = 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: str | None = None |
|
|
|
|
| class OAuthFlow(BaseModelWithConfig): |
| refreshUrl: str | None = 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: OAuthFlowImplicit | None = None |
| password: OAuthFlowPassword | None = None |
| clientCredentials: OAuthFlowClientCredentials | None = None |
| authorizationCode: OAuthFlowAuthorizationCode | None = 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 = APIKey | HTTPBase | OAuth2 | OpenIdConnect | HTTPBearer |
|
|
|
|
| class Components(BaseModelWithConfig): |
| schemas: dict[str, Schema | Reference] | None = None |
| responses: dict[str, Response | Reference] | None = None |
| parameters: dict[str, Parameter | Reference] | None = None |
| examples: dict[str, Example | Reference] | None = None |
| requestBodies: dict[str, RequestBody | Reference] | None = None |
| headers: dict[str, Header | Reference] | None = None |
| securitySchemes: dict[str, SecurityScheme | Reference] | None = None |
| links: dict[str, Link | Reference] | None = None |
| |
| callbacks: dict[str, dict[str, PathItem] | Reference | Any] | None = None |
| pathItems: dict[str, PathItem | Reference] | None = None |
|
|
|
|
| class Tag(BaseModelWithConfig): |
| name: str |
| description: str | None = None |
| externalDocs: ExternalDocumentation | None = None |
|
|
|
|
| class OpenAPI(BaseModelWithConfig): |
| openapi: str |
| info: Info |
| jsonSchemaDialect: str | None = None |
| servers: list[Server] | None = None |
| |
| paths: dict[str, PathItem | Any] | None = None |
| webhooks: dict[str, PathItem | Reference] | None = None |
| components: Components | None = None |
| security: list[dict[str, list[str]]] | None = None |
| tags: list[Tag] | None = None |
| externalDocs: ExternalDocumentation | None = None |
|
|
|
|
| Schema.model_rebuild() |
| Operation.model_rebuild() |
| Encoding.model_rebuild() |
|
|