utils.py 26 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606
  1. import copy
  2. import http.client
  3. import inspect
  4. import warnings
  5. from collections.abc import Sequence
  6. from typing import Any, Literal, cast
  7. from fastapi import routing
  8. from fastapi._compat import (
  9. ModelField,
  10. get_definitions,
  11. get_flat_models_from_fields,
  12. get_model_name_map,
  13. get_schema_from_model_field,
  14. lenient_issubclass,
  15. )
  16. from fastapi.datastructures import DefaultPlaceholder, _Unset
  17. from fastapi.dependencies.models import Dependant
  18. from fastapi.dependencies.utils import (
  19. _get_flat_fields_from_params,
  20. get_flat_dependant,
  21. get_flat_params,
  22. get_validation_alias,
  23. )
  24. from fastapi.encoders import jsonable_encoder
  25. from fastapi.exceptions import FastAPIDeprecationWarning
  26. from fastapi.openapi.constants import METHODS_WITH_BODY, REF_PREFIX
  27. from fastapi.openapi.models import OpenAPI
  28. from fastapi.params import Body, ParamTypes
  29. from fastapi.responses import Response
  30. from fastapi.sse import _SSE_EVENT_SCHEMA
  31. from fastapi.types import ModelNameMap
  32. from fastapi.utils import (
  33. deep_dict_update,
  34. generate_operation_id_for_path,
  35. is_body_allowed_for_status_code,
  36. )
  37. from pydantic import BaseModel
  38. from starlette.responses import JSONResponse
  39. from starlette.routing import BaseRoute
  40. validation_error_definition = {
  41. "title": "ValidationError",
  42. "type": "object",
  43. "properties": {
  44. "loc": {
  45. "title": "Location",
  46. "type": "array",
  47. "items": {"anyOf": [{"type": "string"}, {"type": "integer"}]},
  48. },
  49. "msg": {"title": "Message", "type": "string"},
  50. "type": {"title": "Error Type", "type": "string"},
  51. "input": {"title": "Input"},
  52. "ctx": {"title": "Context", "type": "object"},
  53. },
  54. "required": ["loc", "msg", "type"],
  55. }
  56. validation_error_response_definition = {
  57. "title": "HTTPValidationError",
  58. "type": "object",
  59. "properties": {
  60. "detail": {
  61. "title": "Detail",
  62. "type": "array",
  63. "items": {"$ref": REF_PREFIX + "ValidationError"},
  64. }
  65. },
  66. }
  67. status_code_ranges: dict[str, str] = {
  68. "1XX": "Information",
  69. "2XX": "Success",
  70. "3XX": "Redirection",
  71. "4XX": "Client Error",
  72. "5XX": "Server Error",
  73. "DEFAULT": "Default Response",
  74. }
  75. def get_openapi_security_definitions(
  76. flat_dependant: Dependant,
  77. ) -> tuple[dict[str, Any], list[dict[str, Any]]]:
  78. security_definitions = {}
  79. # Use a dict to merge scopes for same security scheme
  80. operation_security_dict: dict[str, list[str]] = {}
  81. for security_dependency in flat_dependant._security_dependencies:
  82. security_definition = jsonable_encoder(
  83. security_dependency._security_scheme.model,
  84. by_alias=True,
  85. exclude_none=True,
  86. )
  87. security_name = security_dependency._security_scheme.scheme_name
  88. security_definitions[security_name] = security_definition
  89. # Merge scopes for the same security scheme
  90. if security_name not in operation_security_dict:
  91. operation_security_dict[security_name] = []
  92. for scope in security_dependency.oauth_scopes or []:
  93. if scope not in operation_security_dict[security_name]:
  94. operation_security_dict[security_name].append(scope)
  95. operation_security = [
  96. {name: scopes} for name, scopes in operation_security_dict.items()
  97. ]
  98. return security_definitions, operation_security
  99. def _get_openapi_operation_parameters(
  100. *,
  101. dependant: Dependant,
  102. model_name_map: ModelNameMap,
  103. field_mapping: dict[
  104. tuple[ModelField, Literal["validation", "serialization"]], dict[str, Any]
  105. ],
  106. separate_input_output_schemas: bool = True,
  107. ) -> list[dict[str, Any]]:
  108. parameters = []
  109. flat_dependant = get_flat_dependant(dependant, skip_repeats=True)
  110. path_params = _get_flat_fields_from_params(flat_dependant.path_params)
  111. query_params = _get_flat_fields_from_params(flat_dependant.query_params)
  112. header_params = _get_flat_fields_from_params(flat_dependant.header_params)
  113. cookie_params = _get_flat_fields_from_params(flat_dependant.cookie_params)
  114. parameter_groups = [
  115. (ParamTypes.path, path_params),
  116. (ParamTypes.query, query_params),
  117. (ParamTypes.header, header_params),
  118. (ParamTypes.cookie, cookie_params),
  119. ]
  120. default_convert_underscores = True
  121. if len(flat_dependant.header_params) == 1:
  122. first_field = flat_dependant.header_params[0]
  123. if lenient_issubclass(first_field.field_info.annotation, BaseModel):
  124. default_convert_underscores = getattr(
  125. first_field.field_info, "convert_underscores", True
  126. )
  127. for param_type, param_group in parameter_groups:
  128. for param in param_group:
  129. field_info = param.field_info
  130. # field_info = cast(Param, field_info)
  131. if not getattr(field_info, "include_in_schema", True):
  132. continue
  133. param_schema = get_schema_from_model_field(
  134. field=param,
  135. model_name_map=model_name_map,
  136. field_mapping=field_mapping,
  137. separate_input_output_schemas=separate_input_output_schemas,
  138. )
  139. name = get_validation_alias(param)
  140. convert_underscores = getattr(
  141. param.field_info,
  142. "convert_underscores",
  143. default_convert_underscores,
  144. )
  145. if (
  146. param_type == ParamTypes.header
  147. and name == param.name
  148. and convert_underscores
  149. ):
  150. name = param.name.replace("_", "-")
  151. parameter = {
  152. "name": name,
  153. "in": param_type.value,
  154. "required": param.field_info.is_required(),
  155. "schema": param_schema,
  156. }
  157. if field_info.description:
  158. parameter["description"] = field_info.description
  159. openapi_examples = getattr(field_info, "openapi_examples", None)
  160. example = getattr(field_info, "example", None)
  161. if openapi_examples:
  162. parameter["examples"] = jsonable_encoder(openapi_examples)
  163. elif example is not _Unset:
  164. parameter["example"] = jsonable_encoder(example)
  165. if getattr(field_info, "deprecated", None):
  166. parameter["deprecated"] = True
  167. parameters.append(parameter)
  168. return parameters
  169. def get_openapi_operation_request_body(
  170. *,
  171. body_field: ModelField | None,
  172. model_name_map: ModelNameMap,
  173. field_mapping: dict[
  174. tuple[ModelField, Literal["validation", "serialization"]], dict[str, Any]
  175. ],
  176. separate_input_output_schemas: bool = True,
  177. ) -> dict[str, Any] | None:
  178. if not body_field:
  179. return None
  180. assert isinstance(body_field, ModelField)
  181. body_schema = get_schema_from_model_field(
  182. field=body_field,
  183. model_name_map=model_name_map,
  184. field_mapping=field_mapping,
  185. separate_input_output_schemas=separate_input_output_schemas,
  186. )
  187. field_info = cast(Body, body_field.field_info)
  188. request_media_type = field_info.media_type
  189. required = body_field.field_info.is_required()
  190. request_body_oai: dict[str, Any] = {}
  191. if required:
  192. request_body_oai["required"] = required
  193. request_media_content: dict[str, Any] = {"schema": body_schema}
  194. if field_info.openapi_examples:
  195. request_media_content["examples"] = jsonable_encoder(
  196. field_info.openapi_examples
  197. )
  198. elif field_info.example is not _Unset:
  199. request_media_content["example"] = jsonable_encoder(field_info.example)
  200. request_body_oai["content"] = {request_media_type: request_media_content}
  201. return request_body_oai
  202. def generate_operation_id(
  203. *, route: routing.APIRoute, method: str
  204. ) -> str: # pragma: nocover
  205. warnings.warn(
  206. message="fastapi.openapi.utils.generate_operation_id() was deprecated, "
  207. "it is not used internally, and will be removed soon",
  208. category=FastAPIDeprecationWarning,
  209. stacklevel=2,
  210. )
  211. if route.operation_id:
  212. return route.operation_id
  213. path: str = route.path_format
  214. return generate_operation_id_for_path(name=route.name, path=path, method=method)
  215. def generate_operation_summary(*, route: routing.APIRoute, method: str) -> str:
  216. if route.summary:
  217. return route.summary
  218. return route.name.replace("_", " ").title()
  219. def get_openapi_operation_metadata(
  220. *, route: routing.APIRoute, method: str, operation_ids: set[str]
  221. ) -> dict[str, Any]:
  222. operation: dict[str, Any] = {}
  223. if route.tags:
  224. operation["tags"] = route.tags
  225. operation["summary"] = generate_operation_summary(route=route, method=method)
  226. if route.description:
  227. operation["description"] = route.description
  228. operation_id = route.operation_id or route.unique_id
  229. if operation_id in operation_ids:
  230. endpoint_name = getattr(route.endpoint, "__name__", "<unnamed_endpoint>")
  231. message = f"Duplicate Operation ID {operation_id} for function {endpoint_name}"
  232. file_name = getattr(route.endpoint, "__globals__", {}).get("__file__")
  233. if file_name:
  234. message += f" at {file_name}"
  235. warnings.warn(message, stacklevel=1)
  236. operation_ids.add(operation_id)
  237. operation["operationId"] = operation_id
  238. if route.deprecated:
  239. operation["deprecated"] = route.deprecated
  240. return operation
  241. def get_openapi_path(
  242. *,
  243. route: routing.APIRoute,
  244. operation_ids: set[str],
  245. model_name_map: ModelNameMap,
  246. field_mapping: dict[
  247. tuple[ModelField, Literal["validation", "serialization"]], dict[str, Any]
  248. ],
  249. separate_input_output_schemas: bool = True,
  250. ) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]:
  251. path = {}
  252. security_schemes: dict[str, Any] = {}
  253. definitions: dict[str, Any] = {}
  254. assert route.methods is not None, "Methods must be a list"
  255. if isinstance(route.response_class, DefaultPlaceholder):
  256. current_response_class: type[Response] = route.response_class.value
  257. else:
  258. current_response_class = route.response_class
  259. assert current_response_class, "A response class is needed to generate OpenAPI"
  260. route_response_media_type: str | None = current_response_class.media_type
  261. if route.include_in_schema:
  262. for method in route.methods:
  263. operation = get_openapi_operation_metadata(
  264. route=route, method=method, operation_ids=operation_ids
  265. )
  266. parameters: list[dict[str, Any]] = []
  267. flat_dependant = get_flat_dependant(route.dependant, skip_repeats=True)
  268. security_definitions, operation_security = get_openapi_security_definitions(
  269. flat_dependant=flat_dependant
  270. )
  271. if operation_security:
  272. operation.setdefault("security", []).extend(operation_security)
  273. if security_definitions:
  274. security_schemes.update(security_definitions)
  275. operation_parameters = _get_openapi_operation_parameters(
  276. dependant=route.dependant,
  277. model_name_map=model_name_map,
  278. field_mapping=field_mapping,
  279. separate_input_output_schemas=separate_input_output_schemas,
  280. )
  281. parameters.extend(operation_parameters)
  282. if parameters:
  283. all_parameters = {
  284. (param["in"], param["name"]): param for param in parameters
  285. }
  286. required_parameters = {
  287. (param["in"], param["name"]): param
  288. for param in parameters
  289. if param.get("required")
  290. }
  291. # Make sure required definitions of the same parameter take precedence
  292. # over non-required definitions
  293. all_parameters.update(required_parameters)
  294. operation["parameters"] = list(all_parameters.values())
  295. if method in METHODS_WITH_BODY:
  296. request_body_oai = get_openapi_operation_request_body(
  297. body_field=route.body_field,
  298. model_name_map=model_name_map,
  299. field_mapping=field_mapping,
  300. separate_input_output_schemas=separate_input_output_schemas,
  301. )
  302. if request_body_oai:
  303. operation["requestBody"] = request_body_oai
  304. if route.callbacks:
  305. callbacks = {}
  306. for callback in route.callbacks:
  307. if isinstance(callback, routing.APIRoute):
  308. (
  309. cb_path,
  310. cb_security_schemes,
  311. cb_definitions,
  312. ) = get_openapi_path(
  313. route=callback,
  314. operation_ids=operation_ids,
  315. model_name_map=model_name_map,
  316. field_mapping=field_mapping,
  317. separate_input_output_schemas=separate_input_output_schemas,
  318. )
  319. callbacks[callback.name] = {callback.path: cb_path}
  320. operation["callbacks"] = callbacks
  321. if route.status_code is not None:
  322. status_code = str(route.status_code)
  323. else:
  324. # It would probably make more sense for all response classes to have an
  325. # explicit default status_code, and to extract it from them, instead of
  326. # doing this inspection tricks, that would probably be in the future
  327. # TODO: probably make status_code a default class attribute for all
  328. # responses in Starlette
  329. response_signature = inspect.signature(current_response_class.__init__)
  330. status_code_param = response_signature.parameters.get("status_code")
  331. if status_code_param is not None:
  332. if isinstance(status_code_param.default, int):
  333. status_code = str(status_code_param.default)
  334. operation.setdefault("responses", {}).setdefault(status_code, {})[
  335. "description"
  336. ] = route.response_description
  337. if is_body_allowed_for_status_code(route.status_code):
  338. # Check for JSONL streaming (generator endpoints)
  339. if route.is_json_stream:
  340. jsonl_content: dict[str, Any] = {}
  341. if route.stream_item_field:
  342. item_schema = get_schema_from_model_field(
  343. field=route.stream_item_field,
  344. model_name_map=model_name_map,
  345. field_mapping=field_mapping,
  346. separate_input_output_schemas=separate_input_output_schemas,
  347. )
  348. jsonl_content["itemSchema"] = item_schema
  349. else:
  350. jsonl_content["itemSchema"] = {}
  351. operation.setdefault("responses", {}).setdefault(
  352. status_code, {}
  353. ).setdefault("content", {})["application/jsonl"] = jsonl_content
  354. elif route.is_sse_stream:
  355. sse_content: dict[str, Any] = {}
  356. item_schema = copy.deepcopy(_SSE_EVENT_SCHEMA)
  357. if route.stream_item_field:
  358. content_schema = get_schema_from_model_field(
  359. field=route.stream_item_field,
  360. model_name_map=model_name_map,
  361. field_mapping=field_mapping,
  362. separate_input_output_schemas=separate_input_output_schemas,
  363. )
  364. item_schema["required"] = ["data"]
  365. item_schema["properties"]["data"] = {
  366. "type": "string",
  367. "contentMediaType": "application/json",
  368. "contentSchema": content_schema,
  369. }
  370. sse_content["itemSchema"] = item_schema
  371. operation.setdefault("responses", {}).setdefault(
  372. status_code, {}
  373. ).setdefault("content", {})["text/event-stream"] = sse_content
  374. elif route_response_media_type:
  375. response_schema = {"type": "string"}
  376. if lenient_issubclass(current_response_class, JSONResponse):
  377. if route.response_field:
  378. response_schema = get_schema_from_model_field(
  379. field=route.response_field,
  380. model_name_map=model_name_map,
  381. field_mapping=field_mapping,
  382. separate_input_output_schemas=separate_input_output_schemas,
  383. )
  384. else:
  385. response_schema = {}
  386. operation.setdefault("responses", {}).setdefault(
  387. status_code, {}
  388. ).setdefault("content", {}).setdefault(
  389. route_response_media_type, {}
  390. )["schema"] = response_schema
  391. if route.responses:
  392. operation_responses = operation.setdefault("responses", {})
  393. for (
  394. additional_status_code,
  395. additional_response,
  396. ) in route.responses.items():
  397. process_response = copy.deepcopy(additional_response)
  398. process_response.pop("model", None)
  399. status_code_key = str(additional_status_code).upper()
  400. if status_code_key == "DEFAULT":
  401. status_code_key = "default"
  402. openapi_response = operation_responses.setdefault(
  403. status_code_key, {}
  404. )
  405. assert isinstance(process_response, dict), (
  406. "An additional response must be a dict"
  407. )
  408. field = route.response_fields.get(additional_status_code)
  409. additional_field_schema: dict[str, Any] | None = None
  410. if field:
  411. additional_field_schema = get_schema_from_model_field(
  412. field=field,
  413. model_name_map=model_name_map,
  414. field_mapping=field_mapping,
  415. separate_input_output_schemas=separate_input_output_schemas,
  416. )
  417. media_type = route_response_media_type or "application/json"
  418. additional_schema = (
  419. process_response.setdefault("content", {})
  420. .setdefault(media_type, {})
  421. .setdefault("schema", {})
  422. )
  423. deep_dict_update(additional_schema, additional_field_schema)
  424. status_text: str | None = status_code_ranges.get(
  425. str(additional_status_code).upper()
  426. ) or http.client.responses.get(int(additional_status_code))
  427. description = (
  428. process_response.get("description")
  429. or openapi_response.get("description")
  430. or status_text
  431. or "Additional Response"
  432. )
  433. deep_dict_update(openapi_response, process_response)
  434. openapi_response["description"] = description
  435. http422 = "422"
  436. all_route_params = get_flat_params(route.dependant)
  437. if (all_route_params or route.body_field) and not any(
  438. status in operation["responses"]
  439. for status in [http422, "4XX", "default"]
  440. ):
  441. operation["responses"][http422] = {
  442. "description": "Validation Error",
  443. "content": {
  444. "application/json": {
  445. "schema": {"$ref": REF_PREFIX + "HTTPValidationError"}
  446. }
  447. },
  448. }
  449. if "ValidationError" not in definitions:
  450. definitions.update(
  451. {
  452. "ValidationError": validation_error_definition,
  453. "HTTPValidationError": validation_error_response_definition,
  454. }
  455. )
  456. if route.openapi_extra:
  457. deep_dict_update(operation, route.openapi_extra)
  458. path[method.lower()] = operation
  459. return path, security_schemes, definitions
  460. def get_fields_from_routes(
  461. routes: Sequence[BaseRoute],
  462. ) -> list[ModelField]:
  463. body_fields_from_routes: list[ModelField] = []
  464. responses_from_routes: list[ModelField] = []
  465. request_fields_from_routes: list[ModelField] = []
  466. callback_flat_models: list[ModelField] = []
  467. for route in routes:
  468. if not isinstance(route, routing.APIRoute):
  469. continue
  470. if route.include_in_schema:
  471. if route.body_field:
  472. assert isinstance(route.body_field, ModelField), (
  473. "A request body must be a Pydantic Field"
  474. )
  475. body_fields_from_routes.append(route.body_field)
  476. if route.response_field:
  477. responses_from_routes.append(route.response_field)
  478. if route.response_fields:
  479. responses_from_routes.extend(route.response_fields.values())
  480. if route.stream_item_field:
  481. responses_from_routes.append(route.stream_item_field)
  482. if route.callbacks:
  483. callback_flat_models.extend(get_fields_from_routes(route.callbacks))
  484. params = get_flat_params(route.dependant)
  485. request_fields_from_routes.extend(params)
  486. flat_models = callback_flat_models + list(
  487. body_fields_from_routes + responses_from_routes + request_fields_from_routes
  488. )
  489. return flat_models
  490. def get_openapi(
  491. *,
  492. title: str,
  493. version: str,
  494. openapi_version: str = "3.1.0",
  495. summary: str | None = None,
  496. description: str | None = None,
  497. routes: Sequence[BaseRoute],
  498. webhooks: Sequence[BaseRoute] | None = None,
  499. tags: list[dict[str, Any]] | None = None,
  500. servers: list[dict[str, str | Any]] | None = None,
  501. terms_of_service: str | None = None,
  502. contact: dict[str, str | Any] | None = None,
  503. license_info: dict[str, str | Any] | None = None,
  504. separate_input_output_schemas: bool = True,
  505. external_docs: dict[str, Any] | None = None,
  506. ) -> dict[str, Any]:
  507. info: dict[str, Any] = {"title": title, "version": version}
  508. if summary:
  509. info["summary"] = summary
  510. if description:
  511. info["description"] = description
  512. if terms_of_service:
  513. info["termsOfService"] = terms_of_service
  514. if contact:
  515. info["contact"] = contact
  516. if license_info:
  517. info["license"] = license_info
  518. output: dict[str, Any] = {"openapi": openapi_version, "info": info}
  519. if servers:
  520. output["servers"] = servers
  521. components: dict[str, dict[str, Any]] = {}
  522. paths: dict[str, dict[str, Any]] = {}
  523. webhook_paths: dict[str, dict[str, Any]] = {}
  524. operation_ids: set[str] = set()
  525. all_fields = get_fields_from_routes(list(routes or []) + list(webhooks or []))
  526. flat_models = get_flat_models_from_fields(all_fields, known_models=set())
  527. model_name_map = get_model_name_map(flat_models)
  528. field_mapping, definitions = get_definitions(
  529. fields=all_fields,
  530. model_name_map=model_name_map,
  531. separate_input_output_schemas=separate_input_output_schemas,
  532. )
  533. for route in routes or []:
  534. if isinstance(route, routing.APIRoute):
  535. result = get_openapi_path(
  536. route=route,
  537. operation_ids=operation_ids,
  538. model_name_map=model_name_map,
  539. field_mapping=field_mapping,
  540. separate_input_output_schemas=separate_input_output_schemas,
  541. )
  542. if result:
  543. path, security_schemes, path_definitions = result
  544. if path:
  545. paths.setdefault(route.path_format, {}).update(path)
  546. if security_schemes:
  547. components.setdefault("securitySchemes", {}).update(
  548. security_schemes
  549. )
  550. if path_definitions:
  551. definitions.update(path_definitions)
  552. for webhook in webhooks or []:
  553. if isinstance(webhook, routing.APIRoute):
  554. result = get_openapi_path(
  555. route=webhook,
  556. operation_ids=operation_ids,
  557. model_name_map=model_name_map,
  558. field_mapping=field_mapping,
  559. separate_input_output_schemas=separate_input_output_schemas,
  560. )
  561. if result:
  562. path, security_schemes, path_definitions = result
  563. if path:
  564. webhook_paths.setdefault(webhook.path_format, {}).update(path)
  565. if security_schemes:
  566. components.setdefault("securitySchemes", {}).update(
  567. security_schemes
  568. )
  569. if path_definitions:
  570. definitions.update(path_definitions)
  571. if definitions:
  572. components["schemas"] = {k: definitions[k] for k in sorted(definitions)}
  573. if components:
  574. output["components"] = components
  575. output["paths"] = paths
  576. if webhook_paths:
  577. output["webhooks"] = webhook_paths
  578. if tags:
  579. output["tags"] = tags
  580. if external_docs:
  581. output["externalDocs"] = external_docs
  582. return jsonable_encoder(OpenAPI(**output), by_alias=True, exclude_none=True) # type: ignore # ty: ignore[unused-ignore-comment]