PK}oN)>fastapi/__init__.py"""FastAPI framework, high performance, easy to learn, fast to code, ready for production""" __version__ = "0.27.0" from starlette.background import BackgroundTasks from .applications import FastAPI from .datastructures import UploadFile from .exceptions import HTTPException from .param_functions import ( Body, Cookie, Depends, File, Form, Header, Path, Query, Security, ) from .routing import APIRouter PK}oNyUUfastapi/applications.pyfrom typing import Any, Callable, Dict, List, Optional, Set, Type, Union from fastapi import routing from fastapi.exception_handlers import ( http_exception_handler, request_validation_exception_handler, ) from fastapi.exceptions import RequestValidationError from fastapi.openapi.docs import ( get_redoc_html, get_swagger_ui_html, get_swagger_ui_oauth2_redirect_html, ) from fastapi.openapi.utils import get_openapi from fastapi.params import Depends from starlette.applications import Starlette from starlette.exceptions import ExceptionMiddleware, HTTPException from starlette.middleware.errors import ServerErrorMiddleware from starlette.requests import Request from starlette.responses import HTMLResponse, JSONResponse, Response from starlette.routing import BaseRoute class FastAPI(Starlette): def __init__( self, debug: bool = False, routes: List[BaseRoute] = None, template_directory: str = None, title: str = "Fast API", description: str = "", version: str = "0.1.0", openapi_url: Optional[str] = "/openapi.json", openapi_prefix: str = "", docs_url: Optional[str] = "/docs", redoc_url: Optional[str] = "/redoc", swagger_ui_oauth2_redirect_url: Optional[str] = "/docs/oauth2-redirect", **extra: Dict[str, Any], ) -> None: self._debug = debug self.router: routing.APIRouter = routing.APIRouter(routes) self.exception_middleware = ExceptionMiddleware(self.router, debug=debug) self.error_middleware = ServerErrorMiddleware( self.exception_middleware, debug=debug ) self.title = title self.description = description self.version = version self.openapi_url = openapi_url self.openapi_prefix = openapi_prefix.rstrip("/") self.docs_url = docs_url self.redoc_url = redoc_url self.swagger_ui_oauth2_redirect_url = swagger_ui_oauth2_redirect_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.docs_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, openapi_prefix=self.openapi_prefix, ) return self.openapi_schema def setup(self) -> None: if self.openapi_url: async def openapi(req: Request) -> JSONResponse: return JSONResponse(self.openapi()) self.add_route(self.openapi_url, openapi, include_in_schema=False) openapi_url = self.openapi_prefix + self.openapi_url if self.openapi_url and self.docs_url: async def swagger_ui_html(req: Request) -> HTMLResponse: return get_swagger_ui_html( openapi_url=openapi_url, title=self.title + " - Swagger UI", oauth2_redirect_url=self.swagger_ui_oauth2_redirect_url, ) self.add_route(self.docs_url, swagger_ui_html, include_in_schema=False) if self.swagger_ui_oauth2_redirect_url: async def swagger_ui_redirect(req: Request) -> HTMLResponse: return get_swagger_ui_oauth2_redirect_html() self.add_route( self.swagger_ui_oauth2_redirect_url, swagger_ui_redirect, include_in_schema=False, ) if self.openapi_url and self.redoc_url: async def redoc_html(req: Request) -> HTMLResponse: return get_redoc_html( openapi_url=openapi_url, title=self.title + " - ReDoc" ) self.add_route(self.redoc_url, redoc_html, include_in_schema=False) self.add_exception_handler(HTTPException, http_exception_handler) self.add_exception_handler( RequestValidationError, request_validation_exception_handler ) def add_api_route( self, path: str, endpoint: Callable, *, response_model: Type[Any] = None, status_code: int = 200, tags: List[str] = None, dependencies: List[Depends] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", responses: Dict[Union[int, str], Dict[str, Any]] = None, deprecated: bool = None, methods: List[str] = None, operation_id: str = None, response_model_include: Set[str] = None, response_model_exclude: Set[str] = set(), response_model_by_alias: bool = True, response_model_skip_defaults: bool = False, include_in_schema: bool = True, response_class: Type[Response] = JSONResponse, name: str = None, ) -> None: self.router.add_api_route( path, endpoint=endpoint, response_model=response_model, status_code=status_code, tags=tags or [], dependencies=dependencies or [], summary=summary, description=description, response_description=response_description, responses=responses or {}, deprecated=deprecated, methods=methods, operation_id=operation_id, response_model_include=response_model_include, response_model_exclude=response_model_exclude, response_model_by_alias=response_model_by_alias, response_model_skip_defaults=response_model_skip_defaults, include_in_schema=include_in_schema, response_class=response_class, name=name, ) def api_route( self, path: str, *, response_model: Type[Any] = None, status_code: int = 200, tags: List[str] = None, dependencies: List[Depends] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", responses: Dict[Union[int, str], Dict[str, Any]] = None, deprecated: bool = None, methods: List[str] = None, operation_id: str = None, response_model_include: Set[str] = None, response_model_exclude: Set[str] = set(), response_model_by_alias: bool = True, response_model_skip_defaults: bool = False, include_in_schema: bool = True, response_class: Type[Response] = JSONResponse, name: str = None, ) -> Callable: def decorator(func: Callable) -> Callable: self.router.add_api_route( path, func, response_model=response_model, status_code=status_code, tags=tags or [], dependencies=dependencies or [], summary=summary, description=description, response_description=response_description, responses=responses or {}, deprecated=deprecated, methods=methods, operation_id=operation_id, response_model_include=response_model_include, response_model_exclude=response_model_exclude, response_model_by_alias=response_model_by_alias, response_model_skip_defaults=response_model_skip_defaults, include_in_schema=include_in_schema, response_class=response_class, name=name, ) return func return decorator def add_api_websocket_route( self, path: str, endpoint: Callable, name: str = None ) -> None: self.router.add_api_websocket_route(path, endpoint, name=name) def websocket(self, path: str, name: str = None) -> Callable: def decorator(func: Callable) -> Callable: self.add_api_websocket_route(path, func, name=name) return func return decorator def include_router( self, router: routing.APIRouter, *, prefix: str = "", tags: List[str] = None, dependencies: List[Depends] = None, responses: Dict[Union[int, str], Dict[str, Any]] = None, ) -> None: self.router.include_router( router, prefix=prefix, tags=tags, dependencies=dependencies, responses=responses or {}, ) def get( self, path: str, *, response_model: Type[Any] = None, status_code: int = 200, tags: List[str] = None, dependencies: List[Depends] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", responses: Dict[Union[int, str], Dict[str, Any]] = None, deprecated: bool = None, operation_id: str = None, response_model_include: Set[str] = None, response_model_exclude: Set[str] = set(), response_model_by_alias: bool = True, response_model_skip_defaults: bool = False, include_in_schema: bool = True, response_class: Type[Response] = JSONResponse, name: str = None, ) -> Callable: return self.router.get( path, response_model=response_model, status_code=status_code, tags=tags or [], dependencies=dependencies or [], summary=summary, description=description, response_description=response_description, responses=responses or {}, deprecated=deprecated, operation_id=operation_id, response_model_include=response_model_include, response_model_exclude=response_model_exclude, response_model_by_alias=response_model_by_alias, response_model_skip_defaults=response_model_skip_defaults, include_in_schema=include_in_schema, response_class=response_class, name=name, ) def put( self, path: str, *, response_model: Type[Any] = None, status_code: int = 200, tags: List[str] = None, dependencies: List[Depends] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", responses: Dict[Union[int, str], Dict[str, Any]] = None, deprecated: bool = None, operation_id: str = None, response_model_include: Set[str] = None, response_model_exclude: Set[str] = set(), response_model_by_alias: bool = True, response_model_skip_defaults: bool = False, include_in_schema: bool = True, response_class: Type[Response] = JSONResponse, name: str = None, ) -> Callable: return self.router.put( path, response_model=response_model, status_code=status_code, tags=tags or [], dependencies=dependencies or [], summary=summary, description=description, response_description=response_description, responses=responses or {}, deprecated=deprecated, operation_id=operation_id, response_model_include=response_model_include, response_model_exclude=response_model_exclude, response_model_by_alias=response_model_by_alias, response_model_skip_defaults=response_model_skip_defaults, include_in_schema=include_in_schema, response_class=response_class, name=name, ) def post( self, path: str, *, response_model: Type[Any] = None, status_code: int = 200, tags: List[str] = None, dependencies: List[Depends] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", responses: Dict[Union[int, str], Dict[str, Any]] = None, deprecated: bool = None, operation_id: str = None, response_model_include: Set[str] = None, response_model_exclude: Set[str] = set(), response_model_by_alias: bool = True, response_model_skip_defaults: bool = False, include_in_schema: bool = True, response_class: Type[Response] = JSONResponse, name: str = None, ) -> Callable: return self.router.post( path, response_model=response_model, status_code=status_code, tags=tags or [], dependencies=dependencies or [], summary=summary, description=description, response_description=response_description, responses=responses or {}, deprecated=deprecated, operation_id=operation_id, response_model_include=response_model_include, response_model_exclude=response_model_exclude, response_model_by_alias=response_model_by_alias, response_model_skip_defaults=response_model_skip_defaults, include_in_schema=include_in_schema, response_class=response_class, name=name, ) def delete( self, path: str, *, response_model: Type[Any] = None, status_code: int = 200, tags: List[str] = None, dependencies: List[Depends] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", responses: Dict[Union[int, str], Dict[str, Any]] = None, deprecated: bool = None, operation_id: str = None, response_model_include: Set[str] = None, response_model_exclude: Set[str] = set(), response_model_by_alias: bool = True, response_model_skip_defaults: bool = False, include_in_schema: bool = True, response_class: Type[Response] = JSONResponse, name: str = None, ) -> Callable: return self.router.delete( path, response_model=response_model, status_code=status_code, tags=tags or [], dependencies=dependencies or [], summary=summary, description=description, response_description=response_description, responses=responses or {}, deprecated=deprecated, response_model_include=response_model_include, response_model_exclude=response_model_exclude, response_model_by_alias=response_model_by_alias, operation_id=operation_id, response_model_skip_defaults=response_model_skip_defaults, include_in_schema=include_in_schema, response_class=response_class, name=name, ) def options( self, path: str, *, response_model: Type[Any] = None, status_code: int = 200, tags: List[str] = None, dependencies: List[Depends] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", responses: Dict[Union[int, str], Dict[str, Any]] = None, deprecated: bool = None, operation_id: str = None, response_model_include: Set[str] = None, response_model_exclude: Set[str] = set(), response_model_by_alias: bool = True, response_model_skip_defaults: bool = False, include_in_schema: bool = True, response_class: Type[Response] = JSONResponse, name: str = None, ) -> Callable: return self.router.options( path, response_model=response_model, status_code=status_code, tags=tags or [], dependencies=dependencies or [], summary=summary, description=description, response_description=response_description, responses=responses or {}, deprecated=deprecated, operation_id=operation_id, response_model_include=response_model_include, response_model_exclude=response_model_exclude, response_model_by_alias=response_model_by_alias, response_model_skip_defaults=response_model_skip_defaults, include_in_schema=include_in_schema, response_class=response_class, name=name, ) def head( self, path: str, *, response_model: Type[Any] = None, status_code: int = 200, tags: List[str] = None, dependencies: List[Depends] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", responses: Dict[Union[int, str], Dict[str, Any]] = None, deprecated: bool = None, operation_id: str = None, response_model_include: Set[str] = None, response_model_exclude: Set[str] = set(), response_model_by_alias: bool = True, response_model_skip_defaults: bool = False, include_in_schema: bool = True, response_class: Type[Response] = JSONResponse, name: str = None, ) -> Callable: return self.router.head( path, response_model=response_model, status_code=status_code, tags=tags or [], dependencies=dependencies or [], summary=summary, description=description, response_description=response_description, responses=responses or {}, deprecated=deprecated, operation_id=operation_id, response_model_include=response_model_include, response_model_exclude=response_model_exclude, response_model_by_alias=response_model_by_alias, response_model_skip_defaults=response_model_skip_defaults, include_in_schema=include_in_schema, response_class=response_class, name=name, ) def patch( self, path: str, *, response_model: Type[Any] = None, status_code: int = 200, tags: List[str] = None, dependencies: List[Depends] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", responses: Dict[Union[int, str], Dict[str, Any]] = None, deprecated: bool = None, operation_id: str = None, response_model_include: Set[str] = None, response_model_exclude: Set[str] = set(), response_model_by_alias: bool = True, response_model_skip_defaults: bool = False, include_in_schema: bool = True, response_class: Type[Response] = JSONResponse, name: str = None, ) -> Callable: return self.router.patch( path, response_model=response_model, status_code=status_code, tags=tags or [], dependencies=dependencies or [], summary=summary, description=description, response_description=response_description, responses=responses or {}, deprecated=deprecated, operation_id=operation_id, response_model_include=response_model_include, response_model_exclude=response_model_exclude, response_model_by_alias=response_model_by_alias, response_model_skip_defaults=response_model_skip_defaults, include_in_schema=include_in_schema, response_class=response_class, name=name, ) def trace( self, path: str, *, response_model: Type[Any] = None, status_code: int = 200, tags: List[str] = None, dependencies: List[Depends] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", responses: Dict[Union[int, str], Dict[str, Any]] = None, deprecated: bool = None, operation_id: str = None, response_model_include: Set[str] = None, response_model_exclude: Set[str] = set(), response_model_by_alias: bool = True, response_model_skip_defaults: bool = False, include_in_schema: bool = True, response_class: Type[Response] = JSONResponse, name: str = None, ) -> Callable: return self.router.trace( path, response_model=response_model, status_code=status_code, tags=tags or [], dependencies=dependencies or [], summary=summary, description=description, response_description=response_description, responses=responses or {}, deprecated=deprecated, operation_id=operation_id, response_model_include=response_model_include, response_model_exclude=response_model_exclude, response_model_by_alias=response_model_by_alias, response_model_skip_defaults=response_model_skip_defaults, include_in_schema=include_in_schema, response_class=response_class, name=name, ) PK}oNfastapi/datastructures.pyfrom typing import Any, Callable, Iterable, Type from starlette.datastructures import UploadFile as StarletteUploadFile class UploadFile(StarletteUploadFile): @classmethod def __get_validators__(cls: Type["UploadFile"]) -> Iterable[Callable]: yield cls.validate @classmethod def validate(cls: Type["UploadFile"], v: Any) -> Any: if not isinstance(v, StarletteUploadFile): raise ValueError(f"Expected UploadFile, received: {type(v)}") return v PK}oN Ŀfastapi/encoders.pyfrom enum import Enum from types import GeneratorType from typing import Any, List, Set from pydantic import BaseModel from pydantic.json import ENCODERS_BY_TYPE def jsonable_encoder( obj: Any, include: Set[str] = None, exclude: Set[str] = set(), by_alias: bool = True, skip_defaults: bool = False, include_none: bool = True, custom_encoder: dict = {}, sqlalchemy_safe: bool = True, ) -> Any: if include is not None and not isinstance(include, set): include = set(include) if exclude is not None and not isinstance(exclude, set): exclude = set(exclude) if isinstance(obj, BaseModel): encoder = getattr(obj.Config, "json_encoders", custom_encoder) return jsonable_encoder( obj.dict( include=include, exclude=exclude, by_alias=by_alias, skip_defaults=skip_defaults, ), include_none=include_none, custom_encoder=encoder, sqlalchemy_safe=sqlalchemy_safe, ) if isinstance(obj, Enum): return obj.value if isinstance(obj, (str, int, float, type(None))): return obj if isinstance(obj, dict): encoded_dict = {} for key, value in obj.items(): if ( ( not sqlalchemy_safe or (not isinstance(key, str)) or (not key.startswith("_sa")) ) and (value is not None or include_none) and ((include and key in include) or key not in exclude) ): encoded_key = jsonable_encoder( key, by_alias=by_alias, skip_defaults=skip_defaults, include_none=include_none, custom_encoder=custom_encoder, sqlalchemy_safe=sqlalchemy_safe, ) encoded_value = jsonable_encoder( value, by_alias=by_alias, skip_defaults=skip_defaults, include_none=include_none, custom_encoder=custom_encoder, sqlalchemy_safe=sqlalchemy_safe, ) encoded_dict[encoded_key] = encoded_value return encoded_dict if isinstance(obj, (list, set, frozenset, GeneratorType, tuple)): encoded_list = [] for item in obj: encoded_list.append( jsonable_encoder( item, include=include, exclude=exclude, by_alias=by_alias, skip_defaults=skip_defaults, include_none=include_none, custom_encoder=custom_encoder, sqlalchemy_safe=sqlalchemy_safe, ) ) return encoded_list errors: List[Exception] = [] try: if custom_encoder and type(obj) in custom_encoder: encoder = custom_encoder[type(obj)] else: encoder = ENCODERS_BY_TYPE[type(obj)] return encoder(obj) except KeyError as e: errors.append(e) try: data = dict(obj) except Exception as e: errors.append(e) try: data = vars(obj) except Exception as e: errors.append(e) raise ValueError(errors) return jsonable_encoder( data, by_alias=by_alias, skip_defaults=skip_defaults, include_none=include_none, custom_encoder=custom_encoder, sqlalchemy_safe=sqlalchemy_safe, ) PK}oNGGfastapi/exception_handlers.pyfrom fastapi.exceptions import RequestValidationError from starlette.exceptions import HTTPException from starlette.requests import Request from starlette.responses import JSONResponse from starlette.status import HTTP_422_UNPROCESSABLE_ENTITY async def http_exception_handler(request: Request, exc: HTTPException) -> JSONResponse: headers = getattr(exc, "headers", None) if headers: return JSONResponse( {"detail": exc.detail}, status_code=exc.status_code, headers=headers ) else: return JSONResponse({"detail": exc.detail}, status_code=exc.status_code) async def request_validation_exception_handler( request: Request, exc: RequestValidationError ) -> JSONResponse: return JSONResponse( status_code=HTTP_422_UNPROCESSABLE_ENTITY, content={"detail": exc.errors()} ) PK}oNfastapi/exceptions.pyfrom pydantic import ValidationError from starlette.exceptions import HTTPException as StarletteHTTPException class HTTPException(StarletteHTTPException): def __init__( self, status_code: int, detail: str = None, headers: dict = None ) -> None: super().__init__(status_code=status_code, detail=detail) self.headers = headers class RequestValidationError(ValidationError): pass class WebSocketRequestValidationError(ValidationError): pass PK}oNhSSfastapi/param_functions.pyfrom typing import Any, Callable, Sequence from fastapi import params def Path( # noqa: N802 default: Any, *, 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, deprecated: bool = None, **extra: Any, ) -> Any: return params.Path( default=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, deprecated=deprecated, **extra, ) def Query( # noqa: N802 default: Any, *, 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, deprecated: bool = None, **extra: Any, ) -> Any: return params.Query( 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, deprecated=deprecated, **extra, ) def Header( # noqa: N802 default: Any, *, 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, deprecated: bool = None, **extra: Any, ) -> Any: return params.Header( default, alias=alias, convert_underscores=convert_underscores, title=title, description=description, gt=gt, ge=ge, lt=lt, le=le, min_length=min_length, max_length=max_length, regex=regex, deprecated=deprecated, **extra, ) def Cookie( # noqa: N802 default: Any, *, 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, deprecated: bool = None, **extra: Any, ) -> Any: return params.Cookie( 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, deprecated=deprecated, **extra, ) def Body( # noqa: N802 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, ) -> Any: return params.Body( default, embed=embed, 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, ) def Form( # noqa: N802 default: Any, *, 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, ) -> Any: return params.Form( default, 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, ) def File( # noqa: N802 default: Any, *, 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, ) -> Any: return params.File( default, 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, ) def Depends(dependency: Callable = None) -> Any: # noqa: N802 return params.Depends(dependency=dependency) def Security( # noqa: N802 dependency: Callable = None, scopes: Sequence[str] = None ) -> Any: return params.Security(dependency=dependency, scopes=scopes) PK}oNVPCfastapi/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, *, 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, deprecated: bool = 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, *, 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, deprecated: bool = None, **extra: Any, ): self.in_ = self.in_ super().__init__( ..., alias=alias, title=title, description=description, gt=gt, ge=ge, lt=lt, le=le, min_length=min_length, max_length=max_length, regex=regex, deprecated=deprecated, **extra, ) class Query(Param): in_ = ParamTypes.query def __init__( self, default: Any, *, 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, deprecated: bool = None, **extra: Any, ): 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, deprecated=deprecated, **extra, ) class Header(Param): in_ = ParamTypes.header def __init__( self, default: Any, *, 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, deprecated: bool = None, **extra: Any, ): 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, deprecated=deprecated, **extra, ) class Cookie(Param): in_ = ParamTypes.cookie def __init__( self, default: Any, *, 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, deprecated: bool = None, **extra: Any, ): 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, deprecated=deprecated, **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, *, 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=True, 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, *, 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, 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) PK}oNfastapi/py.typedPK}oNYL7vvfastapi/routing.pyimport asyncio import inspect import logging import re from typing import Any, Callable, Dict, List, Optional, Set, Type, Union from fastapi import params from fastapi.dependencies.models import Dependant from fastapi.dependencies.utils import ( get_body_field, get_dependant, get_parameterless_sub_dependant, solve_dependencies, ) from fastapi.encoders import jsonable_encoder from fastapi.exceptions import RequestValidationError, WebSocketRequestValidationError 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.requests import Request from starlette.responses import JSONResponse, Response from starlette.routing import ( compile_path, get_name, request_response, websocket_session, ) from starlette.status import WS_1008_POLICY_VIOLATION from starlette.websockets import WebSocket def serialize_response( *, field: Field = None, response: Response, include: Set[str] = None, exclude: Set[str] = set(), by_alias: bool = True, skip_defaults: bool = False, ) -> Any: encoded = jsonable_encoder( response, include=include, exclude=exclude, by_alias=by_alias, skip_defaults=skip_defaults, ) if field: errors = [] value, errors_ = field.validate(encoded, {}, 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, include=include, exclude=exclude, by_alias=by_alias, skip_defaults=skip_defaults, ) else: return encoded def get_app( dependant: Dependant, body_field: Field = None, status_code: int = 200, response_class: Type[Response] = JSONResponse, response_field: Field = None, response_model_include: Set[str] = None, response_model_exclude: Set[str] = set(), response_model_by_alias: bool = True, response_model_skip_defaults: bool = False, ) -> Callable: assert dependant.call is not None, "dependant.call must be 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: body = await request.form() else: body_bytes = await request.body() if body_bytes: body = await request.json() except Exception as e: logging.error(f"Error getting request body: {e}") raise HTTPException( status_code=400, detail="There was an error parsing the body" ) from e values, errors, background_tasks = await solve_dependencies( request=request, dependant=dependant, body=body ) if errors: raise RequestValidationError(errors) else: assert dependant.call is not None, "dependant.call must be 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): if raw_response.background is None: raw_response.background = background_tasks return raw_response response_data = serialize_response( field=response_field, response=raw_response, include=response_model_include, exclude=response_model_exclude, by_alias=response_model_by_alias, skip_defaults=response_model_skip_defaults, ) return response_class( content=response_data, status_code=status_code, background=background_tasks, ) return app def get_websocket_app(dependant: Dependant) -> Callable: async def app(websocket: WebSocket) -> None: values, errors, _ = await solve_dependencies( request=websocket, dependant=dependant ) if errors: await websocket.close(code=WS_1008_POLICY_VIOLATION) raise WebSocketRequestValidationError(errors) assert dependant.call is not None, "dependant.call must me a function" await dependant.call(**values) return app class APIWebSocketRoute(routing.WebSocketRoute): def __init__(self, path: str, endpoint: Callable, *, name: str = None) -> None: self.path = path self.endpoint = endpoint self.name = get_name(endpoint) if name is None else name self.dependant = get_dependant(path=path, call=self.endpoint) self.app = websocket_session(get_websocket_app(dependant=self.dependant)) regex = "^" + path + "$" regex = re.sub("{([a-zA-Z_][a-zA-Z0-9_]*)}", r"(?P<\1>[^/]+)", regex) self.path_regex, self.path_format, self.param_convertors = compile_path(path) class APIRoute(routing.Route): def __init__( self, path: str, endpoint: Callable, *, response_model: Type[Any] = None, status_code: int = 200, tags: List[str] = None, dependencies: List[params.Depends] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", responses: Dict[Union[int, str], Dict[str, Any]] = None, deprecated: bool = None, name: str = None, methods: List[str] = None, operation_id: str = None, response_model_include: Set[str] = None, response_model_exclude: Set[str] = set(), response_model_by_alias: bool = True, response_model_skip_defaults: bool = False, include_in_schema: bool = True, response_class: 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( response_class, 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.dependencies = dependencies or [] self.summary = summary self.description = description or inspect.cleandoc(self.endpoint.__doc__ or "") self.response_description = response_description self.responses = responses or {} response_fields = {} for additional_status_code, response in self.responses.items(): assert isinstance(response, dict), "An additional response must be a dict" model = response.get("model") if model: assert lenient_issubclass( model, BaseModel ), "A response model must be a Pydantic model" response_name = f"Response_{additional_status_code}_{self.name}" response_field = Field( name=response_name, type_=model, class_validators=None, default=None, required=False, model_config=BaseConfig, schema=Schema(None), ) response_fields[additional_status_code] = response_field if response_fields: self.response_fields: Dict[Union[int, str], Field] = response_fields else: self.response_fields = {} self.deprecated = deprecated if methods is None: methods = ["GET"] self.methods = methods self.operation_id = operation_id self.response_model_include = response_model_include self.response_model_exclude = response_model_exclude self.response_model_by_alias = response_model_by_alias self.response_model_skip_defaults = response_model_skip_defaults self.include_in_schema = include_in_schema self.response_class = response_class self.path_regex, self.path_format, self.param_convertors = 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=self.path_format, call=self.endpoint) for depends in self.dependencies[::-1]: self.dependant.dependencies.insert( 0, get_parameterless_sub_dependant(depends=depends, path=self.path_format), ) 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, response_class=self.response_class, response_field=self.response_field, response_model_include=self.response_model_include, response_model_exclude=self.response_model_exclude, response_model_by_alias=self.response_model_by_alias, response_model_skip_defaults=self.response_model_skip_defaults, ) ) class APIRouter(routing.Router): def add_api_route( self, path: str, endpoint: Callable, *, response_model: Type[Any] = None, status_code: int = 200, tags: List[str] = None, dependencies: List[params.Depends] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", responses: Dict[Union[int, str], Dict[str, Any]] = None, deprecated: bool = None, methods: List[str] = None, operation_id: str = None, response_model_include: Set[str] = None, response_model_exclude: Set[str] = set(), response_model_by_alias: bool = True, response_model_skip_defaults: bool = False, include_in_schema: bool = True, response_class: Type[Response] = JSONResponse, name: str = None, ) -> None: route = APIRoute( path, endpoint=endpoint, response_model=response_model, status_code=status_code, tags=tags or [], dependencies=dependencies or [], summary=summary, description=description, response_description=response_description, responses=responses or {}, deprecated=deprecated, methods=methods, operation_id=operation_id, response_model_include=response_model_include, response_model_exclude=response_model_exclude, response_model_by_alias=response_model_by_alias, response_model_skip_defaults=response_model_skip_defaults, include_in_schema=include_in_schema, response_class=response_class, name=name, ) self.routes.append(route) def api_route( self, path: str, *, response_model: Type[Any] = None, status_code: int = 200, tags: List[str] = None, dependencies: List[params.Depends] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", responses: Dict[Union[int, str], Dict[str, Any]] = None, deprecated: bool = None, methods: List[str] = None, operation_id: str = None, response_model_include: Set[str] = None, response_model_exclude: Set[str] = set(), response_model_by_alias: bool = True, response_model_skip_defaults: bool = False, include_in_schema: bool = True, response_class: Type[Response] = JSONResponse, name: str = None, ) -> Callable: def decorator(func: Callable) -> Callable: self.add_api_route( path, func, response_model=response_model, status_code=status_code, tags=tags or [], dependencies=dependencies or [], summary=summary, description=description, response_description=response_description, responses=responses or {}, deprecated=deprecated, methods=methods, operation_id=operation_id, response_model_include=response_model_include, response_model_exclude=response_model_exclude, response_model_by_alias=response_model_by_alias, response_model_skip_defaults=response_model_skip_defaults, include_in_schema=include_in_schema, response_class=response_class, name=name, ) return func return decorator def add_api_websocket_route( self, path: str, endpoint: Callable, name: str = None ) -> None: route = APIWebSocketRoute(path, endpoint=endpoint, name=name) self.routes.append(route) def websocket(self, path: str, name: str = None) -> Callable: def decorator(func: Callable) -> Callable: self.add_api_websocket_route(path, func, name=name) return func return decorator def include_router( self, router: "APIRouter", *, prefix: str = "", tags: List[str] = None, dependencies: List[params.Depends] = None, responses: Dict[Union[int, str], Dict[str, Any]] = None, ) -> 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 '/'" if responses is None: responses = {} for route in router.routes: if isinstance(route, APIRoute): combined_responses = {**responses, **route.responses} self.add_api_route( prefix + route.path, route.endpoint, response_model=route.response_model, status_code=route.status_code, tags=(route.tags or []) + (tags or []), dependencies=(dependencies or []) + (route.dependencies or []), summary=route.summary, description=route.description, response_description=route.response_description, responses=combined_responses, deprecated=route.deprecated, methods=route.methods, operation_id=route.operation_id, response_model_include=route.response_model_include, response_model_exclude=route.response_model_exclude, response_model_by_alias=route.response_model_by_alias, response_model_skip_defaults=route.response_model_skip_defaults, include_in_schema=route.include_in_schema, response_class=route.response_class, name=route.name, ) elif isinstance(route, routing.Route): self.add_route( prefix + route.path, route.endpoint, methods=route.methods, include_in_schema=route.include_in_schema, name=route.name, ) elif isinstance(route, APIWebSocketRoute): self.add_api_websocket_route( prefix + route.path, route.endpoint, name=route.name ) elif isinstance(route, routing.WebSocketRoute): self.add_websocket_route( prefix + route.path, route.endpoint, name=route.name ) def get( self, path: str, *, response_model: Type[Any] = None, status_code: int = 200, tags: List[str] = None, dependencies: List[params.Depends] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", responses: Dict[Union[int, str], Dict[str, Any]] = None, deprecated: bool = None, operation_id: str = None, response_model_include: Set[str] = None, response_model_exclude: Set[str] = set(), response_model_by_alias: bool = True, response_model_skip_defaults: bool = False, include_in_schema: bool = True, response_class: Type[Response] = JSONResponse, name: str = None, ) -> Callable: return self.api_route( path=path, response_model=response_model, status_code=status_code, tags=tags or [], dependencies=dependencies or [], summary=summary, description=description, response_description=response_description, responses=responses or {}, deprecated=deprecated, methods=["GET"], operation_id=operation_id, response_model_include=response_model_include, response_model_exclude=response_model_exclude, response_model_by_alias=response_model_by_alias, response_model_skip_defaults=response_model_skip_defaults, include_in_schema=include_in_schema, response_class=response_class, name=name, ) def put( self, path: str, *, response_model: Type[Any] = None, status_code: int = 200, tags: List[str] = None, dependencies: List[params.Depends] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", responses: Dict[Union[int, str], Dict[str, Any]] = None, deprecated: bool = None, operation_id: str = None, response_model_include: Set[str] = None, response_model_exclude: Set[str] = set(), response_model_by_alias: bool = True, response_model_skip_defaults: bool = False, include_in_schema: bool = True, response_class: Type[Response] = JSONResponse, name: str = None, ) -> Callable: return self.api_route( path=path, response_model=response_model, status_code=status_code, tags=tags or [], dependencies=dependencies or [], summary=summary, description=description, response_description=response_description, responses=responses or {}, deprecated=deprecated, methods=["PUT"], operation_id=operation_id, response_model_include=response_model_include, response_model_exclude=response_model_exclude, response_model_by_alias=response_model_by_alias, response_model_skip_defaults=response_model_skip_defaults, include_in_schema=include_in_schema, response_class=response_class, name=name, ) def post( self, path: str, *, response_model: Type[Any] = None, status_code: int = 200, tags: List[str] = None, dependencies: List[params.Depends] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", responses: Dict[Union[int, str], Dict[str, Any]] = None, deprecated: bool = None, operation_id: str = None, response_model_include: Set[str] = None, response_model_exclude: Set[str] = set(), response_model_by_alias: bool = True, response_model_skip_defaults: bool = False, include_in_schema: bool = True, response_class: Type[Response] = JSONResponse, name: str = None, ) -> Callable: return self.api_route( path=path, response_model=response_model, status_code=status_code, tags=tags or [], dependencies=dependencies or [], summary=summary, description=description, response_description=response_description, responses=responses or {}, deprecated=deprecated, methods=["POST"], operation_id=operation_id, response_model_include=response_model_include, response_model_exclude=response_model_exclude, response_model_by_alias=response_model_by_alias, response_model_skip_defaults=response_model_skip_defaults, include_in_schema=include_in_schema, response_class=response_class, name=name, ) def delete( self, path: str, *, response_model: Type[Any] = None, status_code: int = 200, tags: List[str] = None, dependencies: List[params.Depends] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", responses: Dict[Union[int, str], Dict[str, Any]] = None, deprecated: bool = None, operation_id: str = None, response_model_include: Set[str] = None, response_model_exclude: Set[str] = set(), response_model_by_alias: bool = True, response_model_skip_defaults: bool = False, include_in_schema: bool = True, response_class: Type[Response] = JSONResponse, name: str = None, ) -> Callable: return self.api_route( path=path, response_model=response_model, status_code=status_code, tags=tags or [], dependencies=dependencies or [], summary=summary, description=description, response_description=response_description, responses=responses or {}, deprecated=deprecated, methods=["DELETE"], operation_id=operation_id, response_model_include=response_model_include, response_model_exclude=response_model_exclude, response_model_by_alias=response_model_by_alias, response_model_skip_defaults=response_model_skip_defaults, include_in_schema=include_in_schema, response_class=response_class, name=name, ) def options( self, path: str, *, response_model: Type[Any] = None, status_code: int = 200, tags: List[str] = None, dependencies: List[params.Depends] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", responses: Dict[Union[int, str], Dict[str, Any]] = None, deprecated: bool = None, operation_id: str = None, response_model_include: Set[str] = None, response_model_exclude: Set[str] = set(), response_model_by_alias: bool = True, response_model_skip_defaults: bool = False, include_in_schema: bool = True, response_class: Type[Response] = JSONResponse, name: str = None, ) -> Callable: return self.api_route( path=path, response_model=response_model, status_code=status_code, tags=tags or [], dependencies=dependencies or [], summary=summary, description=description, response_description=response_description, responses=responses or {}, deprecated=deprecated, methods=["OPTIONS"], operation_id=operation_id, response_model_include=response_model_include, response_model_exclude=response_model_exclude, response_model_by_alias=response_model_by_alias, response_model_skip_defaults=response_model_skip_defaults, include_in_schema=include_in_schema, response_class=response_class, name=name, ) def head( self, path: str, *, response_model: Type[Any] = None, status_code: int = 200, tags: List[str] = None, dependencies: List[params.Depends] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", responses: Dict[Union[int, str], Dict[str, Any]] = None, deprecated: bool = None, operation_id: str = None, response_model_include: Set[str] = None, response_model_exclude: Set[str] = set(), response_model_by_alias: bool = True, response_model_skip_defaults: bool = False, include_in_schema: bool = True, response_class: Type[Response] = JSONResponse, name: str = None, ) -> Callable: return self.api_route( path=path, response_model=response_model, status_code=status_code, tags=tags or [], dependencies=dependencies or [], summary=summary, description=description, response_description=response_description, responses=responses or {}, deprecated=deprecated, methods=["HEAD"], operation_id=operation_id, response_model_include=response_model_include, response_model_exclude=response_model_exclude, response_model_by_alias=response_model_by_alias, response_model_skip_defaults=response_model_skip_defaults, include_in_schema=include_in_schema, response_class=response_class, name=name, ) def patch( self, path: str, *, response_model: Type[Any] = None, status_code: int = 200, tags: List[str] = None, dependencies: List[params.Depends] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", responses: Dict[Union[int, str], Dict[str, Any]] = None, deprecated: bool = None, operation_id: str = None, response_model_include: Set[str] = None, response_model_exclude: Set[str] = set(), response_model_by_alias: bool = True, response_model_skip_defaults: bool = False, include_in_schema: bool = True, response_class: Type[Response] = JSONResponse, name: str = None, ) -> Callable: return self.api_route( path=path, response_model=response_model, status_code=status_code, tags=tags or [], dependencies=dependencies or [], summary=summary, description=description, response_description=response_description, responses=responses or {}, deprecated=deprecated, methods=["PATCH"], operation_id=operation_id, response_model_include=response_model_include, response_model_exclude=response_model_exclude, response_model_by_alias=response_model_by_alias, response_model_skip_defaults=response_model_skip_defaults, include_in_schema=include_in_schema, response_class=response_class, name=name, ) def trace( self, path: str, *, response_model: Type[Any] = None, status_code: int = 200, tags: List[str] = None, dependencies: List[params.Depends] = None, summary: str = None, description: str = None, response_description: str = "Successful Response", responses: Dict[Union[int, str], Dict[str, Any]] = None, deprecated: bool = None, operation_id: str = None, response_model_include: Set[str] = None, response_model_exclude: Set[str] = set(), response_model_by_alias: bool = True, response_model_skip_defaults: bool = False, include_in_schema: bool = True, response_class: Type[Response] = JSONResponse, name: str = None, ) -> Callable: return self.api_route( path=path, response_model=response_model, status_code=status_code, tags=tags or [], dependencies=dependencies or [], summary=summary, description=description, response_description=response_description, responses=responses or {}, deprecated=deprecated, methods=["TRACE"], operation_id=operation_id, response_model_include=response_model_include, response_model_exclude=response_model_exclude, response_model_by_alias=response_model_by_alias, response_model_skip_defaults=response_model_skip_defaults, include_in_schema=include_in_schema, response_class=response_class, name=name, ) PK}oNaDDfastapi/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) if route.response_fields: responses_from_routes.extend(route.response_fields.values()) 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}oN fastapi/dependencies/__init__.pyPK}oN ufastapi/dependencies/models.pyfrom typing import Callable, List, Sequence from fastapi.security.base import SecurityBase from pydantic.fields import Field 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, websocket_param_name: str = None, background_tasks_param_name: str = None, security_scopes_param_name: str = None, security_scopes: List[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.websocket_param_name = websocket_param_name self.background_tasks_param_name = background_tasks_param_name self.security_scopes = security_scopes self.security_scopes_param_name = security_scopes_param_name self.name = name self.call = call PK}oN$4FFfastapi/dependencies/utils.pyimport asyncio import inspect from copy import deepcopy from typing import ( Any, Callable, Dict, List, Mapping, Optional, Sequence, Tuple, Type, Union, cast, ) from fastapi import params from fastapi.dependencies.models import Dependant, SecurityRequirement from fastapi.security.base import SecurityBase from fastapi.security.oauth2 import OAuth2, SecurityScopes from fastapi.security.open_id_connect_url import OpenIdConnect from fastapi.utils import get_path_param_names from pydantic import BaseConfig, BaseModel, Schema, create_model from pydantic.error_wrappers import ErrorWrapper from pydantic.errors import MissingError from pydantic.fields import Field, Required, Shape from pydantic.schema import get_annotation_from_schema from pydantic.utils import lenient_issubclass from starlette.background import BackgroundTasks from starlette.concurrency import run_in_threadpool from starlette.datastructures import FormData, Headers, QueryParams, UploadFile from starlette.requests import Request from starlette.websockets import WebSocket sequence_shapes = { Shape.LIST, Shape.SET, Shape.TUPLE, Shape.SEQUENCE, Shape.TUPLE_ELLIPS, } sequence_types = (list, set, tuple) sequence_shape_to_type = { Shape.LIST: list, Shape.SET: set, Shape.TUPLE: tuple, Shape.SEQUENCE: list, Shape.TUPLE_ELLIPS: list, } def get_param_sub_dependant( *, param: inspect.Parameter, path: str, security_scopes: List[str] = None ) -> Dependant: depends: params.Depends = param.default if depends.dependency: dependency = depends.dependency else: dependency = param.annotation return get_sub_dependant( depends=depends, dependency=dependency, path=path, name=param.name, security_scopes=security_scopes, ) def get_parameterless_sub_dependant(*, depends: params.Depends, path: str) -> Dependant: assert callable( depends.dependency ), "A parameter-less dependency must have a callable dependency" return get_sub_dependant(depends=depends, dependency=depends.dependency, path=path) def get_sub_dependant( *, depends: params.Depends, dependency: Callable, path: str, name: str = None, security_scopes: List[str] = None, ) -> Dependant: security_requirement = None security_scopes = security_scopes or [] if isinstance(depends, params.Security): dependency_scopes = depends.scopes security_scopes.extend(dependency_scopes) if isinstance(dependency, SecurityBase): use_scopes: List[str] = [] if isinstance(dependency, (OAuth2, OpenIdConnect)): use_scopes = security_scopes security_requirement = SecurityRequirement( security_scheme=dependency, scopes=use_scopes ) sub_dependant = get_dependant( path=path, call=dependency, name=name, security_scopes=security_scopes ) if security_requirement: sub_dependant.security_requirements.append(security_requirement) sub_dependant.security_scopes = security_scopes 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: 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 is_scalar_field(field: Field) -> bool: return ( field.shape == Shape.SINGLETON and not lenient_issubclass(field.type_, BaseModel) and not isinstance(field.schema, params.Body) ) def is_scalar_sequence_field(field: Field) -> bool: if field.shape in sequence_shapes and not lenient_issubclass( field.type_, BaseModel ): if field.sub_fields is not None: for sub_field in field.sub_fields: if not is_scalar_field(sub_field): return False return True return False def get_dependant( *, path: str, call: Callable, name: str = None, security_scopes: List[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, param in signature_params.items(): if isinstance(param.default, params.Depends): sub_dependant = get_param_sub_dependant( param=param, path=path, security_scopes=security_scopes ) dependant.dependencies.append(sub_dependant) for param_name, param in signature_params.items(): if isinstance(param.default, params.Depends): continue if add_non_field_param_to_dependency(param=param, dependant=dependant): continue param_field = get_param_field(param=param, default_schema=params.Query) if param_name in path_param_names: assert param.default == param.empty or isinstance( param.default, params.Path ), "Path params must have no defaults or use Path(...)" assert is_scalar_field( field=param_field ), f"Path params must be of one of the supported types" param_field = get_param_field( param=param, default_schema=params.Path, force_type=params.ParamTypes.path, ) add_param_to_fields(field=param_field, dependant=dependant) elif is_scalar_field(field=param_field): add_param_to_fields(field=param_field, dependant=dependant) elif isinstance( param.default, (params.Query, params.Header) ) and is_scalar_sequence_field(param_field): add_param_to_fields(field=param_field, dependant=dependant) else: assert isinstance( param_field.schema, params.Body ), f"Param: {param_field.name} can only be a request body, using Body(...)" dependant.body_params.append(param_field) return dependant def add_non_field_param_to_dependency( *, param: inspect.Parameter, dependant: Dependant ) -> Optional[bool]: if lenient_issubclass(param.annotation, Request): dependant.request_param_name = param.name return True elif lenient_issubclass(param.annotation, WebSocket): dependant.websocket_param_name = param.name return True elif lenient_issubclass(param.annotation, BackgroundTasks): dependant.background_tasks_param_name = param.name return True elif lenient_issubclass(param.annotation, SecurityScopes): dependant.security_scopes_param_name = param.name return True return None def get_param_field( *, param: inspect.Parameter, default_schema: Type[params.Param] = params.Param, force_type: params.ParamTypes = None, ) -> Field: default_value = Required had_schema = False if not param.default == param.empty: default_value = param.default if isinstance(default_value, Schema): had_schema = True schema = default_value default_value = schema.default if isinstance(schema, params.Param) and 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 not had_schema and not is_scalar_field(field=field): field.schema = params.Body(schema.default) return field def add_param_to_fields(*, field: Field, dependant: Dependant) -> None: field.schema = cast(params.Param, field.schema) if field.schema.in_ == params.ParamTypes.path: dependant.path_params.append(field) elif field.schema.in_ == params.ParamTypes.query: dependant.query_params.append(field) elif field.schema.in_ == params.ParamTypes.header: dependant.header_params.append(field) else: assert ( field.schema.in_ == params.ParamTypes.cookie ), f"non-body parameters must be in path, query, header or cookie: {field.name}" dependant.cookie_params.append(field) def is_coroutine_callable(call: Callable) -> bool: if inspect.isfunction(call): return asyncio.iscoroutinefunction(call) if inspect.isclass(call): return False call = getattr(call, "__call__", None) return asyncio.iscoroutinefunction(call) async def solve_dependencies( *, request: Union[Request, WebSocket], dependant: Dependant, body: Dict[str, Any] = None, background_tasks: BackgroundTasks = None, ) -> Tuple[Dict[str, Any], List[ErrorWrapper], Optional[BackgroundTasks]]: values: Dict[str, Any] = {} errors: List[ErrorWrapper] = [] for sub_dependant in dependant.dependencies: sub_values, sub_errors, background_tasks = await solve_dependencies( request=request, dependant=sub_dependant, body=body, background_tasks=background_tasks, ) if sub_errors: errors.extend(sub_errors) continue 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) if sub_dependant.name is not None: 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 and isinstance(request, Request): values[dependant.request_param_name] = request elif dependant.websocket_param_name and isinstance(request, WebSocket): values[dependant.websocket_param_name] = request if dependant.background_tasks_param_name: if background_tasks is None: background_tasks = BackgroundTasks() values[dependant.background_tasks_param_name] = background_tasks if dependant.security_scopes_param_name: values[dependant.security_scopes_param_name] = SecurityScopes( scopes=dependant.security_scopes ) return values, errors, background_tasks def request_params_to_args( required_params: Sequence[Field], received_params: Union[Mapping[str, Any], QueryParams, Headers], ) -> Tuple[Dict[str, Any], List[ErrorWrapper]]: values = {} errors = [] for field in required_params: if field.shape in sequence_shapes and isinstance( received_params, (QueryParams, Headers) ): value = received_params.getlist(field.alias) or field.default else: value = received_params.get(field.alias) schema: params.Param = field.schema assert isinstance(schema, params.Param), "Params must be subclasses of Param" if value is None: if field.required: errors.append( ErrorWrapper( MissingError(), loc=(schema.in_.value, field.alias), config=BaseConfig, ) ) else: values[field.name] = deepcopy(field.default) continue 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: if field.shape in sequence_shapes and isinstance(received_body, FormData): value = received_body.getlist(field.alias) else: value = received_body.get(field.alias) if ( value is None or (isinstance(field.schema, params.Form) and value == "") or ( isinstance(field.schema, params.Form) and field.shape in sequence_shapes and len(value) == 0 ) ): if field.required: errors.append( ErrorWrapper( MissingError(), loc=("body", field.alias), config=BaseConfig ) ) else: values[field.name] = deepcopy(field.default) continue if ( isinstance(field.schema, params.File) and lenient_issubclass(field.type_, bytes) and isinstance(value, UploadFile) ): value = await value.read() elif ( field.shape in sequence_shapes and isinstance(field.schema, params.File) and lenient_issubclass(field.type_, bytes) and isinstance(value, sequence_types) ): awaitables = [sub_value.read() for sub_value in value] contents = await asyncio.gather(*awaitables) value = sequence_shape_to_type[field.shape](contents) 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_schema_compatible_field(*, field: Field) -> Field: out_field = field if lenient_issubclass(field.type_, UploadFile): use_type: type = bytes if field.shape in sequence_shapes: use_type = List[bytes] out_field = Field( name=field.name, type_=use_type, class_validators=field.class_validators, model_config=field.model_config, default=field.default, required=field.required, alias=field.alias, schema=field.schema, ) return out_field def get_body_field(*, dependant: Dependant, name: str) -> Optional[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 get_schema_compatible_field(field=first_param) model_name = "Body_" + name BodyModel = create_model(model_name) for f in flat_dependant.body_params: BodyModel.__fields__[f.name] = get_schema_compatible_field(field=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 PK}oNfastapi/openapi/__init__.pyPK}oNmaafastapi/openapi/constants.pyMETHODS_WITH_BODY = set(("POST", "PUT", "DELETE", "PATCH")) REF_PREFIX = "#/components/schemas/" PK}oN jfastapi/openapi/docs.pyfrom typing import Optional from starlette.responses import HTMLResponse def get_swagger_ui_html( *, openapi_url: str, title: str, swagger_js_url: str = "https://cdn.jsdelivr.net/npm/swagger-ui-dist@3/swagger-ui-bundle.js", swagger_css_url: str = "https://cdn.jsdelivr.net/npm/swagger-ui-dist@3/swagger-ui.css", swagger_favicon_url: str = "https://fastapi.tiangolo.com/img/favicon.png", oauth2_redirect_url: Optional[str] = None, ) -> HTMLResponse: html = f""" {title}
""" return HTMLResponse(html) def get_redoc_html( *, openapi_url: str, title: str, redoc_js_url: str = "https://cdn.jsdelivr.net/npm/redoc@next/bundles/redoc.standalone.js", redoc_favicon_url: str = "https://fastapi.tiangolo.com/img/favicon.png", ) -> HTMLResponse: html = f""" {title} """ return HTMLResponse(html) def get_swagger_ui_oauth2_redirect_html() -> HTMLResponse: html = """ """ return HTMLResponse(content=html) PK}oN#5))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 logger = logging.getLogger("fastapi") try: import email_validator assert email_validator # make autoflake ignore the unused import from pydantic.types import EmailStr # type: ignore except ImportError: # pragma: no cover logger.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[Dict[str, Any], bool]] = 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[SchemaBase, bool]] = 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[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, OAuth2, OpenIdConnect, HTTPBearer] class Components(BaseModel): schemas: Optional[Dict[str, Union[Schema, Reference]]] = None responses: Optional[Dict[str, Union[Response, Reference]]] = None parameters: Optional[Dict[str, Union[Parameter, Reference]]] = None examples: Optional[Dict[str, Union[Example, Reference]]] = None requestBodies: Optional[Dict[str, Union[RequestBody, Reference]]] = None headers: Optional[Dict[str, Union[Header, Reference]]] = None securitySchemes: Optional[Dict[str, Union[SecurityScheme, Reference]]] = None links: Optional[Dict[str, Union[Link, Reference]]] = None callbacks: Optional[Dict[str, Union[Dict[str, PathItem], Reference]]] = 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 PK}oN*,_*_*fastapi/openapi/utils.pyfrom typing import Any, Dict, List, Optional, Sequence, Tuple, Type, cast 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 field_schema, get_model_name_map from pydantic.utils import lenient_issubclass from starlette.responses import JSONResponse from starlette.routing import BaseRoute 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 ) body_field.schema = cast(Body, body_field.schema) request_media_type = body_field.schema.media_type 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_format 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() 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" if route.include_in_schema: 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 if "ValidationError" not in definitions: definitions["ValidationError"] = validation_error_definition definitions[ "HTTPValidationError" ] = validation_error_response_definition if route.responses: for (additional_status_code, response) in route.responses.items(): assert isinstance( response, dict ), "An additional response must be a dict" field = route.response_fields.get(additional_status_code) if field: response_schema, _ = field_schema( field, model_name_map=model_name_map, ref_prefix=REF_PREFIX ) response.setdefault("content", {}).setdefault( "application/json", {} )["schema"] = response_schema response.setdefault("description", "Additional Response") operation.setdefault("responses", {})[ str(additional_status_code) ] = response status_code = str(route.status_code) response_schema = {"type": "string"} if lenient_issubclass(route.response_class, 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 = {} operation.setdefault("responses", {}).setdefault(status_code, {})[ "description" ] = route.response_description operation.setdefault("responses", {}).setdefault( status_code, {} ).setdefault("content", {}).setdefault(route.response_class.media_type, {})[ "schema" ] = response_schema 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], openapi_prefix: str = "" ) -> 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(openapi_prefix + route.path_format, {}).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) PK}oN߸^^fastapi/security/__init__.pyfrom .api_key import APIKeyCookie, APIKeyHeader, APIKeyQuery from .http import ( HTTPAuthorizationCredentials, HTTPBasic, HTTPBasicCredentials, HTTPBearer, HTTPDigest, ) from .oauth2 import ( OAuth2, OAuth2PasswordBearer, OAuth2PasswordRequestForm, SecurityScopes, ) from .open_id_connect_url import OpenIdConnect PK}oN$+8 8 fastapi/security/api_key.pyfrom typing import Optional from fastapi.openapi.models import APIKey, APIKeyIn from fastapi.security.base import SecurityBase from starlette.exceptions import HTTPException from starlette.requests import Request from starlette.status import HTTP_403_FORBIDDEN class APIKeyBase(SecurityBase): pass class APIKeyQuery(APIKeyBase): def __init__(self, *, name: str, scheme_name: str = None, auto_error: bool = True): self.model: APIKey = APIKey(**{"in": APIKeyIn.query}, name=name) self.scheme_name = scheme_name or self.__class__.__name__ self.auto_error = auto_error async def __call__(self, request: Request) -> Optional[str]: api_key: str = request.query_params.get(self.model.name) if not api_key: if self.auto_error: raise HTTPException( status_code=HTTP_403_FORBIDDEN, detail="Not authenticated" ) else: return None return api_key class APIKeyHeader(APIKeyBase): def __init__(self, *, name: str, scheme_name: str = None, auto_error: bool = True): self.model: APIKey = APIKey(**{"in": APIKeyIn.header}, name=name) self.scheme_name = scheme_name or self.__class__.__name__ self.auto_error = auto_error async def __call__(self, request: Request) -> Optional[str]: api_key: str = request.headers.get(self.model.name) if not api_key: if self.auto_error: raise HTTPException( status_code=HTTP_403_FORBIDDEN, detail="Not authenticated" ) else: return None return api_key class APIKeyCookie(APIKeyBase): def __init__(self, *, name: str, scheme_name: str = None, auto_error: bool = True): self.model: APIKey = APIKey(**{"in": APIKeyIn.cookie}, name=name) self.scheme_name = scheme_name or self.__class__.__name__ self.auto_error = auto_error async def __call__(self, request: Request) -> Optional[str]: api_key: str = request.cookies.get(self.model.name) if not api_key: if self.auto_error: raise HTTPException( status_code=HTTP_403_FORBIDDEN, detail="Not authenticated" ) else: return None return api_key PK}oNr(ԍfastapi/security/base.pyfrom fastapi.openapi.models import SecurityBase as SecurityBaseModel class SecurityBase: model: SecurityBaseModel scheme_name: str PK}oNx3fastapi/security/http.pyimport binascii from base64 import b64decode from typing import Optional from fastapi.exceptions import HTTPException from fastapi.openapi.models import ( HTTPBase as HTTPBaseModel, HTTPBearer as HTTPBearerModel, ) from fastapi.security.base import SecurityBase from fastapi.security.utils import get_authorization_scheme_param from pydantic import BaseModel from starlette.requests import Request from starlette.status import HTTP_401_UNAUTHORIZED, HTTP_403_FORBIDDEN class HTTPBasicCredentials(BaseModel): username: str password: str class HTTPAuthorizationCredentials(BaseModel): scheme: str credentials: str class HTTPBase(SecurityBase): def __init__( self, *, scheme: str, scheme_name: str = None, auto_error: bool = True ): self.model = HTTPBaseModel(scheme=scheme) self.scheme_name = scheme_name or self.__class__.__name__ self.auto_error = auto_error async def __call__( self, request: Request ) -> Optional[HTTPAuthorizationCredentials]: authorization: str = request.headers.get("Authorization") scheme, credentials = get_authorization_scheme_param(authorization) if not (authorization and scheme and credentials): if self.auto_error: raise HTTPException( status_code=HTTP_403_FORBIDDEN, detail="Not authenticated" ) else: return None return HTTPAuthorizationCredentials(scheme=scheme, credentials=credentials) class HTTPBasic(HTTPBase): def __init__( self, *, scheme_name: str = None, realm: str = None, auto_error: bool = True ): self.model = HTTPBaseModel(scheme="basic") self.scheme_name = scheme_name or self.__class__.__name__ self.realm = realm self.auto_error = auto_error async def __call__(self, request: Request) -> Optional[HTTPBasicCredentials]: authorization: str = request.headers.get("Authorization") scheme, param = get_authorization_scheme_param(authorization) if self.realm: unauthorized_headers = {"WWW-Authenticate": f'Basic realm="{self.realm}"'} else: unauthorized_headers = {"WWW-Authenticate": "Basic"} invalid_user_credentials_exc = HTTPException( status_code=HTTP_401_UNAUTHORIZED, detail="Invalid authentication credentials", headers=unauthorized_headers, ) if not authorization or scheme.lower() != "basic": if self.auto_error: raise HTTPException( status_code=HTTP_401_UNAUTHORIZED, detail="Not authenticated", headers=unauthorized_headers, ) else: return None try: data = b64decode(param).decode("ascii") except (ValueError, UnicodeDecodeError, binascii.Error): raise invalid_user_credentials_exc username, separator, password = data.partition(":") if not (separator): raise invalid_user_credentials_exc return HTTPBasicCredentials(username=username, password=password) class HTTPBearer(HTTPBase): def __init__( self, *, bearerFormat: str = None, scheme_name: str = None, auto_error: bool = True, ): self.model = HTTPBearerModel(bearerFormat=bearerFormat) self.scheme_name = scheme_name or self.__class__.__name__ self.auto_error = auto_error async def __call__( self, request: Request ) -> Optional[HTTPAuthorizationCredentials]: authorization: str = request.headers.get("Authorization") scheme, credentials = get_authorization_scheme_param(authorization) if not (authorization and scheme and credentials): if self.auto_error: raise HTTPException( status_code=HTTP_403_FORBIDDEN, detail="Not authenticated" ) else: return None if scheme.lower() != "bearer": raise HTTPException( status_code=HTTP_403_FORBIDDEN, detail="Invalid authentication credentials", ) return HTTPAuthorizationCredentials(scheme=scheme, credentials=credentials) class HTTPDigest(HTTPBase): def __init__(self, *, scheme_name: str = None, auto_error: bool = True): self.model = HTTPBaseModel(scheme="digest") self.scheme_name = scheme_name or self.__class__.__name__ self.auto_error = auto_error async def __call__( self, request: Request ) -> Optional[HTTPAuthorizationCredentials]: authorization: str = request.headers.get("Authorization") scheme, credentials = get_authorization_scheme_param(authorization) if not (authorization and scheme and credentials): if self.auto_error: raise HTTPException( status_code=HTTP_403_FORBIDDEN, detail="Not authenticated" ) else: return None if scheme.lower() != "digest": raise HTTPException( status_code=HTTP_403_FORBIDDEN, detail="Invalid authentication credentials", ) return HTTPAuthorizationCredentials(scheme=scheme, credentials=credentials) PK}oN%fastapi/security/oauth2.pyfrom typing import List, Optional from fastapi.exceptions import HTTPException from fastapi.openapi.models import OAuth2 as OAuth2Model, OAuthFlows as OAuthFlowsModel from fastapi.params import Form from fastapi.security.base import SecurityBase from fastapi.security.utils import get_authorization_scheme_param from starlette.requests import Request from starlette.status import HTTP_401_UNAUTHORIZED, HTTP_403_FORBIDDEN class OAuth2PasswordRequestForm: """ This is a dependency class, use it like: @app.post("/login") def login(form_data: Oauth2PasswordRequestForm = Depends()): data = form_data.parse() print(data.username) print(data.password) for scope in data.scopes: print(scope) if data.client_id: print(data.client_id) if data.client_secret: print(data.client_secret) return data It creates the following Form request parameters in your endpoint: grant_type: the OAuth2 spec says it is required and MUST be the fixed string "password". Nevertheless, this dependency class is permissive and allows not passing it. If you want to enforce it, use instead the OAuth2PasswordRequestFormStrict dependency. username: username string. The OAuth2 spec requires the exact field name "username". password: password string. The OAuth2 spec requires the exact field name "password". scope: Optional string. Several scopes (each one a string) separated by spaces. E.g. "items:read items:write users:read profile openid" client_id: optional string. OAuth2 recommends sending the client_id and client_secret (if any) using HTTP Basic auth, as: client_id:client_secret client_secret: optional string. OAuth2 recommends sending the client_id and client_secret (if any) using HTTP Basic auth, as: client_id:client_secret """ def __init__( self, grant_type: str = Form(None, regex="password"), username: str = Form(...), password: str = Form(...), scope: str = Form(""), client_id: Optional[str] = Form(None), client_secret: Optional[str] = Form(None), ): self.grant_type = grant_type self.username = username self.password = password self.scopes = scope.split() self.client_id = client_id self.client_secret = client_secret class OAuth2PasswordRequestFormStrict(OAuth2PasswordRequestForm): """ This is a dependency class, use it like: @app.post("/login") def login(form_data: Oauth2PasswordRequestFormStrict = Depends()): data = form_data.parse() print(data.username) print(data.password) for scope in data.scopes: print(scope) if data.client_id: print(data.client_id) if data.client_secret: print(data.client_secret) return data It creates the following Form request parameters in your endpoint: grant_type: the OAuth2 spec says it is required and MUST be the fixed string "password". This dependency is strict about it. If you want to be permissive, use instead the OAuth2PasswordRequestFormStrict dependency class. username: username string. The OAuth2 spec requires the exact field name "username". password: password string. The OAuth2 spec requires the exact field name "password". scope: Optional string. Several scopes (each one a string) separated by spaces. E.g. "items:read items:write users:read profile openid" client_id: optional string. OAuth2 recommends sending the client_id and client_secret (if any) using HTTP Basic auth, as: client_id:client_secret client_secret: optional string. OAuth2 recommends sending the client_id and client_secret (if any) using HTTP Basic auth, as: client_id:client_secret """ def __init__( self, grant_type: str = Form(..., regex="password"), username: str = Form(...), password: str = Form(...), scope: str = Form(""), client_id: Optional[str] = Form(None), client_secret: Optional[str] = Form(None), ): super().__init__( grant_type=grant_type, username=username, password=password, scope=scope, client_id=client_id, client_secret=client_secret, ) class OAuth2(SecurityBase): def __init__( self, *, flows: OAuthFlowsModel = OAuthFlowsModel(), scheme_name: str = None, auto_error: bool = True ): self.model = OAuth2Model(flows=flows) self.scheme_name = scheme_name or self.__class__.__name__ self.auto_error = auto_error async def __call__(self, request: Request) -> Optional[str]: authorization: str = request.headers.get("Authorization") if not authorization: if self.auto_error: raise HTTPException( status_code=HTTP_403_FORBIDDEN, detail="Not authenticated" ) else: return None return authorization class OAuth2PasswordBearer(OAuth2): def __init__( self, tokenUrl: str, scheme_name: str = None, scopes: dict = None, auto_error: bool = True, ): if not scopes: scopes = {} flows = OAuthFlowsModel(password={"tokenUrl": tokenUrl, "scopes": scopes}) super().__init__(flows=flows, scheme_name=scheme_name, auto_error=auto_error) async def __call__(self, request: Request) -> Optional[str]: authorization: str = request.headers.get("Authorization") scheme, param = get_authorization_scheme_param(authorization) if not authorization or scheme.lower() != "bearer": if self.auto_error: raise HTTPException( status_code=HTTP_401_UNAUTHORIZED, detail="Not authenticated", headers={"WWW-Authenticate": "Bearer"}, ) else: return None return param class SecurityScopes: def __init__(self, scopes: List[str] = None): self.scopes = scopes or [] self.scope_str = " ".join(self.scopes) PK}oN'fastapi/security/open_id_connect_url.pyfrom typing import Optional from fastapi.openapi.models import OpenIdConnect as OpenIdConnectModel from fastapi.security.base import SecurityBase from starlette.exceptions import HTTPException from starlette.requests import Request from starlette.status import HTTP_403_FORBIDDEN class OpenIdConnect(SecurityBase): def __init__( self, *, openIdConnectUrl: str, scheme_name: str = None, auto_error: bool = True ): self.model = OpenIdConnectModel(openIdConnectUrl=openIdConnectUrl) self.scheme_name = scheme_name or self.__class__.__name__ self.auto_error = auto_error async def __call__(self, request: Request) -> Optional[str]: authorization: str = request.headers.get("Authorization") if not authorization: if self.auto_error: raise HTTPException( status_code=HTTP_403_FORBIDDEN, detail="Not authenticated" ) else: return None return authorization PK}oN  fastapi/security/utils.pyfrom typing import Tuple def get_authorization_scheme_param(authorization_header_value: str) -> Tuple[str, str]: if not authorization_header_value: return "", "" scheme, _, param = authorization_header_value.partition(" ") return scheme, param PK}oN{>> fastapi-0.27.0.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!HPOfastapi-0.27.0.dist-info/WHEEL HM K-*ϳR03rOK-J,/RH,szd&Y)r$[)T&UrPK!H70C!fastapi-0.27.0.dist-info/METADATA\krHS;%Z~FG~-my,ydYxfOt\cs=~UxPH31"DUVV>JkY__eU싁c/ƾ.L9oq8oO8,/{b&S|江'HE$9/T^(b / (5Q W /帔ЕGQ:ұFO*6Qe,%7.Be~^lWr&2OˍdGyT&r]I&"6ً+?RWiq+QقdByLՊ(Y& X!orc*h#0J &z>Tx;mjWIy6H4r#^Ze>+ רJz9`*h"G VdP߽MOW1*pD7TѐQt|/ /sa5I;1PΑ#ā/vq9:PD'>y[4 L`HSQY:-@эX[7?|R,w[w 4t_.Sk/Sb*E0 7㺮t >ݿCXZ:js4ʙ'\󒙗Z/b0P-r'n˝lLîcD6Hh1ÜOAQ ¼Ia@:sslץ/0er.G2AZ5_0К}_usy[bjJZ<퉏ՆNXYL]!`Hu-rI"Qu&!< MJ,u4(- 1#Qܥ]ό)gd<4; /--b+2m 4מH*/ten4ʑ<өLE gN6eQ(C=@0s|6+*`81k}XE!y0:l0H ΐҼV/DeGوqƘ-Fa,b¼l]:*r>GPVH3I^}ER? c9V]6qeA[qE44|k۬b4'zwrz|?ŋt2 '4Z+|ryZIfdBH.L9)iq5GC/N߈39`U:; أơjg&~TYW^㟚~V+fǹsGfuNy?c `, ( 9>Ru%^Un gT-4R{::TWО9>i\~H'rtվ:G?#^+tr5w*3(xpY=2lajHO༊fbAH'%P$&8[艓e%FAa!lMˉPS~w0 KL{slH%DjBƏTx޽{{Wǭ6a/hy-8E3~L2t&СB3|[❄P )GYi,r4O*q]aYʹ2ʠ,p«gTf`!$`48J8p8D:!2Bѳ揧D4]N =x'e޻kC68WQ\^{!W9XUVl̮?W=awZ-Cfϳ/><}Gn/|AY1}x|Pe7*02*&Z88.)-oum'c~y[:X.M;!ͣd-ws_ۣѽ-`kO|2wf@'|_YR,5"y٥tSSӜ#%a\n8ZJ9 [Z WŰǣ 끆myit)(at~9~]6%%@'`x}('>KUG Ŏ!nU rw3|Zj)=~A}FGޕ ee`lb>m$JHcjJ 7Zk[ ;)n) v}/z-gÇFM`HWL eXUy 1-Y!X-hH -bSzȖL)a@LFVl5Fʬirv KAaN;;;x=|H`s)Wi׭_e=F(It6k^ti=ki_G6#Bap;H$ڜ4V\ ՝ 1[38xդ!ʩ.b=.o#^(nHi/XΫy/VL'T1ϛZ(-ŐWfAQw5 Ryn(.X S}`E' IgbݠFD~[*EY]5eLHu6wi=W&?|lXbu\ Ͷc}qᚴ>n%\ -e6 ,˼M#o;9s}=H/V(J2~(ū|U$s1`ΰbSY1)*E#lU@*4CR@1[Y`RI5BG eԚ!Nt$hWY;B*xL-4,FYye|MY|+g$9=Y8뼉֕TИF2 uC@[ϰBW|tp_3wlPZ䵵m֙6 `D~/[❁uH.#d:lQxC@%V(HmLy@5P\2Ko!pF>i kr^dγ/2( ٱݒ O>i6Lm x=hl^[lJ|@ z:rU<{Ͳ6FM(};PxƉ=k`Yk76.jn< Y?Չ8tn7MO|qDTO"d I8g/IyChUo /l1k58}: c&d`E,:RnH`*i^)dj,(T|C C^d4 u5dh̳SՖk4>j5.oմd&WkG,$y{+fNB[9Z"+Ml~*3?#jqg2h[e8x€؆-IG='ˤ:rTnE{A!1 bhF`欗^B|W|_M$M//_`S(\\PN,]mEQf0HJd(`bSʞaH?3HiMnWkRa'O>ƺ֒ly|0 ^{16nIPPc7$^p͔#NSߌ&m,dL0&ia1OQF[/,.H\~C*Gau5qNb/XA`f:pcC] {v(~Q1LI2pL!#-66DqhQBŜ{fHkuR36\TE4Ixd*C˃5qhG#*sZWnƫRao֣@`[vkQD놲Ai&OIiLN>q؍Pڜtp4V+zXE ^mHS3>W-T`gAO atNLO0MmJ0dLěxq.kW˗~Szh`<G|hmVuXx BgP#`v˛)Uzo( t_&2_쩻E"pIcܦJ~Y:*!?ZzuevSHyHѹ!\JR%Lݳ,&i8h^P7|`hMna153pk =yڂUM ncBزmI=l$Y-bôگ#̧F3ݽr6=09v yZtc]AJXPK!HN%h fastapi-0.27.0.dist-info/RECORDIH}b@@A7PT7uNgso"Ƥ1l݌ UBчujӲ${{i^*R:dsͱN io"bzP< h`-.QKaL<]/ҘQ68%BmM 5&U{km 9P!LtyG'.378,WL9I}(F1@NG4BBX"4]'׸7wj L=`*+t]u {KgR/a(/#KCCp~fo2B#<0lzu[uȥȊijWQ[\[qۮW'Mf|RƦNm9j C!DRxGSX=cւW /nl5Zzߌ4|/[VWܢ"\`X{2jE6MMȍ"W,(yIoiD=[nПfMDK̖ *rwC=&s,"L=3B0 Å>cٴQ' Nƒ9&'! WvX$&Cf5=Rs5ApNkR苡wśE<7$)_!MFXtbDUHa;خcEOqLHnY%s1nFP=Y.{1oú|d,\6..eY)iwuK]&qr3c3JJw;d.MJ 0 }aDaS(`[5 tו0=U[LY?Y-{M)'Ow`7+ ?ܖ3jȮ ^~ :3ZRm$|R"p ])Ki[M[B PK}oN)>fastapi/__init__.pyPK}oNyUUfastapi/applications.pyPK}oN Xfastapi/datastructures.pyPK}oN Ŀ6Zfastapi/encoders.pyPK}oNGGifastapi/exception_handlers.pyPK}oNlfastapi/exceptions.pyPK}oNhSSnfastapi/param_functions.pyPK}oNVPC+fastapi/params.pyPK}oNfastapi/py.typedPK}oNYL7vvfastapi/routing.pyPK}oNaDD;fastapi/utils.pyPK}oN fastapi/dependencies/__init__.pyPK}oN ufastapi/dependencies/models.pyPK}oN$4FF%fastapi/dependencies/utils.pyPK}oNlfastapi/openapi/__init__.pyPK}oNmaamfastapi/openapi/constants.pyPK}oN jmfastapi/openapi/docs.pyPK}oN#5))fastapi/openapi/models.pyPK}oN*,_*_*fastapi/openapi/utils.pyPK}oN߸^^Ffastapi/security/__init__.pyPK}oN$+8 8 fastapi/security/api_key.pyPK}oNr(ԍOfastapi/security/base.pyPK}oNx3fastapi/security/http.pyPK}oN%dfastapi/security/oauth2.pyPK}oN'fastapi/security/open_id_connect_url.pyPK}oN  fastapi/security/utils.pyPK}oN{>> 0fastapi-0.27.0.dist-info/LICENSEPK!HPOfastapi-0.27.0.dist-info/WHEELPK!H70C!8fastapi-0.27.0.dist-info/METADATAPK!HN%h 2fastapi-0.27.0.dist-info/RECORDPKb7