PKMQ#  fastapi/__init__.py"""FastAPI framework, high performance, easy to learn, fast to code, ready for production""" __version__ = "0.1.6" from .applications import FastAPI from .routing import APIRouter from .params import Body, Path, Query, Header, Cookie, Form, File, Security, Depends PK]MH"55fastapi/applications.pyfrom typing import Any, Callable, Dict, List, Optional, Type from fastapi import routing from fastapi.openapi.docs import get_redoc_html, get_swagger_ui_html from fastapi.openapi.utils import get_openapi from pydantic import BaseModel from starlette.applications import Starlette from starlette.exceptions import ExceptionMiddleware, HTTPException from starlette.middleware.errors import ServerErrorMiddleware from starlette.middleware.lifespan import LifespanMiddleware from starlette.requests import Request from starlette.responses import JSONResponse, Response async def http_exception(request: Request, exc: HTTPException) -> JSONResponse: return JSONResponse({"detail": exc.detail}, status_code=exc.status_code) class FastAPI(Starlette): def __init__( self, debug: bool = False, template_directory: str = None, title: str = "Fast API", description: str = "", version: str = "0.1.0", openapi_url: str = "/openapi.json", swagger_ui_url: str = "/docs", redoc_url: str = "/redoc", **extra: Dict[str, Any], ) -> None: self._debug = debug self.router: routing.APIRouter = routing.APIRouter() self.exception_middleware = ExceptionMiddleware(self.router, debug=debug) self.error_middleware = ServerErrorMiddleware( self.exception_middleware, debug=debug ) self.lifespan_middleware = LifespanMiddleware(self.error_middleware) self.schema_generator = None self.template_env = self.load_template_env(template_directory) self.title = title self.description = description self.version = version self.openapi_url = openapi_url self.swagger_ui_url = swagger_ui_url self.redoc_url = redoc_url self.extra = extra self.openapi_version = "3.0.2" if self.openapi_url: assert self.title, "A title must be provided for OpenAPI, e.g.: 'My API'" assert self.version, "A version must be provided for OpenAPI, e.g.: '2.1.0'" if self.swagger_ui_url or self.redoc_url: assert self.openapi_url, "The openapi_url is required for the docs" self.openapi_schema: Optional[Dict[str, Any]] = None self.setup() def openapi(self) -> Dict: if not self.openapi_schema: self.openapi_schema = get_openapi( title=self.title, version=self.version, openapi_version=self.openapi_version, description=self.description, routes=self.routes, ) return self.openapi_schema def setup(self) -> None: if self.openapi_url: self.add_route( self.openapi_url, lambda req: JSONResponse(self.openapi()), include_in_schema=False, ) if self.swagger_ui_url: self.add_route( self.swagger_ui_url, lambda r: get_swagger_ui_html( openapi_url=self.openapi_url, title=self.title + " - Swagger UI" ), include_in_schema=False, ) if self.redoc_url: self.add_route( self.redoc_url, lambda r: get_redoc_html( openapi_url=self.openapi_url, title=self.title + " - ReDoc" ), include_in_schema=False, ) self.add_exception_handler(HTTPException, http_exception) def add_api_route( self, path: str, endpoint: Callable, *, response_model: Type[BaseModel] = None, status_code: int = 200, tags: List[str] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", deprecated: bool = None, name: str = None, methods: List[str] = None, operation_id: str = None, include_in_schema: bool = True, content_type: Type[Response] = JSONResponse, ) -> None: self.router.add_api_route( path, endpoint=endpoint, response_model=response_model, status_code=status_code, tags=tags or [], summary=summary, description=description, response_description=response_description, deprecated=deprecated, name=name, methods=methods, operation_id=operation_id, include_in_schema=include_in_schema, content_type=content_type, ) def api_route( self, path: str, *, response_model: Type[BaseModel] = None, status_code: int = 200, tags: List[str] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", deprecated: bool = None, name: str = None, methods: List[str] = None, operation_id: str = None, include_in_schema: bool = True, content_type: Type[Response] = JSONResponse, ) -> Callable: def decorator(func: Callable) -> Callable: self.router.add_api_route( path, func, response_model=response_model, status_code=status_code, tags=tags or [], summary=summary, description=description, response_description=response_description, deprecated=deprecated, name=name, methods=methods, operation_id=operation_id, include_in_schema=include_in_schema, content_type=content_type, ) return func return decorator def include_router(self, router: routing.APIRouter, *, prefix: str = "") -> None: self.router.include_router(router, prefix=prefix) def get( self, path: str, *, response_model: Type[BaseModel] = None, status_code: int = 200, tags: List[str] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", deprecated: bool = None, name: str = None, operation_id: str = None, include_in_schema: bool = True, content_type: Type[Response] = JSONResponse, ) -> Callable: return self.router.get( path, response_model=response_model, status_code=status_code, tags=tags or [], summary=summary, description=description, response_description=response_description, deprecated=deprecated, name=name, operation_id=operation_id, include_in_schema=include_in_schema, content_type=content_type, ) def put( self, path: str, *, response_model: Type[BaseModel] = None, status_code: int = 200, tags: List[str] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", deprecated: bool = None, name: str = None, operation_id: str = None, include_in_schema: bool = True, content_type: Type[Response] = JSONResponse, ) -> Callable: return self.router.put( path, response_model=response_model, status_code=status_code, tags=tags or [], summary=summary, description=description, response_description=response_description, deprecated=deprecated, name=name, operation_id=operation_id, include_in_schema=include_in_schema, content_type=content_type, ) def post( self, path: str, *, response_model: Type[BaseModel] = None, status_code: int = 200, tags: List[str] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", deprecated: bool = None, name: str = None, operation_id: str = None, include_in_schema: bool = True, content_type: Type[Response] = JSONResponse, ) -> Callable: return self.router.post( path, response_model=response_model, status_code=status_code, tags=tags or [], summary=summary, description=description, response_description=response_description, deprecated=deprecated, name=name, operation_id=operation_id, include_in_schema=include_in_schema, content_type=content_type, ) def delete( self, path: str, *, response_model: Type[BaseModel] = None, status_code: int = 200, tags: List[str] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", deprecated: bool = None, name: str = None, operation_id: str = None, include_in_schema: bool = True, content_type: Type[Response] = JSONResponse, ) -> Callable: return self.router.delete( path, response_model=response_model, status_code=status_code, tags=tags or [], summary=summary, description=description, response_description=response_description, deprecated=deprecated, name=name, operation_id=operation_id, include_in_schema=include_in_schema, content_type=content_type, ) def options( self, path: str, *, response_model: Type[BaseModel] = None, status_code: int = 200, tags: List[str] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", deprecated: bool = None, name: str = None, operation_id: str = None, include_in_schema: bool = True, content_type: Type[Response] = JSONResponse, ) -> Callable: return self.router.options( path, response_model=response_model, status_code=status_code, tags=tags or [], summary=summary, description=description, response_description=response_description, deprecated=deprecated, name=name, operation_id=operation_id, include_in_schema=include_in_schema, content_type=content_type, ) def head( self, path: str, *, response_model: Type[BaseModel] = None, status_code: int = 200, tags: List[str] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", deprecated: bool = None, name: str = None, operation_id: str = None, include_in_schema: bool = True, content_type: Type[Response] = JSONResponse, ) -> Callable: return self.router.head( path, response_model=response_model, status_code=status_code, tags=tags or [], summary=summary, description=description, response_description=response_description, deprecated=deprecated, name=name, operation_id=operation_id, include_in_schema=include_in_schema, content_type=content_type, ) def patch( self, path: str, *, response_model: Type[BaseModel] = None, status_code: int = 200, tags: List[str] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", deprecated: bool = None, name: str = None, operation_id: str = None, include_in_schema: bool = True, content_type: Type[Response] = JSONResponse, ) -> Callable: return self.router.patch( path, response_model=response_model, status_code=status_code, tags=tags or [], summary=summary, description=description, response_description=response_description, deprecated=deprecated, name=name, operation_id=operation_id, include_in_schema=include_in_schema, content_type=content_type, ) def trace( self, path: str, *, response_model: Type[BaseModel] = None, status_code: int = 200, tags: List[str] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", deprecated: bool = None, name: str = None, operation_id: str = None, include_in_schema: bool = True, content_type: Type[Response] = JSONResponse, ) -> Callable: return self.router.trace( path, response_model=response_model, status_code=status_code, tags=tags or [], summary=summary, description=description, response_description=response_description, deprecated=deprecated, name=name, operation_id=operation_id, include_in_schema=include_in_schema, content_type=content_type, ) PKM4 ??fastapi/encoders.pyfrom enum import Enum from types import GeneratorType from typing import Any, Set from pydantic import BaseModel from pydantic.json import pydantic_encoder def jsonable_encoder( obj: Any, include: Set[str] = None, exclude: Set[str] = set(), by_alias: bool = False, include_none: bool = True, ) -> Any: if isinstance(obj, BaseModel): return jsonable_encoder( obj.dict(include=include, exclude=exclude, by_alias=by_alias), include_none=include_none, ) if isinstance(obj, Enum): return obj.value if isinstance(obj, (str, int, float, type(None))): return obj if isinstance(obj, dict): return { jsonable_encoder( key, by_alias=by_alias, include_none=include_none ): jsonable_encoder(value, by_alias=by_alias, include_none=include_none) for key, value in obj.items() if value is not None or include_none } if isinstance(obj, (list, set, frozenset, GeneratorType, tuple)): return [ jsonable_encoder( item, include=include, exclude=exclude, by_alias=by_alias, include_none=include_none, ) for item in obj ] return pydantic_encoder(obj) PKM+͖fastapi/params.pyfrom enum import Enum from typing import Any, Callable, Sequence from pydantic import Schema class ParamTypes(Enum): query = "query" header = "header" path = "path" cookie = "cookie" class Param(Schema): in_: ParamTypes def __init__( self, default: Any, *, deprecated: bool = None, alias: str = None, title: str = None, description: str = None, gt: float = None, ge: float = None, lt: float = None, le: float = None, min_length: int = None, max_length: int = None, regex: str = None, **extra: Any, ): self.deprecated = deprecated super().__init__( default, alias=alias, title=title, description=description, gt=gt, ge=ge, lt=lt, le=le, min_length=min_length, max_length=max_length, regex=regex, **extra, ) class Path(Param): in_ = ParamTypes.path def __init__( self, default: Any, *, deprecated: bool = None, alias: str = None, title: str = None, description: str = None, gt: float = None, ge: float = None, lt: float = None, le: float = None, min_length: int = None, max_length: int = None, regex: str = None, **extra: Any, ): self.description = description self.deprecated = deprecated self.in_ = self.in_ super().__init__( default, alias=alias, title=title, description=description, gt=gt, ge=ge, lt=lt, le=le, min_length=min_length, max_length=max_length, regex=regex, **extra, ) class Query(Param): in_ = ParamTypes.query def __init__( self, default: Any, *, deprecated: bool = None, alias: str = None, title: str = None, description: str = None, gt: float = None, ge: float = None, lt: float = None, le: float = None, min_length: int = None, max_length: int = None, regex: str = None, **extra: Any, ): self.description = description self.deprecated = deprecated super().__init__( default, alias=alias, title=title, description=description, gt=gt, ge=ge, lt=lt, le=le, min_length=min_length, max_length=max_length, regex=regex, **extra, ) class Header(Param): in_ = ParamTypes.header def __init__( self, default: Any, *, deprecated: bool = None, alias: str = None, convert_underscores: bool = True, title: str = None, description: str = None, gt: float = None, ge: float = None, lt: float = None, le: float = None, min_length: int = None, max_length: int = None, regex: str = None, **extra: Any, ): self.description = description self.deprecated = deprecated self.convert_underscores = convert_underscores super().__init__( default, alias=alias, title=title, description=description, gt=gt, ge=ge, lt=lt, le=le, min_length=min_length, max_length=max_length, regex=regex, **extra, ) class Cookie(Param): in_ = ParamTypes.cookie def __init__( self, default: Any, *, deprecated: bool = None, alias: str = None, title: str = None, description: str = None, gt: float = None, ge: float = None, lt: float = None, le: float = None, min_length: int = None, max_length: int = None, regex: str = None, **extra: Any, ): self.description = description self.deprecated = deprecated super().__init__( default, alias=alias, title=title, description=description, gt=gt, ge=ge, lt=lt, le=le, min_length=min_length, max_length=max_length, regex=regex, **extra, ) class Body(Schema): def __init__( self, default: Any, *, embed: bool = False, media_type: str = "application/json", alias: str = None, title: str = None, description: str = None, gt: float = None, ge: float = None, lt: float = None, le: float = None, min_length: int = None, max_length: int = None, regex: str = None, **extra: Any, ): self.embed = embed self.media_type = media_type super().__init__( default, alias=alias, title=title, description=description, gt=gt, ge=ge, lt=lt, le=le, min_length=min_length, max_length=max_length, regex=regex, **extra, ) class Form(Body): def __init__( self, default: Any, *, sub_key: bool = False, media_type: str = "application/x-www-form-urlencoded", alias: str = None, title: str = None, description: str = None, gt: float = None, ge: float = None, lt: float = None, le: float = None, min_length: int = None, max_length: int = None, regex: str = None, **extra: Any, ): super().__init__( default, embed=sub_key, media_type=media_type, alias=alias, title=title, description=description, gt=gt, ge=ge, lt=lt, le=le, min_length=min_length, max_length=max_length, regex=regex, **extra, ) class File(Form): def __init__( self, default: Any, *, sub_key: bool = False, media_type: str = "multipart/form-data", alias: str = None, title: str = None, description: str = None, gt: float = None, ge: float = None, lt: float = None, le: float = None, min_length: int = None, max_length: int = None, regex: str = None, **extra: Any, ): super().__init__( default, embed=sub_key, media_type=media_type, alias=alias, title=title, description=description, gt=gt, ge=ge, lt=lt, le=le, min_length=min_length, max_length=max_length, regex=regex, **extra, ) class Depends: def __init__(self, dependency: Callable = None): self.dependency = dependency class Security(Depends): def __init__(self, dependency: Callable = None, scopes: Sequence[str] = None): self.scopes = scopes or [] super().__init__(dependency=dependency) PKM oKKfastapi/routing.pyimport asyncio import inspect import logging from typing import Any, Callable, List, Optional, Type from fastapi import params from fastapi.dependencies.models import Dependant from fastapi.dependencies.utils import get_body_field, get_dependant, solve_dependencies from fastapi.encoders import jsonable_encoder from pydantic import BaseConfig, BaseModel, Schema from pydantic.error_wrappers import ErrorWrapper, ValidationError from pydantic.fields import Field from pydantic.utils import lenient_issubclass from starlette import routing from starlette.concurrency import run_in_threadpool from starlette.exceptions import HTTPException from starlette.formparsers import UploadFile from starlette.requests import Request from starlette.responses import JSONResponse, Response from starlette.routing import get_name, request_response from starlette.status import HTTP_422_UNPROCESSABLE_ENTITY def serialize_response(*, field: Field = None, response: Response) -> Any: if field: errors = [] value, errors_ = field.validate(response, {}, loc=("response",)) if isinstance(errors_, ErrorWrapper): errors.append(errors_) elif isinstance(errors_, list): errors.extend(errors_) if errors: raise ValidationError(errors) return jsonable_encoder(value) else: return jsonable_encoder(response) def get_app( dependant: Dependant, body_field: Field = None, status_code: int = 200, content_type: Type[Response] = JSONResponse, response_field: Field = None, ) -> Callable: assert dependant.call is not None, "dependant.call must me a function" is_coroutine = asyncio.iscoroutinefunction(dependant.call) is_body_form = body_field and isinstance(body_field.schema, params.Form) async def app(request: Request) -> Response: try: body = None if body_field: if is_body_form: raw_body = await request.form() body = {} for field, value in raw_body.items(): if isinstance(value, UploadFile): body[field] = await value.read() else: body[field] = value else: body = await request.json() except Exception as e: logging.error("Error getting request body", e) raise HTTPException( status_code=400, detail="There was an error parsing the body" ) try: values, errors = await solve_dependencies( request=request, dependant=dependant, body=body ) except Exception as e: logging.error("Error solving dependencies", e) raise HTTPException(status_code=400, detail="Error processing request") if errors: errors_out = ValidationError(errors) raise HTTPException( status_code=HTTP_422_UNPROCESSABLE_ENTITY, detail=errors_out.errors() ) else: assert dependant.call is not None, "dependant.call must me a function" if is_coroutine: raw_response = await dependant.call(**values) else: raw_response = await run_in_threadpool(dependant.call, **values) if isinstance(raw_response, Response): return raw_response if isinstance(raw_response, BaseModel): return content_type( content=serialize_response( field=response_field, response=raw_response ), status_code=status_code, ) errors = [] try: return content_type( content=serialize_response( field=response_field, response=raw_response ), status_code=status_code, ) except Exception as e: errors.append(e) try: response = dict(raw_response) return content_type( content=serialize_response(field=response_field, response=response), status_code=status_code, ) except Exception as e: errors.append(e) try: response = vars(raw_response) return content_type( content=serialize_response(field=response_field, response=response), status_code=status_code, ) except Exception as e: errors.append(e) raise ValueError(errors) return app class APIRoute(routing.Route): def __init__( self, path: str, endpoint: Callable, *, response_model: Type[BaseModel] = None, status_code: int = 200, tags: List[str] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", deprecated: bool = None, name: str = None, methods: List[str] = None, operation_id: str = None, include_in_schema: bool = True, content_type: Type[Response] = JSONResponse, ) -> None: assert path.startswith("/"), "Routed paths must always start with '/'" self.path = path self.endpoint = endpoint self.name = get_name(endpoint) if name is None else name self.response_model = response_model if self.response_model: assert lenient_issubclass( content_type, JSONResponse ), "To declare a type the response must be a JSON response" response_name = "Response_" + self.name self.response_field: Optional[Field] = Field( name=response_name, type_=self.response_model, class_validators=[], default=None, required=False, model_config=BaseConfig(), schema=Schema(None), ) else: self.response_field = None self.status_code = status_code self.tags = tags or [] self.summary = summary self.description = description or self.endpoint.__doc__ self.response_description = response_description self.deprecated = deprecated if methods is None: methods = ["GET"] self.methods = methods self.operation_id = operation_id self.include_in_schema = include_in_schema self.content_type = content_type self.path_regex, self.path_format, self.param_convertors = self.compile_path( path ) assert inspect.isfunction(endpoint) or inspect.ismethod( endpoint ), f"An endpoint must be a function or method" self.dependant = get_dependant(path=path, call=self.endpoint) self.body_field = get_body_field(dependant=self.dependant, name=self.name) self.app = request_response( get_app( dependant=self.dependant, body_field=self.body_field, status_code=self.status_code, content_type=self.content_type, response_field=self.response_field, ) ) class APIRouter(routing.Router): def add_api_route( self, path: str, endpoint: Callable, *, response_model: Type[BaseModel] = None, status_code: int = 200, tags: List[str] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", deprecated: bool = None, name: str = None, methods: List[str] = None, operation_id: str = None, include_in_schema: bool = True, content_type: Type[Response] = JSONResponse, ) -> None: route = APIRoute( path, endpoint=endpoint, response_model=response_model, status_code=status_code, tags=tags or [], summary=summary, description=description, response_description=response_description, deprecated=deprecated, name=name, methods=methods, operation_id=operation_id, include_in_schema=include_in_schema, content_type=content_type, ) self.routes.append(route) def api_route( self, path: str, *, response_model: Type[BaseModel] = None, status_code: int = 200, tags: List[str] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", deprecated: bool = None, name: str = None, methods: List[str] = None, operation_id: str = None, include_in_schema: bool = True, content_type: Type[Response] = JSONResponse, ) -> Callable: def decorator(func: Callable) -> Callable: self.add_api_route( path, func, response_model=response_model, status_code=status_code, tags=tags or [], summary=summary, description=description, response_description=response_description, deprecated=deprecated, name=name, methods=methods, operation_id=operation_id, include_in_schema=include_in_schema, content_type=content_type, ) return func return decorator def include_router(self, router: "APIRouter", *, prefix: str = "") -> None: if prefix: assert prefix.startswith("/"), "A path prefix must start with '/'" assert not prefix.endswith( "/" ), "A path prefix must not end with '/', as the routes will start with '/'" for route in router.routes: if isinstance(route, APIRoute): self.add_api_route( prefix + route.path, route.endpoint, response_model=route.response_model, status_code=route.status_code, tags=route.tags or [], summary=route.summary, description=route.description, response_description=route.response_description, deprecated=route.deprecated, name=route.name, methods=route.methods, operation_id=route.operation_id, include_in_schema=route.include_in_schema, content_type=route.content_type, ) elif isinstance(route, routing.Route): self.add_route( prefix + route.path, route.endpoint, methods=route.methods, name=route.name, include_in_schema=route.include_in_schema, ) def get( self, path: str, *, response_model: Type[BaseModel] = None, status_code: int = 200, tags: List[str] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", deprecated: bool = None, name: str = None, operation_id: str = None, include_in_schema: bool = True, content_type: Type[Response] = JSONResponse, ) -> Callable: return self.api_route( path=path, response_model=response_model, status_code=status_code, tags=tags or [], summary=summary, description=description, response_description=response_description, deprecated=deprecated, name=name, methods=["GET"], operation_id=operation_id, include_in_schema=include_in_schema, content_type=content_type, ) def put( self, path: str, *, response_model: Type[BaseModel] = None, status_code: int = 200, tags: List[str] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", deprecated: bool = None, name: str = None, operation_id: str = None, include_in_schema: bool = True, content_type: Type[Response] = JSONResponse, ) -> Callable: return self.api_route( path=path, response_model=response_model, status_code=status_code, tags=tags or [], summary=summary, description=description, response_description=response_description, deprecated=deprecated, name=name, methods=["PUT"], operation_id=operation_id, include_in_schema=include_in_schema, content_type=content_type, ) def post( self, path: str, *, response_model: Type[BaseModel] = None, status_code: int = 200, tags: List[str] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", deprecated: bool = None, name: str = None, operation_id: str = None, include_in_schema: bool = True, content_type: Type[Response] = JSONResponse, ) -> Callable: return self.api_route( path=path, response_model=response_model, status_code=status_code, tags=tags or [], summary=summary, description=description, response_description=response_description, deprecated=deprecated, name=name, methods=["POST"], operation_id=operation_id, include_in_schema=include_in_schema, content_type=content_type, ) def delete( self, path: str, *, response_model: Type[BaseModel] = None, status_code: int = 200, tags: List[str] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", deprecated: bool = None, name: str = None, operation_id: str = None, include_in_schema: bool = True, content_type: Type[Response] = JSONResponse, ) -> Callable: return self.api_route( path=path, response_model=response_model, status_code=status_code, tags=tags or [], summary=summary, description=description, response_description=response_description, deprecated=deprecated, name=name, methods=["DELETE"], operation_id=operation_id, include_in_schema=include_in_schema, content_type=content_type, ) def options( self, path: str, *, response_model: Type[BaseModel] = None, status_code: int = 200, tags: List[str] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", deprecated: bool = None, name: str = None, operation_id: str = None, include_in_schema: bool = True, content_type: Type[Response] = JSONResponse, ) -> Callable: return self.api_route( path=path, response_model=response_model, status_code=status_code, tags=tags or [], summary=summary, description=description, response_description=response_description, deprecated=deprecated, name=name, methods=["OPTIONS"], operation_id=operation_id, include_in_schema=include_in_schema, content_type=content_type, ) def head( self, path: str, *, response_model: Type[BaseModel] = None, status_code: int = 200, tags: List[str] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", deprecated: bool = None, name: str = None, operation_id: str = None, include_in_schema: bool = True, content_type: Type[Response] = JSONResponse, ) -> Callable: return self.api_route( path=path, response_model=response_model, status_code=status_code, tags=tags or [], summary=summary, description=description, response_description=response_description, deprecated=deprecated, name=name, methods=["HEAD"], operation_id=operation_id, include_in_schema=include_in_schema, content_type=content_type, ) def patch( self, path: str, *, response_model: Type[BaseModel] = None, status_code: int = 200, tags: List[str] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", deprecated: bool = None, name: str = None, operation_id: str = None, include_in_schema: bool = True, content_type: Type[Response] = JSONResponse, ) -> Callable: return self.api_route( path=path, response_model=response_model, status_code=status_code, tags=tags or [], summary=summary, description=description, response_description=response_description, deprecated=deprecated, name=name, methods=["PATCH"], operation_id=operation_id, include_in_schema=include_in_schema, content_type=content_type, ) def trace( self, path: str, *, response_model: Type[BaseModel] = None, status_code: int = 200, tags: List[str] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", deprecated: bool = None, name: str = None, operation_id: str = None, include_in_schema: bool = True, content_type: Type[Response] = JSONResponse, ) -> Callable: return self.api_route( path=path, response_model=response_model, status_code=status_code, tags=tags or [], summary=summary, description=description, response_description=response_description, deprecated=deprecated, name=name, methods=["TRACE"], operation_id=operation_id, include_in_schema=include_in_schema, content_type=content_type, ) PKM.fastapi/utils.pyimport re from typing import Any, Dict, List, Sequence, Set, Type from fastapi import routing from fastapi.openapi.constants import REF_PREFIX from pydantic import BaseModel from pydantic.fields import Field from pydantic.schema import get_flat_models_from_fields, model_process_schema from starlette.routing import BaseRoute def get_flat_models_from_routes( routes: Sequence[Type[BaseRoute]] ) -> Set[Type[BaseModel]]: body_fields_from_routes: List[Field] = [] responses_from_routes: List[Field] = [] for route in routes: if getattr(route, "include_in_schema", None) and isinstance( route, routing.APIRoute ): if route.body_field: assert isinstance( route.body_field, Field ), "A request body must be a Pydantic Field" body_fields_from_routes.append(route.body_field) if route.response_field: responses_from_routes.append(route.response_field) flat_models = get_flat_models_from_fields( body_fields_from_routes + responses_from_routes ) return flat_models def get_model_definitions( *, flat_models: Set[Type[BaseModel]], model_name_map: Dict[Type[BaseModel], str] ) -> Dict[str, Any]: definitions: Dict[str, Dict] = {} for model in flat_models: m_schema, m_definitions = model_process_schema( model, model_name_map=model_name_map, ref_prefix=REF_PREFIX ) definitions.update(m_definitions) model_name = model_name_map[model] definitions[model_name] = m_schema return definitions def get_path_param_names(path: str) -> Set[str]: return {item.strip("{}") for item in re.findall("{[^}]*}", path)} PKρM fastapi/dependencies/__init__.pyPKM.pQFFfastapi/dependencies/models.pyfrom typing import Any, Callable, Dict, List, Sequence, Tuple from fastapi.security.base import SecurityBase from pydantic import BaseConfig, Schema from pydantic.error_wrappers import ErrorWrapper from pydantic.errors import MissingError from pydantic.fields import Field, Required from pydantic.schema import get_annotation_from_schema from starlette.concurrency import run_in_threadpool from starlette.requests import Request param_supported_types = (str, int, float, bool) class SecurityRequirement: def __init__(self, security_scheme: SecurityBase, scopes: Sequence[str] = None): self.security_scheme = security_scheme self.scopes = scopes class Dependant: def __init__( self, *, path_params: List[Field] = None, query_params: List[Field] = None, header_params: List[Field] = None, cookie_params: List[Field] = None, body_params: List[Field] = None, dependencies: List["Dependant"] = None, security_schemes: List[SecurityRequirement] = None, name: str = None, call: Callable = None, request_param_name: str = None, ) -> None: self.path_params = path_params or [] self.query_params = query_params or [] self.header_params = header_params or [] self.cookie_params = cookie_params or [] self.body_params = body_params or [] self.dependencies = dependencies or [] self.security_requirements = security_schemes or [] self.request_param_name = request_param_name self.name = name self.call = call PKݐM9r22fastapi/dependencies/utils.pyimport asyncio import inspect from copy import deepcopy from typing import Any, Callable, Dict, List, Mapping, Sequence, Tuple, Type from fastapi import params from fastapi.dependencies.models import Dependant, SecurityRequirement from fastapi.security.base import SecurityBase from fastapi.utils import get_path_param_names from pydantic import BaseConfig, Schema, create_model from pydantic.error_wrappers import ErrorWrapper from pydantic.errors import MissingError from pydantic.fields import Field, Required from pydantic.schema import get_annotation_from_schema from pydantic.utils import lenient_issubclass from starlette.concurrency import run_in_threadpool from starlette.requests import Request param_supported_types = (str, int, float, bool) def get_sub_dependant(*, param: inspect.Parameter, path: str) -> Dependant: depends: params.Depends = param.default if depends.dependency: dependency = depends.dependency else: dependency = param.annotation sub_dependant = get_dependant(path=path, call=dependency, name=param.name) if isinstance(depends, params.Security) and isinstance(dependency, SecurityBase): security_requirement = SecurityRequirement( security_scheme=dependency, scopes=depends.scopes ) sub_dependant.security_requirements.append(security_requirement) return sub_dependant def get_flat_dependant(dependant: Dependant) -> Dependant: flat_dependant = Dependant( path_params=dependant.path_params.copy(), query_params=dependant.query_params.copy(), header_params=dependant.header_params.copy(), cookie_params=dependant.cookie_params.copy(), body_params=dependant.body_params.copy(), security_schemes=dependant.security_requirements.copy(), ) for sub_dependant in dependant.dependencies: if sub_dependant is dependant: raise ValueError("recursion", dependant.dependencies) flat_sub = get_flat_dependant(sub_dependant) flat_dependant.path_params.extend(flat_sub.path_params) flat_dependant.query_params.extend(flat_sub.query_params) flat_dependant.header_params.extend(flat_sub.header_params) flat_dependant.cookie_params.extend(flat_sub.cookie_params) flat_dependant.body_params.extend(flat_sub.body_params) flat_dependant.security_requirements.extend(flat_sub.security_requirements) return flat_dependant def get_dependant(*, path: str, call: Callable, name: str = None) -> Dependant: path_param_names = get_path_param_names(path) endpoint_signature = inspect.signature(call) signature_params = endpoint_signature.parameters dependant = Dependant(call=call, name=name) for param_name in signature_params: param = signature_params[param_name] if isinstance(param.default, params.Depends): sub_dependant = get_sub_dependant(param=param, path=path) dependant.dependencies.append(sub_dependant) for param_name in signature_params: param = signature_params[param_name] if ( (param.default == param.empty) or isinstance(param.default, params.Path) ) and (param_name in path_param_names): assert ( lenient_issubclass(param.annotation, param_supported_types) or param.annotation == param.empty ), f"Path params must be of type str, int, float or boot: {param}" param = signature_params[param_name] add_param_to_fields( param=param, dependant=dependant, default_schema=params.Path, force_type=params.ParamTypes.path, ) elif ( param.default == param.empty or param.default is None or type(param.default) in param_supported_types ) and ( param.annotation == param.empty or lenient_issubclass(param.annotation, param_supported_types) ): add_param_to_fields( param=param, dependant=dependant, default_schema=params.Query ) elif isinstance(param.default, params.Param): if param.annotation != param.empty: assert lenient_issubclass( param.annotation, param_supported_types ), f"Parameters for Path, Query, Header and Cookies must be of type str, int, float or bool: {param}" add_param_to_fields( param=param, dependant=dependant, default_schema=params.Query ) elif lenient_issubclass(param.annotation, Request): dependant.request_param_name = param_name elif not isinstance(param.default, params.Depends): add_param_to_body_fields(param=param, dependant=dependant) return dependant def add_param_to_fields( *, param: inspect.Parameter, dependant: Dependant, default_schema: Type[Schema] = params.Param, force_type: params.ParamTypes = None, ) -> None: default_value = Required if not param.default == param.empty: default_value = param.default if isinstance(default_value, params.Param): schema = default_value default_value = schema.default if getattr(schema, "in_", None) is None: schema.in_ = default_schema.in_ if force_type: schema.in_ = force_type else: schema = default_schema(default_value) required = default_value == Required annotation: Any = Any if not param.annotation == param.empty: annotation = param.annotation annotation = get_annotation_from_schema(annotation, schema) if not schema.alias and getattr(schema, "convert_underscores", None): alias = param.name.replace("_", "-") else: alias = schema.alias or param.name field = Field( name=param.name, type_=annotation, default=None if required else default_value, alias=alias, required=required, model_config=BaseConfig(), class_validators=[], schema=schema, ) if schema.in_ == params.ParamTypes.path: dependant.path_params.append(field) elif schema.in_ == params.ParamTypes.query: dependant.query_params.append(field) elif schema.in_ == params.ParamTypes.header: dependant.header_params.append(field) else: assert ( schema.in_ == params.ParamTypes.cookie ), f"non-body parameters must be in path, query, header or cookie: {param.name}" dependant.cookie_params.append(field) def add_param_to_body_fields(*, param: inspect.Parameter, dependant: Dependant) -> None: default_value = Required if not param.default == param.empty: default_value = param.default if isinstance(default_value, Schema): schema = default_value default_value = schema.default else: schema = Schema(default_value) required = default_value == Required annotation = get_annotation_from_schema(param.annotation, schema) field = Field( name=param.name, type_=annotation, default=None if required else default_value, alias=schema.alias or param.name, required=required, model_config=BaseConfig, class_validators=[], schema=schema, ) dependant.body_params.append(field) def is_coroutine_callable(call: Callable = None) -> bool: if not call: return False if inspect.isfunction(call): return asyncio.iscoroutinefunction(call) if inspect.isclass(call): return False call = getattr(call, "__call__", None) if not call: return False return asyncio.iscoroutinefunction(call) async def solve_dependencies( *, request: Request, dependant: Dependant, body: Dict[str, Any] = None ) -> Tuple[Dict[str, Any], List[ErrorWrapper]]: values: Dict[str, Any] = {} errors: List[ErrorWrapper] = [] for sub_dependant in dependant.dependencies: sub_values, sub_errors = await solve_dependencies( request=request, dependant=sub_dependant, body=body ) if sub_errors: return {}, errors assert sub_dependant.call is not None, "sub_dependant.call must be a function" if is_coroutine_callable(sub_dependant.call): solved = await sub_dependant.call(**sub_values) else: solved = await run_in_threadpool(sub_dependant.call, **sub_values) assert sub_dependant.name is not None, "Subdependants always have a name" values[sub_dependant.name] = solved path_values, path_errors = request_params_to_args( dependant.path_params, request.path_params ) query_values, query_errors = request_params_to_args( dependant.query_params, request.query_params ) header_values, header_errors = request_params_to_args( dependant.header_params, request.headers ) cookie_values, cookie_errors = request_params_to_args( dependant.cookie_params, request.cookies ) values.update(path_values) values.update(query_values) values.update(header_values) values.update(cookie_values) errors = path_errors + query_errors + header_errors + cookie_errors if dependant.body_params: body_values, body_errors = await request_body_to_args( # type: ignore # body_params checked above dependant.body_params, body ) values.update(body_values) errors.extend(body_errors) if dependant.request_param_name: values[dependant.request_param_name] = request return values, errors def request_params_to_args( required_params: Sequence[Field], received_params: Mapping[str, Any] ) -> Tuple[Dict[str, Any], List[ErrorWrapper]]: values = {} errors = [] for field in required_params: value = received_params.get(field.alias) if value is None: if field.required: errors.append( ErrorWrapper(MissingError(), loc=field.alias, config=BaseConfig) ) else: values[field.name] = deepcopy(field.default) continue schema: params.Param = field.schema assert isinstance(schema, params.Param), "Params must be subclasses of Param" v_, errors_ = field.validate(value, values, loc=(schema.in_.value, field.alias)) if isinstance(errors_, ErrorWrapper): errors.append(errors_) elif isinstance(errors_, list): errors.extend(errors_) else: values[field.name] = v_ return values, errors async def request_body_to_args( required_params: List[Field], received_body: Dict[str, Any] ) -> Tuple[Dict[str, Any], List[ErrorWrapper]]: values = {} errors = [] if required_params: field = required_params[0] embed = getattr(field.schema, "embed", None) if len(required_params) == 1 and not embed: received_body = {field.alias: received_body} for field in required_params: value = received_body.get(field.alias) if value is None: if field.required: errors.append( ErrorWrapper( MissingError(), loc=("body", field.alias), config=BaseConfig ) ) else: values[field.name] = deepcopy(field.default) continue v_, errors_ = field.validate(value, values, loc=("body", field.alias)) if isinstance(errors_, ErrorWrapper): errors.append(errors_) elif isinstance(errors_, list): errors.extend(errors_) else: values[field.name] = v_ return values, errors def get_body_field(*, dependant: Dependant, name: str) -> Field: flat_dependant = get_flat_dependant(dependant) if not flat_dependant.body_params: return None first_param = flat_dependant.body_params[0] embed = getattr(first_param.schema, "embed", None) if len(flat_dependant.body_params) == 1 and not embed: return first_param model_name = "Body_" + name BodyModel = create_model(model_name) for f in flat_dependant.body_params: BodyModel.__fields__[f.name] = f required = any(True for f in flat_dependant.body_params if f.required) if any(isinstance(f.schema, params.File) for f in flat_dependant.body_params): BodySchema: Type[params.Body] = params.File elif any(isinstance(f.schema, params.Form) for f in flat_dependant.body_params): BodySchema = params.Form else: BodySchema = params.Body field = Field( name="body", type_=BodyModel, default=None, required=required, model_config=BaseConfig, class_validators=[], alias="body", schema=BodySchema(None), ) return field PKYMfastapi/openapi/__init__.pyPKYMmaafastapi/openapi/constants.pyMETHODS_WITH_BODY = set(("POST", "PUT", "DELETE", "PATCH")) REF_PREFIX = "#/components/schemas/" PKɕMGOfastapi/openapi/docs.pyfrom starlette.responses import HTMLResponse def get_swagger_ui_html(*, openapi_url: str, title: str) -> HTMLResponse: return HTMLResponse( """ """ + title + """
""" ) def get_redoc_html(*, openapi_url: str, title: str) -> HTMLResponse: return HTMLResponse( """ """ + title + """ """ ) PKMeP)P)fastapi/openapi/models.pyimport logging from enum import Enum from typing import Any, Dict, List, Optional, Union from pydantic import BaseModel, Schema as PSchema from pydantic.types import UrlStr try: import email_validator from pydantic.types import EmailStr # type: ignore except ImportError: logging.warning( "email-validator not installed, email fields will be treated as str.\n" + "To install, run: pip install email-validator" ) class EmailStr(str): # type: ignore pass class Contact(BaseModel): name: Optional[str] = None url: Optional[UrlStr] = None email: Optional[EmailStr] = None class License(BaseModel): name: str url: Optional[UrlStr] = None class Info(BaseModel): title: str description: Optional[str] = None termsOfService: Optional[str] = None contact: Optional[Contact] = None license: Optional[License] = None version: str class ServerVariable(BaseModel): enum: Optional[List[str]] = None default: str description: Optional[str] = None class Server(BaseModel): url: UrlStr description: Optional[str] = None variables: Optional[Dict[str, ServerVariable]] = None class Reference(BaseModel): ref: str = PSchema(..., alias="$ref") # type: ignore 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 class ExternalDocumentation(BaseModel): description: Optional[str] = None url: UrlStr class SchemaBase(BaseModel): ref: Optional[str] = PSchema(None, alias="$ref") # type: ignore title: Optional[str] = None multipleOf: Optional[float] = None maximum: Optional[float] = None exclusiveMaximum: Optional[float] = None minimum: Optional[float] = None exclusiveMinimum: Optional[float] = None maxLength: Optional[int] = PSchema(None, gte=0) # type: ignore minLength: Optional[int] = PSchema(None, gte=0) # type: ignore pattern: Optional[str] = None maxItems: Optional[int] = PSchema(None, gte=0) # type: ignore minItems: Optional[int] = PSchema(None, gte=0) # type: ignore uniqueItems: Optional[bool] = None maxProperties: Optional[int] = PSchema(None, gte=0) # type: ignore minProperties: Optional[int] = PSchema(None, gte=0) # type: ignore required: Optional[List[str]] = None enum: Optional[List[str]] = None type: Optional[str] = None allOf: Optional[List[Any]] = None oneOf: Optional[List[Any]] = None anyOf: Optional[List[Any]] = None not_: Optional[List[Any]] = PSchema(None, alias="not") # type: ignore items: Optional[Any] = None properties: Optional[Dict[str, Any]] = None additionalProperties: Optional[Union[bool, Any]] = None description: Optional[str] = None format: Optional[str] = None default: Optional[Any] = None nullable: Optional[bool] = None discriminator: Optional[Discriminator] = None readOnly: Optional[bool] = None writeOnly: Optional[bool] = None xml: Optional[XML] = None externalDocs: Optional[ExternalDocumentation] = None example: Optional[Any] = None deprecated: Optional[bool] = None class Schema(SchemaBase): allOf: Optional[List[SchemaBase]] = None oneOf: Optional[List[SchemaBase]] = None anyOf: Optional[List[SchemaBase]] = None not_: Optional[List[SchemaBase]] = PSchema(None, alias="not") # type: ignore items: Optional[SchemaBase] = None properties: Optional[Dict[str, SchemaBase]] = None additionalProperties: Optional[Union[bool, SchemaBase]] = None class Example(BaseModel): summary: Optional[str] = None description: Optional[str] = None value: Optional[Any] = None externalValue: Optional[UrlStr] = None class ParameterInType(Enum): query = "query" header = "header" path = "path" cookie = "cookie" class Encoding(BaseModel): contentType: Optional[str] = None # Workaround OpenAPI recursive reference, using Any headers: Optional[Dict[str, Union[Any, Reference]]] = None style: Optional[str] = None explode: Optional[bool] = None allowReserved: Optional[bool] = None class MediaType(BaseModel): schema_: Optional[Union[Schema, Reference]] = PSchema( None, alias="schema" ) # type: ignore example: Optional[Any] = None examples: Optional[Dict[str, Union[Example, Reference]]] = None encoding: Optional[Dict[str, Encoding]] = None class ParameterBase(BaseModel): description: Optional[str] = None required: Optional[bool] = None deprecated: Optional[bool] = None # Serialization rules for simple scenarios style: Optional[str] = None explode: Optional[bool] = None allowReserved: Optional[bool] = None schema_: Optional[Union[Schema, Reference]] = PSchema( None, alias="schema" ) # type: ignore example: Optional[Any] = None examples: Optional[Dict[str, Union[Example, Reference]]] = None # Serialization rules for more complex scenarios content: Optional[Dict[str, MediaType]] = None class Parameter(ParameterBase): name: str in_: ParameterInType = PSchema(..., alias="in") # type: ignore class Header(ParameterBase): pass # Workaround OpenAPI recursive reference class EncodingWithHeaders(Encoding): headers: Optional[Dict[str, Union[Header, Reference]]] = None class RequestBody(BaseModel): description: Optional[str] = None content: Dict[str, MediaType] required: Optional[bool] = None 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 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 class Responses(BaseModel): default: Response 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: Union[Responses, Dict[Union[str], Response]] # Workaround OpenAPI recursive reference callbacks: Optional[Dict[str, Union[Dict[str, Any], Reference]]] = None deprecated: Optional[bool] = None security: Optional[List[Dict[str, List[str]]]] = None servers: Optional[List[Server]] = None class PathItem(BaseModel): ref: Optional[str] = PSchema(None, alias="$ref") # type: ignore 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 # Workaround OpenAPI recursive reference class OperationWithCallbacks(BaseModel): callbacks: Optional[Dict[str, Union[Dict[str, PathItem], Reference]]] = None class SecuritySchemeType(Enum): apiKey = "apiKey" http = "http" oauth2 = "oauth2" openIdConnect = "openIdConnect" class SecurityBase(BaseModel): type_: SecuritySchemeType = PSchema(..., alias="type") # type: ignore description: Optional[str] = None class APIKeyIn(Enum): query = "query" header = "header" cookie = "cookie" class APIKey(SecurityBase): type_ = PSchema(SecuritySchemeType.apiKey, alias="type") # type: ignore in_: APIKeyIn = PSchema(..., alias="in") # type: ignore name: str class HTTPBase(SecurityBase): type_ = PSchema(SecuritySchemeType.http, alias="type") # type: ignore scheme: str class HTTPBearer(HTTPBase): scheme = "bearer" bearerFormat: Optional[str] = None class OAuthFlow(BaseModel): 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(BaseModel): implicit: Optional[OAuthFlowImplicit] = None password: Optional[OAuthFlowPassword] = None clientCredentials: Optional[OAuthFlowClientCredentials] = None authorizationCode: Optional[OAuthFlowAuthorizationCode] = None class OAuth2(SecurityBase): type_ = PSchema(SecuritySchemeType.oauth2, alias="type") # type: ignore flows: OAuthFlows class OpenIdConnect(SecurityBase): type_ = PSchema(SecuritySchemeType.openIdConnect, alias="type") # type: ignore openIdConnectUrl: str SecurityScheme = Union[APIKey, HTTPBase, HTTPBearer, OAuth2, OpenIdConnect] 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]]] = None class Tag(BaseModel): name: str description: Optional[str] = None externalDocs: Optional[ExternalDocumentation] = None class OpenAPI(BaseModel): openapi: str info: Info servers: Optional[List[Server]] = None paths: Dict[str, PathItem] components: Optional[Components] = None security: Optional[List[Dict[str, List[str]]]] = None tags: Optional[List[Tag]] = None externalDocs: Optional[ExternalDocumentation] = None PKh?Md>$>$fastapi/openapi/utils.pyfrom typing import Any, Dict, List, Optional, Sequence, Tuple, Type from fastapi import routing from fastapi.dependencies.models import Dependant from fastapi.dependencies.utils import get_flat_dependant from fastapi.encoders import jsonable_encoder from fastapi.openapi.constants import METHODS_WITH_BODY, REF_PREFIX from fastapi.openapi.models import OpenAPI from fastapi.params import Body, Param from fastapi.utils import get_flat_models_from_routes, get_model_definitions from pydantic.fields import Field from pydantic.schema import Schema, field_schema, get_model_name_map from pydantic.utils import lenient_issubclass from starlette.responses import HTMLResponse, JSONResponse from starlette.routing import BaseRoute, Route from starlette.status import HTTP_422_UNPROCESSABLE_ENTITY validation_error_definition = { "title": "ValidationError", "type": "object", "properties": { "loc": {"title": "Location", "type": "array", "items": {"type": "string"}}, "msg": {"title": "Message", "type": "string"}, "type": {"title": "Error Type", "type": "string"}, }, "required": ["loc", "msg", "type"], } validation_error_response_definition = { "title": "HTTPValidationError", "type": "object", "properties": { "detail": { "title": "Detail", "type": "array", "items": {"$ref": REF_PREFIX + "ValidationError"}, } }, } def get_openapi_params(dependant: Dependant) -> List[Field]: flat_dependant = get_flat_dependant(dependant) return ( flat_dependant.path_params + flat_dependant.query_params + flat_dependant.header_params + flat_dependant.cookie_params ) def get_openapi_security_definitions(flat_dependant: Dependant) -> Tuple[Dict, List]: security_definitions = {} operation_security = [] for security_requirement in flat_dependant.security_requirements: security_definition = jsonable_encoder( security_requirement.security_scheme.model, by_alias=True, include_none=False, ) security_name = security_requirement.security_scheme.scheme_name security_definitions[security_name] = security_definition operation_security.append({security_name: security_requirement.scopes}) return security_definitions, operation_security def get_openapi_operation_parameters( all_route_params: Sequence[Field] ) -> Tuple[Dict[str, Dict], List[Dict[str, Any]]]: definitions: Dict[str, Dict] = {} parameters = [] for param in all_route_params: schema: Param = param.schema if "ValidationError" not in definitions: definitions["ValidationError"] = validation_error_definition definitions["HTTPValidationError"] = validation_error_response_definition parameter = { "name": param.alias, "in": schema.in_.value, "required": param.required, "schema": field_schema(param, model_name_map={})[0], } if schema.description: parameter["description"] = schema.description if schema.deprecated: parameter["deprecated"] = schema.deprecated parameters.append(parameter) return definitions, parameters def get_openapi_operation_request_body( *, body_field: Field, model_name_map: Dict[Type, str] ) -> Optional[Dict]: if not body_field: return None assert isinstance(body_field, Field) body_schema, _ = field_schema( body_field, model_name_map=model_name_map, ref_prefix=REF_PREFIX ) schema: Schema = body_field.schema if isinstance(schema, Body): request_media_type = schema.media_type else: # Includes not declared media types (Schema) request_media_type = "application/json" required = body_field.required request_body_oai: Dict[str, Any] = {} if required: request_body_oai["required"] = required request_body_oai["content"] = {request_media_type: {"schema": body_schema}} return request_body_oai def generate_operation_id(*, route: routing.APIRoute, method: str) -> str: if route.operation_id: return route.operation_id path: str = route.path operation_id = route.name + path operation_id = operation_id.replace("{", "_").replace("}", "_").replace("/", "_") operation_id = operation_id + "_" + method.lower() return operation_id def generate_operation_summary(*, route: routing.APIRoute, method: str) -> str: if route.summary: return route.summary return route.name.replace("_", " ").title() + " " + method.title() def get_openapi_operation_metadata(*, route: routing.APIRoute, method: str) -> Dict: operation: Dict[str, Any] = {} if route.tags: operation["tags"] = route.tags operation["summary"] = generate_operation_summary(route=route, method=method) if route.description: operation["description"] = route.description operation["operationId"] = generate_operation_id(route=route, method=method) if route.deprecated: operation["deprecated"] = route.deprecated return operation def get_openapi_path( *, route: routing.APIRoute, model_name_map: Dict[Type, str] ) -> Tuple[Dict, Dict, Dict]: path = {} security_schemes: Dict[str, Any] = {} definitions: Dict[str, Any] = {} assert route.methods is not None, "Methods must be a list" for method in route.methods: operation = get_openapi_operation_metadata(route=route, method=method) parameters: List[Dict] = [] flat_dependant = get_flat_dependant(route.dependant) security_definitions, operation_security = get_openapi_security_definitions( flat_dependant=flat_dependant ) if operation_security: operation.setdefault("security", []).extend(operation_security) if security_definitions: security_schemes.update(security_definitions) all_route_params = get_openapi_params(route.dependant) validation_definitions, operation_parameters = get_openapi_operation_parameters( all_route_params=all_route_params ) definitions.update(validation_definitions) parameters.extend(operation_parameters) if parameters: operation["parameters"] = parameters if method in METHODS_WITH_BODY: request_body_oai = get_openapi_operation_request_body( body_field=route.body_field, model_name_map=model_name_map ) if request_body_oai: operation["requestBody"] = request_body_oai status_code = str(route.status_code) response_schema = {"type": "string"} if lenient_issubclass(route.content_type, JSONResponse): if route.response_field: response_schema, _ = field_schema( route.response_field, model_name_map=model_name_map, ref_prefix=REF_PREFIX, ) else: response_schema = {} content = {route.content_type.media_type: {"schema": response_schema}} operation["responses"] = { status_code: {"description": route.response_description, "content": content} } if all_route_params or route.body_field: operation["responses"][str(HTTP_422_UNPROCESSABLE_ENTITY)] = { "description": "Validation Error", "content": { "application/json": { "schema": {"$ref": REF_PREFIX + "HTTPValidationError"} } }, } path[method.lower()] = operation return path, security_schemes, definitions def get_openapi( *, title: str, version: str, openapi_version: str = "3.0.2", description: str = None, routes: Sequence[BaseRoute] ) -> Dict: info = {"title": title, "version": version} if description: info["description"] = description output = {"openapi": openapi_version, "info": info} components: Dict[str, Dict] = {} paths: Dict[str, Dict] = {} flat_models = get_flat_models_from_routes(routes) model_name_map = get_model_name_map(flat_models) definitions = get_model_definitions( flat_models=flat_models, model_name_map=model_name_map ) for route in routes: if isinstance(route, routing.APIRoute): result = get_openapi_path(route=route, model_name_map=model_name_map) if result: path, security_schemes, path_definitions = result if path: paths.setdefault(route.path, {}).update(path) if security_schemes: components.setdefault("securitySchemes", {}).update( security_schemes ) if path_definitions: definitions.update(path_definitions) if definitions: components.setdefault("schemas", {}).update(definitions) if components: output["components"] = components output["paths"] = paths return jsonable_encoder(OpenAPI(**output), by_alias=True, include_none=False) PKh?M&#fastapi/security/__init__.pyfrom .api_key import APIKeyQuery, APIKeyHeader, APIKeyCookie from .http import HTTPBasic, HTTPBearer, HTTPDigest from .oauth2 import OAuth2PasswordRequestForm, OAuth2 from .open_id_connect_url import OpenIdConnect PKMn^fastapi/security/api_key.pyfrom fastapi.openapi.models import APIKey, APIKeyIn from fastapi.security.base import SecurityBase from starlette.requests import Request class APIKeyBase(SecurityBase): pass class APIKeyQuery(APIKeyBase): def __init__(self, *, name: str, scheme_name: str = None): self.model = APIKey(in_=APIKeyIn.query, name=name) self.scheme_name = scheme_name or self.__class__.__name__ async def __call__(self, requests: Request) -> str: return requests.query_params.get(self.model.name) class APIKeyHeader(APIKeyBase): def __init__(self, *, name: str, scheme_name: str = None): self.model = APIKey(in_=APIKeyIn.header, name=name) self.scheme_name = scheme_name or self.__class__.__name__ async def __call__(self, requests: Request) -> str: return requests.headers.get(self.model.name) class APIKeyCookie(APIKeyBase): def __init__(self, *, name: str, scheme_name: str = None): self.model = APIKey(in_=APIKeyIn.cookie, name=name) self.scheme_name = scheme_name or self.__class__.__name__ async def __call__(self, requests: Request) -> str: return requests.cookies.get(self.model.name) PKMr(ԍfastapi/security/base.pyfrom fastapi.openapi.models import SecurityBase as SecurityBaseModel class SecurityBase: model: SecurityBaseModel scheme_name: str PKM,_^fastapi/security/http.pyfrom fastapi.openapi.models import ( HTTPBase as HTTPBaseModel, HTTPBearer as HTTPBearerModel, ) from fastapi.security.base import SecurityBase from starlette.requests import Request class HTTPBase(SecurityBase): def __init__(self, *, scheme: str, scheme_name: str = None): self.model = HTTPBaseModel(scheme=scheme) self.scheme_name = scheme_name or self.__class__.__name__ async def __call__(self, request: Request) -> str: return request.headers.get("Authorization") class HTTPBasic(HTTPBase): def __init__(self, *, scheme_name: str = None): self.model = HTTPBaseModel(scheme="basic") self.scheme_name = scheme_name or self.__class__.__name__ async def __call__(self, request: Request) -> str: return request.headers.get("Authorization") class HTTPBearer(HTTPBase): def __init__(self, *, bearerFormat: str = None, scheme_name: str = None): self.model = HTTPBearerModel(bearerFormat=bearerFormat) self.scheme_name = scheme_name or self.__class__.__name__ async def __call__(self, request: Request) -> str: return request.headers.get("Authorization") class HTTPDigest(HTTPBase): def __init__(self, *, scheme_name: str = None): self.model = HTTPBaseModel(scheme="digest") self.scheme_name = scheme_name or self.__class__.__name__ async def __call__(self, request: Request) -> str: return request.headers.get("Authorization") PK*M8 fastapi/security/oauth2.pyfrom typing import List, Optional from fastapi.openapi.models import OAuth2 as OAuth2Model, OAuthFlows as OAuthFlowsModel from fastapi.security.base import SecurityBase from pydantic import BaseModel, Schema from starlette.requests import Request class OAuth2PasswordRequestData(BaseModel): grant_type: str = "password" username: str password: str scope: Optional[List[str]] = None # Client ID and secret might come from headers client_id: Optional[str] = None client_secret: Optional[str] = None class OAuth2PasswordRequestForm(BaseModel): grant_type: str = Schema(..., regex="password") # it must have the value "password" username: str password: str scope: str = "" # Client ID and secret might come from headers client_id: Optional[str] = None client_secret: Optional[str] = None def parse(self) -> OAuth2PasswordRequestData: return OAuth2PasswordRequestData( grant_type=self.grant_type, username=self.username, password=self.password, scope=self.scope.split(), client_id=self.client_id, client_secret=self.client_secret, ) class OAuth2(SecurityBase): def __init__( self, *, flows: OAuthFlowsModel = OAuthFlowsModel(), scheme_name: str = None ): self.model = OAuth2Model(flows=flows) self.scheme_name = scheme_name or self.__class__.__name__ async def __call__(self, request: Request) -> str: return request.headers.get("Authorization") PKMJ'fastapi/security/open_id_connect_url.pyfrom fastapi.openapi.models import OpenIdConnect as OpenIdConnectModel from fastapi.security.base import SecurityBase from starlette.requests import Request class OpenIdConnect(SecurityBase): def __init__(self, *, openIdConnectUrl: str, scheme_name: str = None): self.model = OpenIdConnectModel(openIdConnectUrl=openIdConnectUrl) self.scheme_name = scheme_name or self.__class__.__name__ async def __call__(self, request: Request) -> str: return request.headers.get("Authorization") PKYqcM{>>fastapi-0.1.6.dist-info/LICENSEThe MIT License (MIT) Copyright (c) 2018 Sebastián Ramírez Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. PK!H>*RQfastapi-0.1.6.dist-info/WHEEL HM K-*ϳR03rOK-J,/RH,rzd&Y)r$[)T&UrPK!H%Hf<0 fastapi-0.1.6.dist-info/METADATAZrFO1\$ N.dzȎey%y][1XTf`b_HQwS4z{3sLÿ&XLeeea$6L˱Oޜi )obgsQrjL%FJycl,LĢ(MR?L 9skj<ʹד06jLjFtt o_wM9WjRf{~*:ןfLK֥oG_C窊K]|S[V}#&1<8MeUV &cqqu&N xz>?n^}\ݪ؋++m]7b(:~(Ne] ;D'Ѿ) 2+ؤ~>/ظy\:7d2W~{nTyKvhWXTYC;2ɷ}#GGG>H-DT7RʔA a&(Zvs +M ,A}҆c `q&/u珴9ii:T^mpga~\>:v2 @D9E4-NHT*≴U{pbb i5IWvY? 5exoScM8we~nΑ-"iNAW7+[SQS#-꽙NiobL m"3)LE\Wo֘<{1e1+6Y /{G!gG c83uW[(E0ݟXR]@2hi+ 99aST Z̃<|LQ@μ:&uP! msp>`L1̬_},t37 @rF|i+Lm{&kNI "oCbB -ߙ䟫7"_Wڪk5a |*cքΩq}z߅fʢʠm1tN{x@ _U \qËS :D7.[쿸=ybŢEc$RL TYefhPrߪ-Gkĝ[$㊏/$:/jq[B-o1!m'i$'(XȬjyx?S;~j̍-#.Ӫ 9Q0SDzȈ~!uHQ1-3{̤Bt-ѽ융B3 '&V4H(խ6u6RN.B|AQɛy!/Qvndg^PMc<7֕HTQ:o;IO<t tm'NǣdD8qŤmK)'r5x}y^*1a"Y0(>IĆssQҎ۳̚r>mܤdy3h7U;qM|˶F+6%lЇ*N[NO !3#4tts E)`5\Loj?fCA p޽qcEy9?}vO=';J2M]>]WIHal%[fM3p{_ICtsv&w}^|k3F!k0K7}3&ץ+=׏"o+PE+ł7'ASId!=؟_ԲWAEgc]d@8Q撂Hʝ[<_Up~ ^Ȧ*RrZsW8*`98gEh&U łmb#sI<(tVgS<" j>ЉS0 ͱz"?W%x|0 !-#h s[=F~w.](@ϬVKwNc]{OO|w oӜOPw1 ծ/N`QW( sOyO4<CZΑ !\3&#/\ϯ聘LyH%&c\FXC8xLޅCe뢩J+m6M֭젪*-y)Pq}F~̗~["SJ"ʻ&TsB#-o ۛB pBѦ(L #C5hm4͢l?s442\6ujoA59e=dk\kh Ҽ 9?q9蘃bS.:j;naav]<$j*y HS= #! J&iY^\؉xq®$GBz`ӬsCW+*=rNy\~8I}+. ]d1F,zrjr֣Ow· ^X![J[ nv>yDbo)1Fibt=oiOtۋ55ԽRGp7= 8PK!Hn7fastapi-0.1.6.dist-info/RECORDDzHཟaI!aEB.Weu% :'1M9ѭp0A0ȃrwԮLYVb3^joTu2ƲmN%v'(m\6n)!TO!E75If+B0M7B,ؚ9=y-<>_% `kJ*vUhL%KAr+^MSi,kY8sLV"W駤4Γt] BYƞ&^%lѥ쁖Y_z8:Y/pjE>!&'f]֤ۅU-JFweVj&B6YI^(k9ZoYgX 6wѡi"K|ֳDK9:68k+p=ѥ)`XhLWBt,zwydSq]vSRw_`zuLMB/mY';-DFM$>c&cьiK\ь̕',UTL!qgΝV45DiVq C~/5AS*MP7`m^/\9Ռz~G wkXh Y2~ {5#6rV6鍶#{c@;Ļ}*[_[&Zc+ q^v E݌!{ͱ5< ^hY!w_%w2݋ ;tf#VqJZ?J&:v`k8-`M{!c3fAn|yLsa媗P?Se 7Y2.2uSe,g]_dbbؓ@$F>#-.~TdIu79PvGX+޶AϠ=Iq;cXFҶ%KΨS_Q{0p딡ˡ،V*@ WSlNl|C4TJAcK |C#nQ ]7ٔ!?PKMQ#  fastapi/__init__.pyPK]MH"55<fastapi/applications.pyPKM4 ??j7fastapi/encoders.pyPKM+͖<fastapi/params.pyPKM oKKZfastapi/routing.pyPKM.{fastapi/utils.pyPKρM zfastapi/dependencies/__init__.pyPKM.pQFFfastapi/dependencies/models.pyPKݐM9r22:fastapi/dependencies/utils.pyPKYMlfastapi/openapi/__init__.pyPKYMmaafastapi/openapi/constants.pyPKɕMGO@fastapi/openapi/docs.pyPKMeP)P)Dfastapi/openapi/models.pyPKh?Md>$>$fastapi/openapi/utils.pyPKh?M&#?=fastapi/security/__init__.pyPKMn^O>fastapi/security/api_key.pyPKMr(ԍ'Cfastapi/security/base.pyPKM,_^Cfastapi/security/http.pyPK*M8 Ifastapi/security/oauth2.pyPKMJ'Pfastapi/security/open_id_connect_url.pyPKYqcM{>>jRfastapi-0.1.6.dist-info/LICENSEPK!H>*RQVfastapi-0.1.6.dist-info/WHEELPK!H%Hf<0 rWfastapi-0.1.6.dist-info/METADATAPK!Hn7chfastapi-0.1.6.dist-info/RECORDPKl