|
from enum import Enum |
|
from typing import Any, Callable, Dict, Iterable, List, Optional, Set, Type, Union |
|
|
|
from fastapi._compat import ( |
|
PYDANTIC_V2, |
|
CoreSchema, |
|
GetJsonSchemaHandler, |
|
JsonSchemaValue, |
|
_model_rebuild, |
|
general_plain_validator_function, |
|
) |
|
from fastapi.logger import logger |
|
from pydantic import AnyUrl, BaseModel, Field |
|
from typing_extensions import Annotated, 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: CoreSchema, handler: GetJsonSchemaHandler |
|
) -> JsonSchemaValue: |
|
return {"type": "string", "format": "email"} |
|
|
|
@classmethod |
|
def __get_pydantic_core_schema__( |
|
cls, source: Type[Any], handler: Callable[[Any], CoreSchema] |
|
) -> CoreSchema: |
|
return general_plain_validator_function(cls._validate) |
|
|
|
|
|
class Contact(BaseModel): |
|
name: Optional[str] = None |
|
url: Optional[AnyUrl] = None |
|
email: Optional[EmailStr] = None |
|
|
|
if PYDANTIC_V2: |
|
model_config = {"extra": "allow"} |
|
|
|
else: |
|
|
|
class Config: |
|
extra = "allow" |
|
|
|
|
|
class License(BaseModel): |
|
name: str |
|
identifier: Optional[str] = None |
|
url: Optional[AnyUrl] = None |
|
|
|
if PYDANTIC_V2: |
|
model_config = {"extra": "allow"} |
|
|
|
else: |
|
|
|
class Config: |
|
extra = "allow" |
|
|
|
|
|
class Info(BaseModel): |
|
title: str |
|
summary: Optional[str] = None |
|
description: Optional[str] = None |
|
termsOfService: Optional[str] = None |
|
contact: Optional[Contact] = None |
|
license: Optional[License] = None |
|
version: str |
|
|
|
if PYDANTIC_V2: |
|
model_config = {"extra": "allow"} |
|
|
|
else: |
|
|
|
class Config: |
|
extra = "allow" |
|
|
|
|
|
class ServerVariable(BaseModel): |
|
enum: Annotated[Optional[List[str]], Field(min_length=1)] = None |
|
default: str |
|
description: Optional[str] = None |
|
|
|
if PYDANTIC_V2: |
|
model_config = {"extra": "allow"} |
|
|
|
else: |
|
|
|
class Config: |
|
extra = "allow" |
|
|
|
|
|
class Server(BaseModel): |
|
url: Union[AnyUrl, str] |
|
description: Optional[str] = None |
|
variables: Optional[Dict[str, ServerVariable]] = None |
|
|
|
if PYDANTIC_V2: |
|
model_config = {"extra": "allow"} |
|
|
|
else: |
|
|
|
class Config: |
|
extra = "allow" |
|
|
|
|
|
class Reference(BaseModel): |
|
ref: str = Field(alias="$ref") |
|
|
|
|
|
class Discriminator(BaseModel): |
|
propertyName: str |
|
mapping: Optional[Dict[str, str]] = None |
|
|
|
|
|
class XML(BaseModel): |
|
name: Optional[str] = None |
|
namespace: Optional[str] = None |
|
prefix: Optional[str] = None |
|
attribute: Optional[bool] = None |
|
wrapped: Optional[bool] = None |
|
|
|
if PYDANTIC_V2: |
|
model_config = {"extra": "allow"} |
|
|
|
else: |
|
|
|
class Config: |
|
extra = "allow" |
|
|
|
|
|
class ExternalDocumentation(BaseModel): |
|
description: Optional[str] = None |
|
url: AnyUrl |
|
|
|
if PYDANTIC_V2: |
|
model_config = {"extra": "allow"} |
|
|
|
else: |
|
|
|
class Config: |
|
extra = "allow" |
|
|
|
|
|
class Schema(BaseModel): |
|
|
|
|
|
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[Union["SchemaOrBool", List["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[str] = 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 |
|
|
|
if PYDANTIC_V2: |
|
model_config = {"extra": "allow"} |
|
|
|
else: |
|
|
|
class Config: |
|
extra = "allow" |
|
|
|
|
|
|
|
|
|
SchemaOrBool = Union[Schema, bool] |
|
|
|
|
|
class Example(TypedDict, total=False): |
|
summary: Optional[str] |
|
description: Optional[str] |
|
value: Optional[Any] |
|
externalValue: Optional[AnyUrl] |
|
|
|
if PYDANTIC_V2: |
|
__pydantic_config__ = {"extra": "allow"} |
|
|
|
else: |
|
|
|
class Config: |
|
extra = "allow" |
|
|
|
|
|
class ParameterInType(Enum): |
|
query = "query" |
|
header = "header" |
|
path = "path" |
|
cookie = "cookie" |
|
|
|
|
|
class Encoding(BaseModel): |
|
contentType: Optional[str] = None |
|
headers: Optional[Dict[str, Union["Header", Reference]]] = None |
|
style: Optional[str] = None |
|
explode: Optional[bool] = None |
|
allowReserved: Optional[bool] = None |
|
|
|
if PYDANTIC_V2: |
|
model_config = {"extra": "allow"} |
|
|
|
else: |
|
|
|
class Config: |
|
extra = "allow" |
|
|
|
|
|
class MediaType(BaseModel): |
|
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 |
|
|
|
if PYDANTIC_V2: |
|
model_config = {"extra": "allow"} |
|
|
|
else: |
|
|
|
class Config: |
|
extra = "allow" |
|
|
|
|
|
class ParameterBase(BaseModel): |
|
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 |
|
|
|
if PYDANTIC_V2: |
|
model_config = {"extra": "allow"} |
|
|
|
else: |
|
|
|
class Config: |
|
extra = "allow" |
|
|
|
|
|
class Parameter(ParameterBase): |
|
name: str |
|
in_: ParameterInType = Field(alias="in") |
|
|
|
|
|
class Header(ParameterBase): |
|
pass |
|
|
|
|
|
class RequestBody(BaseModel): |
|
description: Optional[str] = None |
|
content: Dict[str, MediaType] |
|
required: Optional[bool] = None |
|
|
|
if PYDANTIC_V2: |
|
model_config = {"extra": "allow"} |
|
|
|
else: |
|
|
|
class Config: |
|
extra = "allow" |
|
|
|
|
|
class Link(BaseModel): |
|
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 |
|
|
|
if PYDANTIC_V2: |
|
model_config = {"extra": "allow"} |
|
|
|
else: |
|
|
|
class Config: |
|
extra = "allow" |
|
|
|
|
|
class Response(BaseModel): |
|
description: str |
|
headers: Optional[Dict[str, Union[Header, Reference]]] = None |
|
content: Optional[Dict[str, MediaType]] = None |
|
links: Optional[Dict[str, Union[Link, Reference]]] = None |
|
|
|
if PYDANTIC_V2: |
|
model_config = {"extra": "allow"} |
|
|
|
else: |
|
|
|
class Config: |
|
extra = "allow" |
|
|
|
|
|
class Operation(BaseModel): |
|
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 |
|
|
|
if PYDANTIC_V2: |
|
model_config = {"extra": "allow"} |
|
|
|
else: |
|
|
|
class Config: |
|
extra = "allow" |
|
|
|
|
|
class PathItem(BaseModel): |
|
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 |
|
|
|
if PYDANTIC_V2: |
|
model_config = {"extra": "allow"} |
|
|
|
else: |
|
|
|
class Config: |
|
extra = "allow" |
|
|
|
|
|
class SecuritySchemeType(Enum): |
|
apiKey = "apiKey" |
|
http = "http" |
|
oauth2 = "oauth2" |
|
openIdConnect = "openIdConnect" |
|
|
|
|
|
class SecurityBase(BaseModel): |
|
type_: SecuritySchemeType = Field(alias="type") |
|
description: Optional[str] = None |
|
|
|
if PYDANTIC_V2: |
|
model_config = {"extra": "allow"} |
|
|
|
else: |
|
|
|
class Config: |
|
extra = "allow" |
|
|
|
|
|
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(BaseModel): |
|
refreshUrl: Optional[str] = None |
|
scopes: Dict[str, str] = {} |
|
|
|
if PYDANTIC_V2: |
|
model_config = {"extra": "allow"} |
|
|
|
else: |
|
|
|
class Config: |
|
extra = "allow" |
|
|
|
|
|
class OAuthFlowImplicit(OAuthFlow): |
|
authorizationUrl: str |
|
|
|
|
|
class OAuthFlowPassword(OAuthFlow): |
|
tokenUrl: str |
|
|
|
|
|
class OAuthFlowClientCredentials(OAuthFlow): |
|
tokenUrl: str |
|
|
|
|
|
class OAuthFlowAuthorizationCode(OAuthFlow): |
|
authorizationUrl: str |
|
tokenUrl: str |
|
|
|
|
|
class OAuthFlows(BaseModel): |
|
implicit: Optional[OAuthFlowImplicit] = None |
|
password: Optional[OAuthFlowPassword] = None |
|
clientCredentials: Optional[OAuthFlowClientCredentials] = None |
|
authorizationCode: Optional[OAuthFlowAuthorizationCode] = None |
|
|
|
if PYDANTIC_V2: |
|
model_config = {"extra": "allow"} |
|
|
|
else: |
|
|
|
class Config: |
|
extra = "allow" |
|
|
|
|
|
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(BaseModel): |
|
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 |
|
|
|
if PYDANTIC_V2: |
|
model_config = {"extra": "allow"} |
|
|
|
else: |
|
|
|
class Config: |
|
extra = "allow" |
|
|
|
|
|
class Tag(BaseModel): |
|
name: str |
|
description: Optional[str] = None |
|
externalDocs: Optional[ExternalDocumentation] = None |
|
|
|
if PYDANTIC_V2: |
|
model_config = {"extra": "allow"} |
|
|
|
else: |
|
|
|
class Config: |
|
extra = "allow" |
|
|
|
|
|
class OpenAPI(BaseModel): |
|
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 |
|
|
|
if PYDANTIC_V2: |
|
model_config = {"extra": "allow"} |
|
|
|
else: |
|
|
|
class Config: |
|
extra = "allow" |
|
|
|
|
|
_model_rebuild(Schema) |
|
_model_rebuild(Operation) |
|
_model_rebuild(Encoding) |
|
|