models.py 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435
  1. from collections.abc import Callable, Iterable, Mapping
  2. from enum import Enum
  3. from typing import Annotated, Any, Literal, Optional, Union
  4. from fastapi._compat import with_info_plain_validator_function
  5. from fastapi.logger import logger
  6. from pydantic import (
  7. AnyUrl,
  8. BaseModel,
  9. Field,
  10. GetJsonSchemaHandler,
  11. )
  12. from typing_extensions import TypedDict
  13. from typing_extensions import deprecated as typing_deprecated
  14. try:
  15. import email_validator
  16. assert email_validator # make autoflake ignore the unused import
  17. from pydantic import EmailStr
  18. except ImportError: # pragma: no cover
  19. class EmailStr(str): # type: ignore # ty: ignore[unused-ignore-comment]
  20. @classmethod
  21. def __get_validators__(cls) -> Iterable[Callable[..., Any]]:
  22. yield cls.validate
  23. @classmethod
  24. def validate(cls, v: Any) -> str:
  25. logger.warning(
  26. "email-validator not installed, email fields will be treated as str.\n"
  27. "To install, run: pip install email-validator"
  28. )
  29. return str(v)
  30. @classmethod
  31. def _validate(cls, __input_value: Any, _: Any) -> str:
  32. logger.warning(
  33. "email-validator not installed, email fields will be treated as str.\n"
  34. "To install, run: pip install email-validator"
  35. )
  36. return str(__input_value)
  37. @classmethod
  38. def __get_pydantic_json_schema__(
  39. cls, core_schema: Mapping[str, Any], handler: GetJsonSchemaHandler
  40. ) -> dict[str, Any]:
  41. return {"type": "string", "format": "email"}
  42. @classmethod
  43. def __get_pydantic_core_schema__(
  44. cls, source: type[Any], handler: Callable[[Any], Mapping[str, Any]]
  45. ) -> Mapping[str, Any]:
  46. return with_info_plain_validator_function(cls._validate)
  47. class BaseModelWithConfig(BaseModel):
  48. model_config = {"extra": "allow"}
  49. class Contact(BaseModelWithConfig):
  50. name: str | None = None
  51. url: AnyUrl | None = None
  52. email: EmailStr | None = None
  53. class License(BaseModelWithConfig):
  54. name: str
  55. identifier: str | None = None
  56. url: AnyUrl | None = None
  57. class Info(BaseModelWithConfig):
  58. title: str
  59. summary: str | None = None
  60. description: str | None = None
  61. termsOfService: str | None = None
  62. contact: Contact | None = None
  63. license: License | None = None
  64. version: str
  65. class ServerVariable(BaseModelWithConfig):
  66. enum: Annotated[list[str] | None, Field(min_length=1)] = None
  67. default: str
  68. description: str | None = None
  69. class Server(BaseModelWithConfig):
  70. url: AnyUrl | str
  71. description: str | None = None
  72. variables: dict[str, ServerVariable] | None = None
  73. class Reference(BaseModel):
  74. ref: str = Field(alias="$ref")
  75. class Discriminator(BaseModel):
  76. propertyName: str
  77. mapping: dict[str, str] | None = None
  78. class XML(BaseModelWithConfig):
  79. name: str | None = None
  80. namespace: str | None = None
  81. prefix: str | None = None
  82. attribute: bool | None = None
  83. wrapped: bool | None = None
  84. class ExternalDocumentation(BaseModelWithConfig):
  85. description: str | None = None
  86. url: AnyUrl
  87. # Ref JSON Schema 2020-12: https://json-schema.org/draft/2020-12/json-schema-validation#name-type
  88. SchemaType = Literal[
  89. "array", "boolean", "integer", "null", "number", "object", "string"
  90. ]
  91. class Schema(BaseModelWithConfig):
  92. # Ref: JSON Schema 2020-12: https://json-schema.org/draft/2020-12/json-schema-core.html#name-the-json-schema-core-vocabu
  93. # Core Vocabulary
  94. schema_: str | None = Field(default=None, alias="$schema")
  95. vocabulary: str | None = Field(default=None, alias="$vocabulary")
  96. id: str | None = Field(default=None, alias="$id")
  97. anchor: str | None = Field(default=None, alias="$anchor")
  98. dynamicAnchor: str | None = Field(default=None, alias="$dynamicAnchor")
  99. ref: str | None = Field(default=None, alias="$ref")
  100. dynamicRef: str | None = Field(default=None, alias="$dynamicRef")
  101. defs: dict[str, "SchemaOrBool"] | None = Field(default=None, alias="$defs")
  102. comment: str | None = Field(default=None, alias="$comment")
  103. # Ref: JSON Schema 2020-12: https://json-schema.org/draft/2020-12/json-schema-core.html#name-a-vocabulary-for-applying-s
  104. # A Vocabulary for Applying Subschemas
  105. allOf: list["SchemaOrBool"] | None = None
  106. anyOf: list["SchemaOrBool"] | None = None
  107. oneOf: list["SchemaOrBool"] | None = None
  108. not_: Optional["SchemaOrBool"] = Field(default=None, alias="not")
  109. if_: Optional["SchemaOrBool"] = Field(default=None, alias="if")
  110. then: Optional["SchemaOrBool"] = None
  111. else_: Optional["SchemaOrBool"] = Field(default=None, alias="else")
  112. dependentSchemas: dict[str, "SchemaOrBool"] | None = None
  113. prefixItems: list["SchemaOrBool"] | None = None
  114. items: Optional["SchemaOrBool"] = None
  115. contains: Optional["SchemaOrBool"] = None
  116. properties: dict[str, "SchemaOrBool"] | None = None
  117. patternProperties: dict[str, "SchemaOrBool"] | None = None
  118. additionalProperties: Optional["SchemaOrBool"] = None
  119. propertyNames: Optional["SchemaOrBool"] = None
  120. unevaluatedItems: Optional["SchemaOrBool"] = None
  121. unevaluatedProperties: Optional["SchemaOrBool"] = None
  122. # Ref: JSON Schema Validation 2020-12: https://json-schema.org/draft/2020-12/json-schema-validation.html#name-a-vocabulary-for-structural
  123. # A Vocabulary for Structural Validation
  124. type: SchemaType | list[SchemaType] | None = None
  125. enum: list[Any] | None = None
  126. const: Any | None = None
  127. multipleOf: float | None = Field(default=None, gt=0)
  128. maximum: float | None = None
  129. exclusiveMaximum: float | None = None
  130. minimum: float | None = None
  131. exclusiveMinimum: float | None = None
  132. maxLength: int | None = Field(default=None, ge=0)
  133. minLength: int | None = Field(default=None, ge=0)
  134. pattern: str | None = None
  135. maxItems: int | None = Field(default=None, ge=0)
  136. minItems: int | None = Field(default=None, ge=0)
  137. uniqueItems: bool | None = None
  138. maxContains: int | None = Field(default=None, ge=0)
  139. minContains: int | None = Field(default=None, ge=0)
  140. maxProperties: int | None = Field(default=None, ge=0)
  141. minProperties: int | None = Field(default=None, ge=0)
  142. required: list[str] | None = None
  143. dependentRequired: dict[str, set[str]] | None = None
  144. # Ref: JSON Schema Validation 2020-12: https://json-schema.org/draft/2020-12/json-schema-validation.html#name-vocabularies-for-semantic-c
  145. # Vocabularies for Semantic Content With "format"
  146. format: str | None = None
  147. # Ref: JSON Schema Validation 2020-12: https://json-schema.org/draft/2020-12/json-schema-validation.html#name-a-vocabulary-for-the-conten
  148. # A Vocabulary for the Contents of String-Encoded Data
  149. contentEncoding: str | None = None
  150. contentMediaType: str | None = None
  151. contentSchema: Optional["SchemaOrBool"] = None
  152. # Ref: JSON Schema Validation 2020-12: https://json-schema.org/draft/2020-12/json-schema-validation.html#name-a-vocabulary-for-basic-meta
  153. # A Vocabulary for Basic Meta-Data Annotations
  154. title: str | None = None
  155. description: str | None = None
  156. default: Any | None = None
  157. deprecated: bool | None = None
  158. readOnly: bool | None = None
  159. writeOnly: bool | None = None
  160. examples: list[Any] | None = None
  161. # Ref: OpenAPI 3.1.0: https://github.com/OAI/OpenAPI-Specification/blob/main/versions/3.1.0.md#schema-object
  162. # Schema Object
  163. discriminator: Discriminator | None = None
  164. xml: XML | None = None
  165. externalDocs: ExternalDocumentation | None = None
  166. example: Annotated[
  167. Any | None,
  168. typing_deprecated(
  169. "Deprecated in OpenAPI 3.1.0 that now uses JSON Schema 2020-12, "
  170. "although still supported. Use examples instead."
  171. ),
  172. ] = None
  173. # Ref: https://json-schema.org/draft/2020-12/json-schema-core.html#name-json-schema-documents
  174. # A JSON Schema MUST be an object or a boolean.
  175. SchemaOrBool = Schema | bool
  176. class Example(TypedDict, total=False):
  177. summary: str | None
  178. description: str | None
  179. value: Any | None
  180. externalValue: AnyUrl | None
  181. __pydantic_config__ = {"extra": "allow"} # type: ignore[misc]
  182. class ParameterInType(Enum):
  183. query = "query"
  184. header = "header"
  185. path = "path"
  186. cookie = "cookie"
  187. class Encoding(BaseModelWithConfig):
  188. contentType: str | None = None
  189. headers: dict[str, Union["Header", Reference]] | None = None
  190. style: str | None = None
  191. explode: bool | None = None
  192. allowReserved: bool | None = None
  193. class MediaType(BaseModelWithConfig):
  194. schema_: Schema | Reference | None = Field(default=None, alias="schema")
  195. example: Any | None = None
  196. examples: dict[str, Example | Reference] | None = None
  197. encoding: dict[str, Encoding] | None = None
  198. class ParameterBase(BaseModelWithConfig):
  199. description: str | None = None
  200. required: bool | None = None
  201. deprecated: bool | None = None
  202. # Serialization rules for simple scenarios
  203. style: str | None = None
  204. explode: bool | None = None
  205. allowReserved: bool | None = None
  206. schema_: Schema | Reference | None = Field(default=None, alias="schema")
  207. example: Any | None = None
  208. examples: dict[str, Example | Reference] | None = None
  209. # Serialization rules for more complex scenarios
  210. content: dict[str, MediaType] | None = None
  211. class Parameter(ParameterBase):
  212. name: str
  213. in_: ParameterInType = Field(alias="in")
  214. class Header(ParameterBase):
  215. pass
  216. class RequestBody(BaseModelWithConfig):
  217. description: str | None = None
  218. content: dict[str, MediaType]
  219. required: bool | None = None
  220. class Link(BaseModelWithConfig):
  221. operationRef: str | None = None
  222. operationId: str | None = None
  223. parameters: dict[str, Any | str] | None = None
  224. requestBody: Any | str | None = None
  225. description: str | None = None
  226. server: Server | None = None
  227. class Response(BaseModelWithConfig):
  228. description: str
  229. headers: dict[str, Header | Reference] | None = None
  230. content: dict[str, MediaType] | None = None
  231. links: dict[str, Link | Reference] | None = None
  232. class Operation(BaseModelWithConfig):
  233. tags: list[str] | None = None
  234. summary: str | None = None
  235. description: str | None = None
  236. externalDocs: ExternalDocumentation | None = None
  237. operationId: str | None = None
  238. parameters: list[Parameter | Reference] | None = None
  239. requestBody: RequestBody | Reference | None = None
  240. # Using Any for Specification Extensions
  241. responses: dict[str, Response | Any] | None = None
  242. callbacks: dict[str, dict[str, "PathItem"] | Reference] | None = None
  243. deprecated: bool | None = None
  244. security: list[dict[str, list[str]]] | None = None
  245. servers: list[Server] | None = None
  246. class PathItem(BaseModelWithConfig):
  247. ref: str | None = Field(default=None, alias="$ref")
  248. summary: str | None = None
  249. description: str | None = None
  250. get: Operation | None = None
  251. put: Operation | None = None
  252. post: Operation | None = None
  253. delete: Operation | None = None
  254. options: Operation | None = None
  255. head: Operation | None = None
  256. patch: Operation | None = None
  257. trace: Operation | None = None
  258. servers: list[Server] | None = None
  259. parameters: list[Parameter | Reference] | None = None
  260. class SecuritySchemeType(Enum):
  261. apiKey = "apiKey"
  262. http = "http"
  263. oauth2 = "oauth2"
  264. openIdConnect = "openIdConnect"
  265. class SecurityBase(BaseModelWithConfig):
  266. type_: SecuritySchemeType = Field(alias="type")
  267. description: str | None = None
  268. class APIKeyIn(Enum):
  269. query = "query"
  270. header = "header"
  271. cookie = "cookie"
  272. class APIKey(SecurityBase):
  273. type_: SecuritySchemeType = Field(default=SecuritySchemeType.apiKey, alias="type")
  274. in_: APIKeyIn = Field(alias="in")
  275. name: str
  276. class HTTPBase(SecurityBase):
  277. type_: SecuritySchemeType = Field(default=SecuritySchemeType.http, alias="type")
  278. scheme: str
  279. class HTTPBearer(HTTPBase):
  280. scheme: Literal["bearer"] = "bearer"
  281. bearerFormat: str | None = None
  282. class OAuthFlow(BaseModelWithConfig):
  283. refreshUrl: str | None = None
  284. scopes: dict[str, str] = {}
  285. class OAuthFlowImplicit(OAuthFlow):
  286. authorizationUrl: str
  287. class OAuthFlowPassword(OAuthFlow):
  288. tokenUrl: str
  289. class OAuthFlowClientCredentials(OAuthFlow):
  290. tokenUrl: str
  291. class OAuthFlowAuthorizationCode(OAuthFlow):
  292. authorizationUrl: str
  293. tokenUrl: str
  294. class OAuthFlows(BaseModelWithConfig):
  295. implicit: OAuthFlowImplicit | None = None
  296. password: OAuthFlowPassword | None = None
  297. clientCredentials: OAuthFlowClientCredentials | None = None
  298. authorizationCode: OAuthFlowAuthorizationCode | None = None
  299. class OAuth2(SecurityBase):
  300. type_: SecuritySchemeType = Field(default=SecuritySchemeType.oauth2, alias="type")
  301. flows: OAuthFlows
  302. class OpenIdConnect(SecurityBase):
  303. type_: SecuritySchemeType = Field(
  304. default=SecuritySchemeType.openIdConnect, alias="type"
  305. )
  306. openIdConnectUrl: str
  307. SecurityScheme = APIKey | HTTPBase | OAuth2 | OpenIdConnect | HTTPBearer
  308. class Components(BaseModelWithConfig):
  309. schemas: dict[str, Schema | Reference] | None = None
  310. responses: dict[str, Response | Reference] | None = None
  311. parameters: dict[str, Parameter | Reference] | None = None
  312. examples: dict[str, Example | Reference] | None = None
  313. requestBodies: dict[str, RequestBody | Reference] | None = None
  314. headers: dict[str, Header | Reference] | None = None
  315. securitySchemes: dict[str, SecurityScheme | Reference] | None = None
  316. links: dict[str, Link | Reference] | None = None
  317. # Using Any for Specification Extensions
  318. callbacks: dict[str, dict[str, PathItem] | Reference | Any] | None = None
  319. pathItems: dict[str, PathItem | Reference] | None = None
  320. class Tag(BaseModelWithConfig):
  321. name: str
  322. description: str | None = None
  323. externalDocs: ExternalDocumentation | None = None
  324. class OpenAPI(BaseModelWithConfig):
  325. openapi: str
  326. info: Info
  327. jsonSchemaDialect: str | None = None
  328. servers: list[Server] | None = None
  329. # Using Any for Specification Extensions
  330. paths: dict[str, PathItem | Any] | None = None
  331. webhooks: dict[str, PathItem | Reference] | None = None
  332. components: Components | None = None
  333. security: list[dict[str, list[str]]] | None = None
  334. tags: list[Tag] | None = None
  335. externalDocs: ExternalDocumentation | None = None
  336. Schema.model_rebuild()
  337. Operation.model_rebuild()
  338. Encoding.model_rebuild()