PK Oнfastapi/__init__.py"""FastAPI framework, high performance, easy to learn, fast to code, ready for production""" __version__ = "0.38.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 O'g[X[Xfastapi/applications.pyfrom typing import Any, Callable, Dict, List, Optional, Sequence, Type, Union from fastapi import routing from fastapi.encoders import DictIntStrAny, SetIntStr 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, dependency_overrides_provider=self ) 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.dependency_overrides: Dict[Callable, Callable] = {} 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: Sequence[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: Union[SetIntStr, DictIntStrAny] = None, response_model_exclude: Union[SetIntStr, DictIntStrAny] = 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, 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: Sequence[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: Union[SetIntStr, DictIntStrAny] = None, response_model_exclude: Union[SetIntStr, DictIntStrAny] = 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, 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: Sequence[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: Sequence[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: Union[SetIntStr, DictIntStrAny] = None, response_model_exclude: Union[SetIntStr, DictIntStrAny] = 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, 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: Sequence[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: Union[SetIntStr, DictIntStrAny] = None, response_model_exclude: Union[SetIntStr, DictIntStrAny] = 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, 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: Sequence[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: Union[SetIntStr, DictIntStrAny] = None, response_model_exclude: Union[SetIntStr, DictIntStrAny] = 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, 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: Sequence[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: Union[SetIntStr, DictIntStrAny] = None, response_model_exclude: Union[SetIntStr, DictIntStrAny] = 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, 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: Sequence[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: Union[SetIntStr, DictIntStrAny] = None, response_model_exclude: Union[SetIntStr, DictIntStrAny] = 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, 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: Sequence[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: Union[SetIntStr, DictIntStrAny] = None, response_model_exclude: Union[SetIntStr, DictIntStrAny] = 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, 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: Sequence[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: Union[SetIntStr, DictIntStrAny] = None, response_model_exclude: Union[SetIntStr, DictIntStrAny] = 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, 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: Sequence[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: Union[SetIntStr, DictIntStrAny] = None, response_model_exclude: Union[SetIntStr, DictIntStrAny] = 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, 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 Ofastapi/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 O`v((fastapi/encoders.pyfrom enum import Enum from types import GeneratorType from typing import Any, Dict, List, Set, Union from pydantic import BaseModel from pydantic.json import ENCODERS_BY_TYPE SetIntStr = Set[Union[int, str]] DictIntStrAny = Dict[Union[int, str], Any] def jsonable_encoder( obj: Any, include: Union[SetIntStr, DictIntStrAny] = None, exclude: Union[SetIntStr, DictIntStrAny] = 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 OGGfastapi/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 OcW::fastapi/exceptions.pyfrom typing import Any, Sequence from pydantic import ValidationError from pydantic.error_wrappers import ErrorList from requests import Request from starlette.exceptions import HTTPException as StarletteHTTPException from starlette.websockets import WebSocket class HTTPException(StarletteHTTPException): def __init__( self, status_code: int, detail: Any = None, headers: dict = None ) -> None: super().__init__(status_code=status_code, detail=detail) self.headers = headers class RequestValidationError(ValidationError): def __init__(self, errors: Sequence[ErrorList]) -> None: super().__init__(errors, Request) class WebSocketRequestValidationError(ValidationError): def __init__(self, errors: Sequence[ErrorList]) -> None: super().__init__(errors, WebSocket) PK O;Hfastapi/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( # noqa: N802 dependency: Callable = None, *, use_cache: bool = True ) -> Any: return params.Depends(dependency=dependency, use_cache=use_cache) def Security( # noqa: N802 dependency: Callable = None, *, scopes: Sequence[str] = None, use_cache: bool = True ) -> Any: return params.Security(dependency=dependency, scopes=scopes, use_cache=use_cache) PK O4$y//fastapi/params.pyfrom enum import Enum from typing import Any, Callable, Sequence from pydantic import Schema class ParamTypes(Enum): query = "query" header = "header" path = "path" cookie = "cookie" class Param(Schema): in_: ParamTypes def __init__( self, default: Any, *, 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, *, use_cache: bool = True): self.dependency = dependency self.use_cache = use_cache class Security(Depends): def __init__( self, dependency: Callable = None, *, scopes: Sequence[str] = None, use_cache: bool = True, ): super().__init__(dependency=dependency, use_cache=use_cache) self.scopes = scopes or [] PK Ofastapi/py.typedPK O~"1fastapi/routing.pyimport asyncio import inspect import logging from typing import Any, Callable, Dict, List, Optional, Sequence, 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 DictIntStrAny, SetIntStr, jsonable_encoder from fastapi.exceptions import RequestValidationError, WebSocketRequestValidationError from fastapi.utils import create_cloned_field, generate_operation_id_for_path 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.types import ASGIApp from starlette.websockets import WebSocket def serialize_response( *, field: Field = None, response: Response, include: Union[SetIntStr, DictIntStrAny] = None, exclude: Union[SetIntStr, DictIntStrAny] = set(), by_alias: bool = True, skip_defaults: bool = False, ) -> Any: if field: errors = [] if skip_defaults and isinstance(response, BaseModel): response = response.dict(skip_defaults=skip_defaults) value, errors_ = field.validate(response, {}, loc=("response",)) if isinstance(errors_, ErrorWrapper): errors.append(errors_) elif isinstance(errors_, list): errors.extend(errors_) if errors: raise ValidationError(errors, field.type_) return jsonable_encoder( value, include=include, exclude=exclude, by_alias=by_alias, skip_defaults=skip_defaults, ) else: return jsonable_encoder(response) 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: Union[SetIntStr, DictIntStrAny] = None, response_model_exclude: Union[SetIntStr, DictIntStrAny] = set(), response_model_by_alias: bool = True, response_model_skip_defaults: bool = False, dependency_overrides_provider: Any = None, ) -> 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 solved_result = await solve_dependencies( request=request, dependant=dependant, body=body, dependency_overrides_provider=dependency_overrides_provider, ) values, errors, background_tasks, sub_response, _ = solved_result 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, ) response = response_class( content=response_data, status_code=status_code, background=background_tasks, ) response.headers.raw.extend(sub_response.headers.raw) if sub_response.status_code: response.status_code = sub_response.status_code return response return app def get_websocket_app( dependant: Dependant, dependency_overrides_provider: Any = None ) -> Callable: async def app(websocket: WebSocket) -> None: solved_result = await solve_dependencies( request=websocket, dependant=dependant, dependency_overrides_provider=dependency_overrides_provider, ) values, errors, _, _2, _3 = solved_result if errors: await websocket.close(code=WS_1008_POLICY_VIOLATION) raise WebSocketRequestValidationError(errors) assert dependant.call is not None, "dependant.call must be a function" await dependant.call(**values) return app class APIWebSocketRoute(routing.WebSocketRoute): def __init__( self, path: str, endpoint: Callable, *, name: str = None, dependency_overrides_provider: Any = 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, dependency_overrides_provider=dependency_overrides_provider, ) ) 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: Sequence[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: Optional[Union[Set[str], List[str]]] = None, operation_id: str = None, response_model_include: Union[SetIntStr, DictIntStrAny] = None, response_model_exclude: Union[SetIntStr, DictIntStrAny] = set(), response_model_by_alias: bool = True, response_model_skip_defaults: bool = False, include_in_schema: bool = True, response_class: Type[Response] = JSONResponse, dependency_overrides_provider: Any = None, ) -> None: self.path = path self.endpoint = endpoint self.name = get_name(endpoint) if name is None else name self.path_regex, self.path_format, self.param_convertors = compile_path(path) if methods is None: methods = ["GET"] self.methods = set([method.upper() for method in methods]) self.unique_id = generate_operation_id_for_path( name=self.name, path=self.path_format, method=list(methods)[0] ) 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.unique_id 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), ) # Create a clone of the field, so that a Pydantic submodel is not returned # as is just because it's an instance of a subclass of a more limited class # e.g. UserInDB (containing hashed_password) could be a subclass of User # that doesn't have the hashed_password. But because it's a subclass, it # would pass the validation and be returned as is. # By being a new field, no inheritance will be passed as is. A new model # will be always created. self.secure_cloned_response_field: Optional[Field] = create_cloned_field( self.response_field ) else: self.response_field = None self.secure_cloned_response_field = None self.status_code = status_code self.tags = tags or [] if dependencies: self.dependencies = list(dependencies) else: self.dependencies = [] 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.unique_id}" 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 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 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.unique_id) self.dependency_overrides_provider = dependency_overrides_provider 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.secure_cloned_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, dependency_overrides_provider=self.dependency_overrides_provider, ) ) class APIRouter(routing.Router): def __init__( self, routes: List[routing.BaseRoute] = None, redirect_slashes: bool = True, default: ASGIApp = None, dependency_overrides_provider: Any = None, route_class: Type[APIRoute] = APIRoute, ) -> None: super().__init__( routes=routes, redirect_slashes=redirect_slashes, default=default ) self.dependency_overrides_provider = dependency_overrides_provider self.route_class = route_class def add_api_route( self, path: str, endpoint: Callable, *, response_model: Type[Any] = None, status_code: int = 200, tags: List[str] = None, dependencies: Sequence[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: Optional[Union[Set[str], List[str]]] = None, operation_id: str = None, response_model_include: Union[SetIntStr, DictIntStrAny] = None, response_model_exclude: Union[SetIntStr, DictIntStrAny] = 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 = self.route_class( path, endpoint=endpoint, response_model=response_model, status_code=status_code, tags=tags or [], dependencies=dependencies, 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, dependency_overrides_provider=self.dependency_overrides_provider, ) self.routes.append(route) def api_route( self, path: str, *, response_model: Type[Any] = None, status_code: int = 200, tags: List[str] = None, dependencies: Sequence[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: Union[SetIntStr, DictIntStrAny] = None, response_model_exclude: Union[SetIntStr, DictIntStrAny] = 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, 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: Sequence[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 '/'" else: for r in router.routes: path = getattr(r, "path") name = getattr(r, "name", "unknown") if path is not None and not path: raise Exception( f"Prefix and path cannot be both empty (path operation: {name})" ) 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=list(dependencies or []) + list(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=list(route.methods or []), 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: Sequence[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: Union[SetIntStr, DictIntStrAny] = None, response_model_exclude: Union[SetIntStr, DictIntStrAny] = 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, 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: Sequence[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: Union[SetIntStr, DictIntStrAny] = None, response_model_exclude: Union[SetIntStr, DictIntStrAny] = 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, 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: Sequence[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: Union[SetIntStr, DictIntStrAny] = None, response_model_exclude: Union[SetIntStr, DictIntStrAny] = 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, 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: Sequence[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: Union[SetIntStr, DictIntStrAny] = None, response_model_exclude: Union[SetIntStr, DictIntStrAny] = 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, 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: Sequence[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: Union[SetIntStr, DictIntStrAny] = None, response_model_exclude: Union[SetIntStr, DictIntStrAny] = 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, 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: Sequence[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: Union[SetIntStr, DictIntStrAny] = None, response_model_exclude: Union[SetIntStr, DictIntStrAny] = 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, 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: Sequence[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: Union[SetIntStr, DictIntStrAny] = None, response_model_exclude: Union[SetIntStr, DictIntStrAny] = 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, 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: Sequence[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: Union[SetIntStr, DictIntStrAny] = None, response_model_exclude: Union[SetIntStr, DictIntStrAny] = 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, 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 O 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, known_models=set() ) 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, m_nested_models = 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)} def create_cloned_field(field: Field) -> Field: original_type = field.type_ if is_dataclass(original_type) and hasattr(original_type, "__pydantic_model__"): original_type = original_type.__pydantic_model__ # type: ignore use_type = original_type if lenient_issubclass(original_type, BaseModel): original_type = cast(Type[BaseModel], original_type) use_type = create_model( # type: ignore original_type.__name__, __config__=original_type.__config__, __validators__=original_type.__validators__, ) for f in original_type.__fields__.values(): use_type.__fields__[f.name] = f new_field = Field( name=field.name, type_=use_type, class_validators={}, default=None, required=False, model_config=BaseConfig, schema=Schema(None), ) new_field.has_alias = field.has_alias new_field.alias = field.alias new_field.class_validators = field.class_validators new_field.default = field.default new_field.required = field.required new_field.model_config = field.model_config new_field.schema = field.schema new_field.allow_none = field.allow_none new_field.validate_always = field.validate_always if field.sub_fields: new_field.sub_fields = [ create_cloned_field(sub_field) for sub_field in field.sub_fields ] if field.key_field: new_field.key_field = create_cloned_field(field.key_field) new_field.validators = field.validators new_field.whole_pre_validators = field.whole_pre_validators new_field.whole_post_validators = field.whole_post_validators new_field.parse_json = field.parse_json new_field.shape = field.shape new_field._populate_validators() return new_field def generate_operation_id_for_path(*, name: str, path: str, method: str) -> str: operation_id = name + path operation_id = operation_id.replace("{", "_").replace("}", "_").replace("/", "_") operation_id = operation_id + "_" + method.lower() return operation_id PK O fastapi/dependencies/__init__.pyPK OIɁfastapi/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, response_param_name: str = None, background_tasks_param_name: str = None, security_scopes_param_name: str = None, security_scopes: List[str] = None, use_cache: bool = True, path: 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.response_param_name = response_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 self.use_cache = use_cache # Store the path to be able to re-generate a dependable from it in overrides self.path = path # Save the cache key at creation to optimize performance self.cache_key = (self.call, tuple(sorted(set(self.security_scopes or [])))) PK O9FUFUfastapi/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.responses import Response 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, use_cache=depends.use_cache, ) if security_requirement: sub_dependant.security_requirements.append(security_requirement) sub_dependant.security_scopes = security_scopes return sub_dependant CacheKey = Tuple[Optional[Callable], Tuple[str, ...]] def get_flat_dependant( dependant: Dependant, *, skip_repeats: bool = False, visited: List[CacheKey] = None ) -> Dependant: if visited is None: visited = [] visited.append(dependant.cache_key) 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(), use_cache=dependant.use_cache, path=dependant.path, ) for sub_dependant in dependant.dependencies: if skip_repeats and sub_dependant.cache_key in visited: continue flat_sub = get_flat_dependant( sub_dependant, skip_repeats=skip_repeats, visited=visited ) 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: if not ( field.shape == Shape.SINGLETON and not lenient_issubclass(field.type_, BaseModel) and not lenient_issubclass(field.type_, sequence_types + (dict,)) and not isinstance(field.schema, params.Body) ): return False if field.sub_fields: if not all(is_scalar_field(f) for f in field.sub_fields): return False return True 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 if lenient_issubclass(field.type_, sequence_types): return True return False def get_dependant( *, path: str, call: Callable, name: str = None, security_scopes: List[str] = None, use_cache: bool = True, ) -> 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, path=path, use_cache=use_cache) 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, Response): dependant.response_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 # type: ignore 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: Optional[Union[Dict[str, Any], FormData]] = None, background_tasks: BackgroundTasks = None, response: Response = None, dependency_overrides_provider: Any = None, dependency_cache: Dict[Tuple[Callable, Tuple[str]], Any] = None, ) -> Tuple[ Dict[str, Any], List[ErrorWrapper], Optional[BackgroundTasks], Response, Dict[Tuple[Callable, Tuple[str]], Any], ]: values: Dict[str, Any] = {} errors: List[ErrorWrapper] = [] response = response or Response( # type: ignore content=None, status_code=None, headers=None, media_type=None, background=None ) dependency_cache = dependency_cache or {} sub_dependant: Dependant for sub_dependant in dependant.dependencies: sub_dependant.call = cast(Callable, sub_dependant.call) sub_dependant.cache_key = cast( Tuple[Callable, Tuple[str]], sub_dependant.cache_key ) call = sub_dependant.call use_sub_dependant = sub_dependant if ( dependency_overrides_provider and dependency_overrides_provider.dependency_overrides ): original_call = sub_dependant.call call = getattr( dependency_overrides_provider, "dependency_overrides", {} ).get(original_call, original_call) use_path: str = sub_dependant.path # type: ignore use_sub_dependant = get_dependant( path=use_path, call=call, name=sub_dependant.name, security_scopes=sub_dependant.security_scopes, ) solved_result = await solve_dependencies( request=request, dependant=use_sub_dependant, body=body, background_tasks=background_tasks, response=response, dependency_overrides_provider=dependency_overrides_provider, dependency_cache=dependency_cache, ) sub_values, sub_errors, background_tasks, sub_response, sub_dependency_cache = ( solved_result ) sub_response = cast(Response, sub_response) response.headers.raw.extend(sub_response.headers.raw) if sub_response.status_code: response.status_code = sub_response.status_code dependency_cache.update(sub_dependency_cache) if sub_errors: errors.extend(sub_errors) continue if sub_dependant.use_cache and sub_dependant.cache_key in dependency_cache: solved = dependency_cache[sub_dependant.cache_key] elif is_coroutine_callable(call): solved = await call(**sub_values) else: solved = await run_in_threadpool(call, **sub_values) if sub_dependant.name is not None: values[sub_dependant.name] = solved if sub_dependant.cache_key not in dependency_cache: dependency_cache[sub_dependant.cache_key] = 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 required_params=dependant.body_params, received_body=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.response_param_name: values[dependant.response_param_name] = response if dependant.security_scopes_param_name: values[dependant.security_scopes_param_name] = SecurityScopes( scopes=dependant.security_scopes ) return values, errors, background_tasks, response, dependency_cache 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 is_scalar_sequence_field(field) and isinstance( received_params, (QueryParams, Headers) ): value = received_params.getlist(field.alias) or field.default else: value = received_params.get(field.alias) schema = 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: Optional[Union[Dict[str, Any], FormData]], ) -> Tuple[Dict[str, Any], List[ErrorWrapper]]: values = {} errors = [] if required_params: field = required_params[0] embed = getattr(field.schema, "embed", None) if len(required_params) == 1 and not embed: received_body = {field.alias: received_body} for field in required_params: value: Any = None if received_body is not None: 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) BodySchema_kwargs: Dict[str, Any] = dict(default=None) 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 body_param_media_types = [ getattr(f.schema, "media_type") for f in flat_dependant.body_params if isinstance(f.schema, params.Body) ] if len(set(body_param_media_types)) == 1: BodySchema_kwargs["media_type"] = body_param_media_types[0] field = Field( name="body", type_=BodyModel, default=None, required=required, model_config=BaseConfig, class_validators={}, alias="body", schema=BodySchema(**BodySchema_kwargs), ) return field PK Ofastapi/openapi/__init__.pyPK Omaafastapi/openapi/constants.pyMETHODS_WITH_BODY = set(("POST", "PUT", "DELETE", "PATCH")) REF_PREFIX = "#/components/schemas/" PK O[aaafastapi/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", with_google_fonts: bool = True, ) -> HTMLResponse: html = f""" {title} """ if with_google_fonts: html += """ """ html += f""" """ return HTMLResponse(html) def get_swagger_ui_oauth2_redirect_html() -> HTMLResponse: html = """ """ return HTMLResponse(content=html) PK OJQ7[))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 # type: ignore 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( # type: ignore None, alias="schema" ) 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( # type: ignore None, alias="schema" ) 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 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: 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 OtJ,J,fastapi/openapi/utils.pyimport http.client from 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 ( generate_operation_id_for_path, 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"}, } }, } status_code_ranges: Dict[str, str] = { "1XX": "Information", "2XX": "Success", "3XX": "Redirection", "4XX": "Client Error", "5XX": "Server Error", "DEFAULT": "Default Response", } def get_openapi_params(dependant: Dependant) -> List[Field]: flat_dependant = get_flat_dependant(dependant, skip_repeats=True) 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] ) -> List[Dict[str, Any]]: parameters = [] for param in all_route_params: schema = param.schema schema = cast(Param, schema) 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 parameters def get_openapi_operation_request_body( *, body_field: Optional[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 return generate_operation_id_for_path(name=route.name, path=path, method=method) 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, skip_repeats=True) 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) operation_parameters = get_openapi_operation_parameters(all_route_params) 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 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( route.response_class.media_type, {} )["schema"] = response_schema status_text: Optional[str] = status_code_ranges.get( str(additional_status_code).upper() ) or http.client.responses.get(int(additional_status_code)) response.setdefault( "description", status_text or "Additional Response" ) status_code_key = str(additional_status_code).upper() if status_code_key == "DEFAULT": status_code_key = "default" operation.setdefault("responses", {})[status_code_key] = 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 http422 = str(HTTP_422_UNPROCESSABLE_ENTITY) if (all_route_params or route.body_field) and not any( [ status in operation["responses"] for status in [http422, "4xx", "default"] ] ): operation["responses"][http422] = { "description": "Validation Error", "content": { "application/json": { "schema": {"$ref": REF_PREFIX + "HTTPValidationError"} } }, } if "ValidationError" not in definitions: definitions.update( { "ValidationError": validation_error_definition, "HTTPValidationError": validation_error_response_definition, } ) 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 O߸^^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 OMl3 3 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 = 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 Or(ԍfastapi/security/base.pyfrom fastapi.openapi.models import SecurityBase as SecurityBaseModel class SecurityBase: model: SecurityBaseModel scheme_name: str PK OuQKfastapi/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__( # type: ignore 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": if self.auto_error: raise HTTPException( status_code=HTTP_403_FORBIDDEN, detail="Invalid authentication credentials", ) else: return None 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_k""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.param_functions 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 O'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 O  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 O{>> fastapi-0.38.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.38.0.dist-info/WHEEL HM K-*ϳR03rOK-J,/RH,szd&Y)r$[)T&UrPK!H!t H!fastapi-0.38.0.dist-info/METADATA\rȑO1/%JlE#-m9זkK!9DsS[׸?)&$ET@2/L$>}wƋy:Rժm;ܭyE^6;?e&UWLx"R,b_vL]FJ~a&`&,, ?:I$I|R ]?zq&Wʗ7?9uR$;r;/d̑Q똮 }\Lj ю{Tj?S)/αA>Cҏ._%ёh{aؾ dB?yuP)ɉ0WU̒b-lE:(/X+>$^F{y +V~Ś>M8E (W~6KׁGp q} yPRkqTE CP9殻n5-\fl-0PJ'zj,\"K $ϋo_}~ 6 Ōnlr1޼VzF(pm^/#ZZH8>j#@-!=1訽Bgjza2NXw.MqGm"g;n唁61 ao]{'`Ҏ$/m0%4<*_`j4?f(-bPa`c?QlI"ꪤ7|; y=C^]yocI[r o%51&Ҭ8*љΚD$҂,;qbȣK/xF$UDO9&4p_/NSc/|"?rQR8V3:;:-U9sS<-]iEJ+^2FK7D*?Aiͮapu )-XL9(\a8| ÊsZו/0eeGR$#V;j {M5s%Oc5 aթ40EELxäEkC;)x<ք񝄗|P yRʦP1ABfYiWtBI{rEHS$.n8PjgNF-W'Q"L +mFvgbb^N|@2(w"ф9AƲ~N5gFOms?p(:LՁ_?zC^#u%|5w*R(xQ=2lf)3ғ(/YsHP I~8Ά]q^5OȈ(Icqvu%XQDR莢LQСI2mwz8ZF|'9SBѳK_j=[䏍W =h>IW #1$SV><h^$1l6!l "LDX"M'(IYcS@N18օ%l)$NE1R8b2DTel 8艗 \('3J"Et:u'Eɗx?PRQyb9 @M} _w nDby1W:#yx>{v~ \W͕CF]sE';$)s` S0*qno+0NcRZi"U)EzViSR u]b*f'Ouo4Q ^Wl7y@,CU?QMPgҒ8U>9Z^g_< =J#$u{ :妳5ިT3C"*YCV 8*/76! 7ڽf+#/$76ϔI~Gפ} , ڿοNUA.Uk۰( t*7ͅ,hWW?̩@±:8joSoJC㳌c!Ucro fj![-3S--@k|el Ѡo1 /CG}o6WkRwEI9uK5O9ezC}JcRdQ&l M.88vz<-^)A V6;ɗUsϪѱ1wE̶FVۯXAx@t nSީ)Aq]  ō{BPA}`kkޣGTP&D*h0*hӄF9~X{P}1lm;ZVqfp$ n@+9|܉(0ȋ<_A1wB<.46nlKK|\ftDaN&3ޫ.o FtsC*J;13+mxJM>S]Sڠ$1]¨iJ .e"6l R&r\xVhOA-bx`Z7(fsQ\PSc!У >.VezNwt &S02)5dhl(Ro2\@`y,IغBٜ#`@y67p&UCͿMB Qc|J$8ogBĔITT=hF)֧2#oj6aC蕒)Dw˷s%.omdO%[$nܭllf9=u:gv!yCϿ*Sr4u*G<^6nIOō^pʹ#,S_kaɳu,5fOG8ɍ7FA4UFEmŶFs۪7#X(I]fc37p #^g=ToIhf6! pZvf2su])f!3(sTheoUYYm5C*-/3,D;J3L@5FPf+݀{;m h+KAw ֣@`[VP#vT4S|v0s AþfRfFv;w\h9m&9 {mDZ7 ~_/:FgN3 'Ν|c OrMm,㦨LB]8ˊ wF40QBQ>zս|/285s>ִJ7}2Z9C!pIcz,~UTjC~^λvΧrl!yp{ (IY0uIM;J3˕G 1b f i#Su:u"F=d h64dUәӡh월\м6{;1Ymf,43'JI9:G2XcwIkgaBԝKT>c(r; LJtK>4fepi>pPa{rTftF.7֟PqX){!"e *7FEVyd'v.da!HݳGVGw^ 8`g (ؗno7#N ~Q~ؗI <=D4!0G[jbirvޛMWltX/,4iU^n~-?ZG36l.zoP5l#+B /2'О=M@f q΁5OfFnm9]BkL7Jqm붖n52Xg s۶wō2YbJ xRS(KG )%< L\o.c&x 珃u]hAKy*%PUvc0T?FgDOj ޼\m{ ޭT&_AFIgwb&TP=@ЏPIڥMZ_}_7/Y_{_6& .430#v^I9"$5r8>U7먬,zÓ۵Td[RGwg^c[!:\0}dVdAypb&?IjJm Pi 򃆝+ҏϳ}Z`iY\퐠4 /tA]Ҿ3j{b 塑x2* IPڨ'A0cC1m_mfN㣚e-x;$Z`uxi*`<#a| t{%r)&[zwR Y6@Nr=LnFᘾI {Zf_&Im$G Eڬ1w=(dd lz> kEfastapi-0.38.0.dist-info/LICENSEPK!HPOIfastapi-0.38.0.dist-info/WHEELPK!H!t H!sJfastapi-0.38.0.dist-info/METADATAPK!Hp&h &cfastapi-0.38.0.dist-info/RECORDPKbh