PK |N fastapi/__init__.py"""FastAPI framework, high performance, easy to learn, fast to code, ready for production"""
__version__ = "0.26.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 |NyU U fastapi/applications.pyfrom typing import Any, Callable, Dict, List, Optional, Set, Type, Union
from fastapi import routing
from fastapi.exception_handlers import (
http_exception_handler,
request_validation_exception_handler,
)
from fastapi.exceptions import RequestValidationError
from fastapi.openapi.docs import (
get_redoc_html,
get_swagger_ui_html,
get_swagger_ui_oauth2_redirect_html,
)
from fastapi.openapi.utils import get_openapi
from fastapi.params import Depends
from starlette.applications import Starlette
from starlette.exceptions import ExceptionMiddleware, HTTPException
from starlette.middleware.errors import ServerErrorMiddleware
from starlette.requests import Request
from starlette.responses import HTMLResponse, JSONResponse, Response
from starlette.routing import BaseRoute
class FastAPI(Starlette):
def __init__(
self,
debug: bool = False,
routes: List[BaseRoute] = None,
template_directory: str = None,
title: str = "Fast API",
description: str = "",
version: str = "0.1.0",
openapi_url: Optional[str] = "/openapi.json",
openapi_prefix: str = "",
docs_url: Optional[str] = "/docs",
redoc_url: Optional[str] = "/redoc",
swagger_ui_oauth2_redirect_url: Optional[str] = "/docs/oauth2-redirect",
**extra: Dict[str, Any],
) -> None:
self._debug = debug
self.router: routing.APIRouter = routing.APIRouter(routes)
self.exception_middleware = ExceptionMiddleware(self.router, debug=debug)
self.error_middleware = ServerErrorMiddleware(
self.exception_middleware, debug=debug
)
self.title = title
self.description = description
self.version = version
self.openapi_url = openapi_url
self.openapi_prefix = openapi_prefix.rstrip("/")
self.docs_url = docs_url
self.redoc_url = redoc_url
self.swagger_ui_oauth2_redirect_url = swagger_ui_oauth2_redirect_url
self.extra = extra
self.openapi_version = "3.0.2"
if self.openapi_url:
assert self.title, "A title must be provided for OpenAPI, e.g.: 'My API'"
assert self.version, "A version must be provided for OpenAPI, e.g.: '2.1.0'"
if self.docs_url or self.redoc_url:
assert self.openapi_url, "The openapi_url is required for the docs"
self.openapi_schema: Optional[Dict[str, Any]] = None
self.setup()
def openapi(self) -> Dict:
if not self.openapi_schema:
self.openapi_schema = get_openapi(
title=self.title,
version=self.version,
openapi_version=self.openapi_version,
description=self.description,
routes=self.routes,
openapi_prefix=self.openapi_prefix,
)
return self.openapi_schema
def setup(self) -> None:
if self.openapi_url:
async def openapi(req: Request) -> JSONResponse:
return JSONResponse(self.openapi())
self.add_route(self.openapi_url, openapi, include_in_schema=False)
openapi_url = self.openapi_prefix + self.openapi_url
if self.openapi_url and self.docs_url:
async def swagger_ui_html(req: Request) -> HTMLResponse:
return get_swagger_ui_html(
openapi_url=openapi_url,
title=self.title + " - Swagger UI",
oauth2_redirect_url=self.swagger_ui_oauth2_redirect_url,
)
self.add_route(self.docs_url, swagger_ui_html, include_in_schema=False)
if self.swagger_ui_oauth2_redirect_url:
async def swagger_ui_redirect(req: Request) -> HTMLResponse:
return get_swagger_ui_oauth2_redirect_html()
self.add_route(
self.swagger_ui_oauth2_redirect_url,
swagger_ui_redirect,
include_in_schema=False,
)
if self.openapi_url and self.redoc_url:
async def redoc_html(req: Request) -> HTMLResponse:
return get_redoc_html(
openapi_url=openapi_url, title=self.title + " - ReDoc"
)
self.add_route(self.redoc_url, redoc_html, include_in_schema=False)
self.add_exception_handler(HTTPException, http_exception_handler)
self.add_exception_handler(
RequestValidationError, request_validation_exception_handler
)
def add_api_route(
self,
path: str,
endpoint: Callable,
*,
response_model: Type[Any] = None,
status_code: int = 200,
tags: List[str] = None,
dependencies: List[Depends] = None,
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
methods: List[str] = None,
operation_id: str = None,
response_model_include: Set[str] = None,
response_model_exclude: Set[str] = set(),
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
name: str = None,
) -> None:
self.router.add_api_route(
path,
endpoint=endpoint,
response_model=response_model,
status_code=status_code,
tags=tags or [],
dependencies=dependencies or [],
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
methods=methods,
operation_id=operation_id,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
response_model_skip_defaults=response_model_skip_defaults,
include_in_schema=include_in_schema,
response_class=response_class,
name=name,
)
def api_route(
self,
path: str,
*,
response_model: Type[Any] = None,
status_code: int = 200,
tags: List[str] = None,
dependencies: List[Depends] = None,
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
methods: List[str] = None,
operation_id: str = None,
response_model_include: Set[str] = None,
response_model_exclude: Set[str] = set(),
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
name: str = None,
) -> Callable:
def decorator(func: Callable) -> Callable:
self.router.add_api_route(
path,
func,
response_model=response_model,
status_code=status_code,
tags=tags or [],
dependencies=dependencies or [],
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
methods=methods,
operation_id=operation_id,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
response_model_skip_defaults=response_model_skip_defaults,
include_in_schema=include_in_schema,
response_class=response_class,
name=name,
)
return func
return decorator
def add_api_websocket_route(
self, path: str, endpoint: Callable, name: str = None
) -> None:
self.router.add_api_websocket_route(path, endpoint, name=name)
def websocket(self, path: str, name: str = None) -> Callable:
def decorator(func: Callable) -> Callable:
self.add_api_websocket_route(path, func, name=name)
return func
return decorator
def include_router(
self,
router: routing.APIRouter,
*,
prefix: str = "",
tags: List[str] = None,
dependencies: List[Depends] = None,
responses: Dict[Union[int, str], Dict[str, Any]] = None,
) -> None:
self.router.include_router(
router,
prefix=prefix,
tags=tags,
dependencies=dependencies,
responses=responses or {},
)
def get(
self,
path: str,
*,
response_model: Type[Any] = None,
status_code: int = 200,
tags: List[str] = None,
dependencies: List[Depends] = None,
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
operation_id: str = None,
response_model_include: Set[str] = None,
response_model_exclude: Set[str] = set(),
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
name: str = None,
) -> Callable:
return self.router.get(
path,
response_model=response_model,
status_code=status_code,
tags=tags or [],
dependencies=dependencies or [],
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
operation_id=operation_id,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
response_model_skip_defaults=response_model_skip_defaults,
include_in_schema=include_in_schema,
response_class=response_class,
name=name,
)
def put(
self,
path: str,
*,
response_model: Type[Any] = None,
status_code: int = 200,
tags: List[str] = None,
dependencies: List[Depends] = None,
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
operation_id: str = None,
response_model_include: Set[str] = None,
response_model_exclude: Set[str] = set(),
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
name: str = None,
) -> Callable:
return self.router.put(
path,
response_model=response_model,
status_code=status_code,
tags=tags or [],
dependencies=dependencies or [],
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
operation_id=operation_id,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
response_model_skip_defaults=response_model_skip_defaults,
include_in_schema=include_in_schema,
response_class=response_class,
name=name,
)
def post(
self,
path: str,
*,
response_model: Type[Any] = None,
status_code: int = 200,
tags: List[str] = None,
dependencies: List[Depends] = None,
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
operation_id: str = None,
response_model_include: Set[str] = None,
response_model_exclude: Set[str] = set(),
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
name: str = None,
) -> Callable:
return self.router.post(
path,
response_model=response_model,
status_code=status_code,
tags=tags or [],
dependencies=dependencies or [],
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
operation_id=operation_id,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
response_model_skip_defaults=response_model_skip_defaults,
include_in_schema=include_in_schema,
response_class=response_class,
name=name,
)
def delete(
self,
path: str,
*,
response_model: Type[Any] = None,
status_code: int = 200,
tags: List[str] = None,
dependencies: List[Depends] = None,
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
operation_id: str = None,
response_model_include: Set[str] = None,
response_model_exclude: Set[str] = set(),
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
name: str = None,
) -> Callable:
return self.router.delete(
path,
response_model=response_model,
status_code=status_code,
tags=tags or [],
dependencies=dependencies or [],
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
operation_id=operation_id,
response_model_skip_defaults=response_model_skip_defaults,
include_in_schema=include_in_schema,
response_class=response_class,
name=name,
)
def options(
self,
path: str,
*,
response_model: Type[Any] = None,
status_code: int = 200,
tags: List[str] = None,
dependencies: List[Depends] = None,
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
operation_id: str = None,
response_model_include: Set[str] = None,
response_model_exclude: Set[str] = set(),
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
name: str = None,
) -> Callable:
return self.router.options(
path,
response_model=response_model,
status_code=status_code,
tags=tags or [],
dependencies=dependencies or [],
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
operation_id=operation_id,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
response_model_skip_defaults=response_model_skip_defaults,
include_in_schema=include_in_schema,
response_class=response_class,
name=name,
)
def head(
self,
path: str,
*,
response_model: Type[Any] = None,
status_code: int = 200,
tags: List[str] = None,
dependencies: List[Depends] = None,
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
operation_id: str = None,
response_model_include: Set[str] = None,
response_model_exclude: Set[str] = set(),
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
name: str = None,
) -> Callable:
return self.router.head(
path,
response_model=response_model,
status_code=status_code,
tags=tags or [],
dependencies=dependencies or [],
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
operation_id=operation_id,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
response_model_skip_defaults=response_model_skip_defaults,
include_in_schema=include_in_schema,
response_class=response_class,
name=name,
)
def patch(
self,
path: str,
*,
response_model: Type[Any] = None,
status_code: int = 200,
tags: List[str] = None,
dependencies: List[Depends] = None,
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
operation_id: str = None,
response_model_include: Set[str] = None,
response_model_exclude: Set[str] = set(),
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
name: str = None,
) -> Callable:
return self.router.patch(
path,
response_model=response_model,
status_code=status_code,
tags=tags or [],
dependencies=dependencies or [],
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
operation_id=operation_id,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
response_model_skip_defaults=response_model_skip_defaults,
include_in_schema=include_in_schema,
response_class=response_class,
name=name,
)
def trace(
self,
path: str,
*,
response_model: Type[Any] = None,
status_code: int = 200,
tags: List[str] = None,
dependencies: List[Depends] = None,
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
operation_id: str = None,
response_model_include: Set[str] = None,
response_model_exclude: Set[str] = set(),
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
name: str = None,
) -> Callable:
return self.router.trace(
path,
response_model=response_model,
status_code=status_code,
tags=tags or [],
dependencies=dependencies or [],
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
operation_id=operation_id,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
response_model_skip_defaults=response_model_skip_defaults,
include_in_schema=include_in_schema,
response_class=response_class,
name=name,
)
PK |N fastapi/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 |N Ŀ fastapi/encoders.pyfrom enum import Enum
from types import GeneratorType
from typing import Any, List, Set
from pydantic import BaseModel
from pydantic.json import ENCODERS_BY_TYPE
def jsonable_encoder(
obj: Any,
include: Set[str] = None,
exclude: Set[str] = set(),
by_alias: bool = True,
skip_defaults: bool = False,
include_none: bool = True,
custom_encoder: dict = {},
sqlalchemy_safe: bool = True,
) -> Any:
if include is not None and not isinstance(include, set):
include = set(include)
if exclude is not None and not isinstance(exclude, set):
exclude = set(exclude)
if isinstance(obj, BaseModel):
encoder = getattr(obj.Config, "json_encoders", custom_encoder)
return jsonable_encoder(
obj.dict(
include=include,
exclude=exclude,
by_alias=by_alias,
skip_defaults=skip_defaults,
),
include_none=include_none,
custom_encoder=encoder,
sqlalchemy_safe=sqlalchemy_safe,
)
if isinstance(obj, Enum):
return obj.value
if isinstance(obj, (str, int, float, type(None))):
return obj
if isinstance(obj, dict):
encoded_dict = {}
for key, value in obj.items():
if (
(
not sqlalchemy_safe
or (not isinstance(key, str))
or (not key.startswith("_sa"))
)
and (value is not None or include_none)
and ((include and key in include) or key not in exclude)
):
encoded_key = jsonable_encoder(
key,
by_alias=by_alias,
skip_defaults=skip_defaults,
include_none=include_none,
custom_encoder=custom_encoder,
sqlalchemy_safe=sqlalchemy_safe,
)
encoded_value = jsonable_encoder(
value,
by_alias=by_alias,
skip_defaults=skip_defaults,
include_none=include_none,
custom_encoder=custom_encoder,
sqlalchemy_safe=sqlalchemy_safe,
)
encoded_dict[encoded_key] = encoded_value
return encoded_dict
if isinstance(obj, (list, set, frozenset, GeneratorType, tuple)):
encoded_list = []
for item in obj:
encoded_list.append(
jsonable_encoder(
item,
include=include,
exclude=exclude,
by_alias=by_alias,
skip_defaults=skip_defaults,
include_none=include_none,
custom_encoder=custom_encoder,
sqlalchemy_safe=sqlalchemy_safe,
)
)
return encoded_list
errors: List[Exception] = []
try:
if custom_encoder and type(obj) in custom_encoder:
encoder = custom_encoder[type(obj)]
else:
encoder = ENCODERS_BY_TYPE[type(obj)]
return encoder(obj)
except KeyError as e:
errors.append(e)
try:
data = dict(obj)
except Exception as e:
errors.append(e)
try:
data = vars(obj)
except Exception as e:
errors.append(e)
raise ValueError(errors)
return jsonable_encoder(
data,
by_alias=by_alias,
skip_defaults=skip_defaults,
include_none=include_none,
custom_encoder=custom_encoder,
sqlalchemy_safe=sqlalchemy_safe,
)
PK |NG G fastapi/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 |N fastapi/exceptions.pyfrom pydantic import ValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
class HTTPException(StarletteHTTPException):
def __init__(
self, status_code: int, detail: str = None, headers: dict = None
) -> None:
super().__init__(status_code=status_code, detail=detail)
self.headers = headers
class RequestValidationError(ValidationError):
pass
class WebSocketRequestValidationError(ValidationError):
pass
PK |NhS S fastapi/param_functions.pyfrom typing import Any, Callable, Sequence
from fastapi import params
def Path( # noqa: N802
default: Any,
*,
alias: str = None,
title: str = None,
description: str = None,
gt: float = None,
ge: float = None,
lt: float = None,
le: float = None,
min_length: int = None,
max_length: int = None,
regex: str = None,
deprecated: bool = None,
**extra: Any,
) -> Any:
return params.Path(
default=default,
alias=alias,
title=title,
description=description,
gt=gt,
ge=ge,
lt=lt,
le=le,
min_length=min_length,
max_length=max_length,
regex=regex,
deprecated=deprecated,
**extra,
)
def Query( # noqa: N802
default: Any,
*,
alias: str = None,
title: str = None,
description: str = None,
gt: float = None,
ge: float = None,
lt: float = None,
le: float = None,
min_length: int = None,
max_length: int = None,
regex: str = None,
deprecated: bool = None,
**extra: Any,
) -> Any:
return params.Query(
default,
alias=alias,
title=title,
description=description,
gt=gt,
ge=ge,
lt=lt,
le=le,
min_length=min_length,
max_length=max_length,
regex=regex,
deprecated=deprecated,
**extra,
)
def Header( # noqa: N802
default: Any,
*,
alias: str = None,
convert_underscores: bool = True,
title: str = None,
description: str = None,
gt: float = None,
ge: float = None,
lt: float = None,
le: float = None,
min_length: int = None,
max_length: int = None,
regex: str = None,
deprecated: bool = None,
**extra: Any,
) -> Any:
return params.Header(
default,
alias=alias,
convert_underscores=convert_underscores,
title=title,
description=description,
gt=gt,
ge=ge,
lt=lt,
le=le,
min_length=min_length,
max_length=max_length,
regex=regex,
deprecated=deprecated,
**extra,
)
def Cookie( # noqa: N802
default: Any,
*,
alias: str = None,
title: str = None,
description: str = None,
gt: float = None,
ge: float = None,
lt: float = None,
le: float = None,
min_length: int = None,
max_length: int = None,
regex: str = None,
deprecated: bool = None,
**extra: Any,
) -> Any:
return params.Cookie(
default,
alias=alias,
title=title,
description=description,
gt=gt,
ge=ge,
lt=lt,
le=le,
min_length=min_length,
max_length=max_length,
regex=regex,
deprecated=deprecated,
**extra,
)
def Body( # noqa: N802
default: Any,
*,
embed: bool = False,
media_type: str = "application/json",
alias: str = None,
title: str = None,
description: str = None,
gt: float = None,
ge: float = None,
lt: float = None,
le: float = None,
min_length: int = None,
max_length: int = None,
regex: str = None,
**extra: Any,
) -> Any:
return params.Body(
default,
embed=embed,
media_type=media_type,
alias=alias,
title=title,
description=description,
gt=gt,
ge=ge,
lt=lt,
le=le,
min_length=min_length,
max_length=max_length,
regex=regex,
**extra,
)
def Form( # noqa: N802
default: Any,
*,
media_type: str = "application/x-www-form-urlencoded",
alias: str = None,
title: str = None,
description: str = None,
gt: float = None,
ge: float = None,
lt: float = None,
le: float = None,
min_length: int = None,
max_length: int = None,
regex: str = None,
**extra: Any,
) -> Any:
return params.Form(
default,
media_type=media_type,
alias=alias,
title=title,
description=description,
gt=gt,
ge=ge,
lt=lt,
le=le,
min_length=min_length,
max_length=max_length,
regex=regex,
**extra,
)
def File( # noqa: N802
default: Any,
*,
media_type: str = "multipart/form-data",
alias: str = None,
title: str = None,
description: str = None,
gt: float = None,
ge: float = None,
lt: float = None,
le: float = None,
min_length: int = None,
max_length: int = None,
regex: str = None,
**extra: Any,
) -> Any:
return params.File(
default,
media_type=media_type,
alias=alias,
title=title,
description=description,
gt=gt,
ge=ge,
lt=lt,
le=le,
min_length=min_length,
max_length=max_length,
regex=regex,
**extra,
)
def Depends(dependency: Callable = None) -> Any: # noqa: N802
return params.Depends(dependency=dependency)
def Security( # noqa: N802
dependency: Callable = None, scopes: Sequence[str] = None
) -> Any:
return params.Security(dependency=dependency, scopes=scopes)
PK |NVPC 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):
self.dependency = dependency
class Security(Depends):
def __init__(self, dependency: Callable = None, scopes: Sequence[str] = None):
self.scopes = scopes or []
super().__init__(dependency=dependency)
PK |N fastapi/py.typedPK |NYL7v v fastapi/routing.pyimport asyncio
import inspect
import logging
import re
from typing import Any, Callable, Dict, List, Optional, Set, Type, Union
from fastapi import params
from fastapi.dependencies.models import Dependant
from fastapi.dependencies.utils import (
get_body_field,
get_dependant,
get_parameterless_sub_dependant,
solve_dependencies,
)
from fastapi.encoders import jsonable_encoder
from fastapi.exceptions import RequestValidationError, WebSocketRequestValidationError
from pydantic import BaseConfig, BaseModel, Schema
from pydantic.error_wrappers import ErrorWrapper, ValidationError
from pydantic.fields import Field
from pydantic.utils import lenient_issubclass
from starlette import routing
from starlette.concurrency import run_in_threadpool
from starlette.exceptions import HTTPException
from starlette.requests import Request
from starlette.responses import JSONResponse, Response
from starlette.routing import (
compile_path,
get_name,
request_response,
websocket_session,
)
from starlette.status import WS_1008_POLICY_VIOLATION
from starlette.websockets import WebSocket
def serialize_response(
*,
field: Field = None,
response: Response,
include: Set[str] = None,
exclude: Set[str] = set(),
by_alias: bool = True,
skip_defaults: bool = False,
) -> Any:
encoded = jsonable_encoder(
response,
include=include,
exclude=exclude,
by_alias=by_alias,
skip_defaults=skip_defaults,
)
if field:
errors = []
value, errors_ = field.validate(encoded, {}, loc=("response",))
if isinstance(errors_, ErrorWrapper):
errors.append(errors_)
elif isinstance(errors_, list):
errors.extend(errors_)
if errors:
raise ValidationError(errors)
return jsonable_encoder(
value,
include=include,
exclude=exclude,
by_alias=by_alias,
skip_defaults=skip_defaults,
)
else:
return encoded
def get_app(
dependant: Dependant,
body_field: Field = None,
status_code: int = 200,
response_class: Type[Response] = JSONResponse,
response_field: Field = None,
response_model_include: Set[str] = None,
response_model_exclude: Set[str] = set(),
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
) -> Callable:
assert dependant.call is not None, "dependant.call must be a function"
is_coroutine = asyncio.iscoroutinefunction(dependant.call)
is_body_form = body_field and isinstance(body_field.schema, params.Form)
async def app(request: Request) -> Response:
try:
body = None
if body_field:
if is_body_form:
body = await request.form()
else:
body_bytes = await request.body()
if body_bytes:
body = await request.json()
except Exception as e:
logging.error(f"Error getting request body: {e}")
raise HTTPException(
status_code=400, detail="There was an error parsing the body"
) from e
values, errors, background_tasks = await solve_dependencies(
request=request, dependant=dependant, body=body
)
if errors:
raise RequestValidationError(errors)
else:
assert dependant.call is not None, "dependant.call must be a function"
if is_coroutine:
raw_response = await dependant.call(**values)
else:
raw_response = await run_in_threadpool(dependant.call, **values)
if isinstance(raw_response, Response):
if raw_response.background is None:
raw_response.background = background_tasks
return raw_response
response_data = serialize_response(
field=response_field,
response=raw_response,
include=response_model_include,
exclude=response_model_exclude,
by_alias=response_model_by_alias,
skip_defaults=response_model_skip_defaults,
)
return response_class(
content=response_data,
status_code=status_code,
background=background_tasks,
)
return app
def get_websocket_app(dependant: Dependant) -> Callable:
async def app(websocket: WebSocket) -> None:
values, errors, _ = await solve_dependencies(
request=websocket, dependant=dependant
)
if errors:
await websocket.close(code=WS_1008_POLICY_VIOLATION)
raise WebSocketRequestValidationError(errors)
assert dependant.call is not None, "dependant.call must me a function"
await dependant.call(**values)
return app
class APIWebSocketRoute(routing.WebSocketRoute):
def __init__(self, path: str, endpoint: Callable, *, name: str = None) -> None:
self.path = path
self.endpoint = endpoint
self.name = get_name(endpoint) if name is None else name
self.dependant = get_dependant(path=path, call=self.endpoint)
self.app = websocket_session(get_websocket_app(dependant=self.dependant))
regex = "^" + path + "$"
regex = re.sub("{([a-zA-Z_][a-zA-Z0-9_]*)}", r"(?P<\1>[^/]+)", regex)
self.path_regex, self.path_format, self.param_convertors = compile_path(path)
class APIRoute(routing.Route):
def __init__(
self,
path: str,
endpoint: Callable,
*,
response_model: Type[Any] = None,
status_code: int = 200,
tags: List[str] = None,
dependencies: List[params.Depends] = None,
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
name: str = None,
methods: List[str] = None,
operation_id: str = None,
response_model_include: Set[str] = None,
response_model_exclude: Set[str] = set(),
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
) -> None:
assert path.startswith("/"), "Routed paths must always start with '/'"
self.path = path
self.endpoint = endpoint
self.name = get_name(endpoint) if name is None else name
self.response_model = response_model
if self.response_model:
assert lenient_issubclass(
response_class, JSONResponse
), "To declare a type the response must be a JSON response"
response_name = "Response_" + self.name
self.response_field: Optional[Field] = Field(
name=response_name,
type_=self.response_model,
class_validators={},
default=None,
required=False,
model_config=BaseConfig,
schema=Schema(None),
)
else:
self.response_field = None
self.status_code = status_code
self.tags = tags or []
self.dependencies = dependencies or []
self.summary = summary
self.description = description or inspect.cleandoc(self.endpoint.__doc__ or "")
self.response_description = response_description
self.responses = responses or {}
response_fields = {}
for additional_status_code, response in self.responses.items():
assert isinstance(response, dict), "An additional response must be a dict"
model = response.get("model")
if model:
assert lenient_issubclass(
model, BaseModel
), "A response model must be a Pydantic model"
response_name = f"Response_{additional_status_code}_{self.name}"
response_field = Field(
name=response_name,
type_=model,
class_validators=None,
default=None,
required=False,
model_config=BaseConfig,
schema=Schema(None),
)
response_fields[additional_status_code] = response_field
if response_fields:
self.response_fields: Dict[Union[int, str], Field] = response_fields
else:
self.response_fields = {}
self.deprecated = deprecated
if methods is None:
methods = ["GET"]
self.methods = methods
self.operation_id = operation_id
self.response_model_include = response_model_include
self.response_model_exclude = response_model_exclude
self.response_model_by_alias = response_model_by_alias
self.response_model_skip_defaults = response_model_skip_defaults
self.include_in_schema = include_in_schema
self.response_class = response_class
self.path_regex, self.path_format, self.param_convertors = compile_path(path)
assert inspect.isfunction(endpoint) or inspect.ismethod(
endpoint
), f"An endpoint must be a function or method"
self.dependant = get_dependant(path=self.path_format, call=self.endpoint)
for depends in self.dependencies[::-1]:
self.dependant.dependencies.insert(
0,
get_parameterless_sub_dependant(depends=depends, path=self.path_format),
)
self.body_field = get_body_field(dependant=self.dependant, name=self.name)
self.app = request_response(
get_app(
dependant=self.dependant,
body_field=self.body_field,
status_code=self.status_code,
response_class=self.response_class,
response_field=self.response_field,
response_model_include=self.response_model_include,
response_model_exclude=self.response_model_exclude,
response_model_by_alias=self.response_model_by_alias,
response_model_skip_defaults=self.response_model_skip_defaults,
)
)
class APIRouter(routing.Router):
def add_api_route(
self,
path: str,
endpoint: Callable,
*,
response_model: Type[Any] = None,
status_code: int = 200,
tags: List[str] = None,
dependencies: List[params.Depends] = None,
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
methods: List[str] = None,
operation_id: str = None,
response_model_include: Set[str] = None,
response_model_exclude: Set[str] = set(),
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
name: str = None,
) -> None:
route = APIRoute(
path,
endpoint=endpoint,
response_model=response_model,
status_code=status_code,
tags=tags or [],
dependencies=dependencies or [],
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
methods=methods,
operation_id=operation_id,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
response_model_skip_defaults=response_model_skip_defaults,
include_in_schema=include_in_schema,
response_class=response_class,
name=name,
)
self.routes.append(route)
def api_route(
self,
path: str,
*,
response_model: Type[Any] = None,
status_code: int = 200,
tags: List[str] = None,
dependencies: List[params.Depends] = None,
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
methods: List[str] = None,
operation_id: str = None,
response_model_include: Set[str] = None,
response_model_exclude: Set[str] = set(),
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
name: str = None,
) -> Callable:
def decorator(func: Callable) -> Callable:
self.add_api_route(
path,
func,
response_model=response_model,
status_code=status_code,
tags=tags or [],
dependencies=dependencies or [],
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
methods=methods,
operation_id=operation_id,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
response_model_skip_defaults=response_model_skip_defaults,
include_in_schema=include_in_schema,
response_class=response_class,
name=name,
)
return func
return decorator
def add_api_websocket_route(
self, path: str, endpoint: Callable, name: str = None
) -> None:
route = APIWebSocketRoute(path, endpoint=endpoint, name=name)
self.routes.append(route)
def websocket(self, path: str, name: str = None) -> Callable:
def decorator(func: Callable) -> Callable:
self.add_api_websocket_route(path, func, name=name)
return func
return decorator
def include_router(
self,
router: "APIRouter",
*,
prefix: str = "",
tags: List[str] = None,
dependencies: List[params.Depends] = None,
responses: Dict[Union[int, str], Dict[str, Any]] = None,
) -> None:
if prefix:
assert prefix.startswith("/"), "A path prefix must start with '/'"
assert not prefix.endswith(
"/"
), "A path prefix must not end with '/', as the routes will start with '/'"
if responses is None:
responses = {}
for route in router.routes:
if isinstance(route, APIRoute):
combined_responses = {**responses, **route.responses}
self.add_api_route(
prefix + route.path,
route.endpoint,
response_model=route.response_model,
status_code=route.status_code,
tags=(route.tags or []) + (tags or []),
dependencies=(dependencies or []) + (route.dependencies or []),
summary=route.summary,
description=route.description,
response_description=route.response_description,
responses=combined_responses,
deprecated=route.deprecated,
methods=route.methods,
operation_id=route.operation_id,
response_model_include=route.response_model_include,
response_model_exclude=route.response_model_exclude,
response_model_by_alias=route.response_model_by_alias,
response_model_skip_defaults=route.response_model_skip_defaults,
include_in_schema=route.include_in_schema,
response_class=route.response_class,
name=route.name,
)
elif isinstance(route, routing.Route):
self.add_route(
prefix + route.path,
route.endpoint,
methods=route.methods,
include_in_schema=route.include_in_schema,
name=route.name,
)
elif isinstance(route, APIWebSocketRoute):
self.add_api_websocket_route(
prefix + route.path, route.endpoint, name=route.name
)
elif isinstance(route, routing.WebSocketRoute):
self.add_websocket_route(
prefix + route.path, route.endpoint, name=route.name
)
def get(
self,
path: str,
*,
response_model: Type[Any] = None,
status_code: int = 200,
tags: List[str] = None,
dependencies: List[params.Depends] = None,
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
operation_id: str = None,
response_model_include: Set[str] = None,
response_model_exclude: Set[str] = set(),
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
name: str = None,
) -> Callable:
return self.api_route(
path=path,
response_model=response_model,
status_code=status_code,
tags=tags or [],
dependencies=dependencies or [],
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
methods=["GET"],
operation_id=operation_id,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
response_model_skip_defaults=response_model_skip_defaults,
include_in_schema=include_in_schema,
response_class=response_class,
name=name,
)
def put(
self,
path: str,
*,
response_model: Type[Any] = None,
status_code: int = 200,
tags: List[str] = None,
dependencies: List[params.Depends] = None,
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
operation_id: str = None,
response_model_include: Set[str] = None,
response_model_exclude: Set[str] = set(),
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
name: str = None,
) -> Callable:
return self.api_route(
path=path,
response_model=response_model,
status_code=status_code,
tags=tags or [],
dependencies=dependencies or [],
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
methods=["PUT"],
operation_id=operation_id,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
response_model_skip_defaults=response_model_skip_defaults,
include_in_schema=include_in_schema,
response_class=response_class,
name=name,
)
def post(
self,
path: str,
*,
response_model: Type[Any] = None,
status_code: int = 200,
tags: List[str] = None,
dependencies: List[params.Depends] = None,
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
operation_id: str = None,
response_model_include: Set[str] = None,
response_model_exclude: Set[str] = set(),
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
name: str = None,
) -> Callable:
return self.api_route(
path=path,
response_model=response_model,
status_code=status_code,
tags=tags or [],
dependencies=dependencies or [],
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
methods=["POST"],
operation_id=operation_id,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
response_model_skip_defaults=response_model_skip_defaults,
include_in_schema=include_in_schema,
response_class=response_class,
name=name,
)
def delete(
self,
path: str,
*,
response_model: Type[Any] = None,
status_code: int = 200,
tags: List[str] = None,
dependencies: List[params.Depends] = None,
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
operation_id: str = None,
response_model_include: Set[str] = None,
response_model_exclude: Set[str] = set(),
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
name: str = None,
) -> Callable:
return self.api_route(
path=path,
response_model=response_model,
status_code=status_code,
tags=tags or [],
dependencies=dependencies or [],
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
methods=["DELETE"],
operation_id=operation_id,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
response_model_skip_defaults=response_model_skip_defaults,
include_in_schema=include_in_schema,
response_class=response_class,
name=name,
)
def options(
self,
path: str,
*,
response_model: Type[Any] = None,
status_code: int = 200,
tags: List[str] = None,
dependencies: List[params.Depends] = None,
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
operation_id: str = None,
response_model_include: Set[str] = None,
response_model_exclude: Set[str] = set(),
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
name: str = None,
) -> Callable:
return self.api_route(
path=path,
response_model=response_model,
status_code=status_code,
tags=tags or [],
dependencies=dependencies or [],
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
methods=["OPTIONS"],
operation_id=operation_id,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
response_model_skip_defaults=response_model_skip_defaults,
include_in_schema=include_in_schema,
response_class=response_class,
name=name,
)
def head(
self,
path: str,
*,
response_model: Type[Any] = None,
status_code: int = 200,
tags: List[str] = None,
dependencies: List[params.Depends] = None,
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
operation_id: str = None,
response_model_include: Set[str] = None,
response_model_exclude: Set[str] = set(),
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
name: str = None,
) -> Callable:
return self.api_route(
path=path,
response_model=response_model,
status_code=status_code,
tags=tags or [],
dependencies=dependencies or [],
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
methods=["HEAD"],
operation_id=operation_id,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
response_model_skip_defaults=response_model_skip_defaults,
include_in_schema=include_in_schema,
response_class=response_class,
name=name,
)
def patch(
self,
path: str,
*,
response_model: Type[Any] = None,
status_code: int = 200,
tags: List[str] = None,
dependencies: List[params.Depends] = None,
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
operation_id: str = None,
response_model_include: Set[str] = None,
response_model_exclude: Set[str] = set(),
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
name: str = None,
) -> Callable:
return self.api_route(
path=path,
response_model=response_model,
status_code=status_code,
tags=tags or [],
dependencies=dependencies or [],
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
methods=["PATCH"],
operation_id=operation_id,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
response_model_skip_defaults=response_model_skip_defaults,
include_in_schema=include_in_schema,
response_class=response_class,
name=name,
)
def trace(
self,
path: str,
*,
response_model: Type[Any] = None,
status_code: int = 200,
tags: List[str] = None,
dependencies: List[params.Depends] = None,
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
operation_id: str = None,
response_model_include: Set[str] = None,
response_model_exclude: Set[str] = set(),
response_model_by_alias: bool = True,
response_model_skip_defaults: bool = False,
include_in_schema: bool = True,
response_class: Type[Response] = JSONResponse,
name: str = None,
) -> Callable:
return self.api_route(
path=path,
response_model=response_model,
status_code=status_code,
tags=tags or [],
dependencies=dependencies or [],
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
methods=["TRACE"],
operation_id=operation_id,
response_model_include=response_model_include,
response_model_exclude=response_model_exclude,
response_model_by_alias=response_model_by_alias,
response_model_skip_defaults=response_model_skip_defaults,
include_in_schema=include_in_schema,
response_class=response_class,
name=name,
)
PK |NaD D fastapi/utils.pyimport re
from typing import Any, Dict, List, Sequence, Set, Type
from fastapi import routing
from fastapi.openapi.constants import REF_PREFIX
from pydantic import BaseModel
from pydantic.fields import Field
from pydantic.schema import get_flat_models_from_fields, model_process_schema
from starlette.routing import BaseRoute
def get_flat_models_from_routes(
routes: Sequence[Type[BaseRoute]]
) -> Set[Type[BaseModel]]:
body_fields_from_routes: List[Field] = []
responses_from_routes: List[Field] = []
for route in routes:
if getattr(route, "include_in_schema", None) and isinstance(
route, routing.APIRoute
):
if route.body_field:
assert isinstance(
route.body_field, Field
), "A request body must be a Pydantic Field"
body_fields_from_routes.append(route.body_field)
if route.response_field:
responses_from_routes.append(route.response_field)
if route.response_fields:
responses_from_routes.extend(route.response_fields.values())
flat_models = get_flat_models_from_fields(
body_fields_from_routes + responses_from_routes
)
return flat_models
def get_model_definitions(
*, flat_models: Set[Type[BaseModel]], model_name_map: Dict[Type[BaseModel], str]
) -> Dict[str, Any]:
definitions: Dict[str, Dict] = {}
for model in flat_models:
m_schema, m_definitions = model_process_schema(
model, model_name_map=model_name_map, ref_prefix=REF_PREFIX
)
definitions.update(m_definitions)
model_name = model_name_map[model]
definitions[model_name] = m_schema
return definitions
def get_path_param_names(path: str) -> Set[str]:
return {item.strip("{}") for item in re.findall("{[^}]*}", path)}
PK |N fastapi/dependencies/__init__.pyPK |N u 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,
background_tasks_param_name: str = None,
security_scopes_param_name: str = None,
security_scopes: List[str] = None,
) -> None:
self.path_params = path_params or []
self.query_params = query_params or []
self.header_params = header_params or []
self.cookie_params = cookie_params or []
self.body_params = body_params or []
self.dependencies = dependencies or []
self.security_requirements = security_schemes or []
self.request_param_name = request_param_name
self.websocket_param_name = websocket_param_name
self.background_tasks_param_name = background_tasks_param_name
self.security_scopes = security_scopes
self.security_scopes_param_name = security_scopes_param_name
self.name = name
self.call = call
PK |N+0H H fastapi/dependencies/utils.pyimport asyncio
import inspect
from copy import deepcopy
from datetime import date, datetime, time, timedelta
from decimal import Decimal
from typing import (
Any,
Callable,
Dict,
List,
Mapping,
Optional,
Sequence,
Tuple,
Type,
Union,
)
from uuid import UUID
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, Schema, create_model
from pydantic.error_wrappers import ErrorWrapper
from pydantic.errors import MissingError
from pydantic.fields import Field, Required, Shape
from pydantic.schema import get_annotation_from_schema
from pydantic.utils import lenient_issubclass
from starlette.background import BackgroundTasks
from starlette.concurrency import run_in_threadpool
from starlette.datastructures import FormData, Headers, QueryParams, UploadFile
from starlette.requests import Request
from starlette.websockets import WebSocket
param_supported_types = (
str,
int,
float,
bool,
UUID,
date,
datetime,
time,
timedelta,
Decimal,
)
sequence_shapes = {Shape.LIST, Shape.SET, Shape.TUPLE}
sequence_types = (list, set, tuple)
sequence_shape_to_type = {Shape.LIST: list, Shape.SET: set, Shape.TUPLE: tuple}
def get_param_sub_dependant(
*, param: inspect.Parameter, path: str, security_scopes: List[str] = None
) -> Dependant:
depends: params.Depends = param.default
if depends.dependency:
dependency = depends.dependency
else:
dependency = param.annotation
return get_sub_dependant(
depends=depends,
dependency=dependency,
path=path,
name=param.name,
security_scopes=security_scopes,
)
def get_parameterless_sub_dependant(*, depends: params.Depends, path: str) -> Dependant:
assert callable(
depends.dependency
), "A parameter-less dependency must have a callable dependency"
return get_sub_dependant(depends=depends, dependency=depends.dependency, path=path)
def get_sub_dependant(
*,
depends: params.Depends,
dependency: Callable,
path: str,
name: str = None,
security_scopes: List[str] = None,
) -> Dependant:
security_requirement = None
security_scopes = security_scopes or []
if isinstance(depends, params.Security):
dependency_scopes = depends.scopes
security_scopes.extend(dependency_scopes)
if isinstance(dependency, SecurityBase):
use_scopes: List[str] = []
if isinstance(dependency, (OAuth2, OpenIdConnect)):
use_scopes = security_scopes
security_requirement = SecurityRequirement(
security_scheme=dependency, scopes=use_scopes
)
sub_dependant = get_dependant(
path=path, call=dependency, name=name, security_scopes=security_scopes
)
if security_requirement:
sub_dependant.security_requirements.append(security_requirement)
sub_dependant.security_scopes = security_scopes
return sub_dependant
def get_flat_dependant(dependant: Dependant) -> Dependant:
flat_dependant = Dependant(
path_params=dependant.path_params.copy(),
query_params=dependant.query_params.copy(),
header_params=dependant.header_params.copy(),
cookie_params=dependant.cookie_params.copy(),
body_params=dependant.body_params.copy(),
security_schemes=dependant.security_requirements.copy(),
)
for sub_dependant in dependant.dependencies:
flat_sub = get_flat_dependant(sub_dependant)
flat_dependant.path_params.extend(flat_sub.path_params)
flat_dependant.query_params.extend(flat_sub.query_params)
flat_dependant.header_params.extend(flat_sub.header_params)
flat_dependant.cookie_params.extend(flat_sub.cookie_params)
flat_dependant.body_params.extend(flat_sub.body_params)
flat_dependant.security_requirements.extend(flat_sub.security_requirements)
return flat_dependant
def get_dependant(
*, path: str, call: Callable, name: str = None, security_scopes: List[str] = None
) -> Dependant:
path_param_names = get_path_param_names(path)
endpoint_signature = inspect.signature(call)
signature_params = endpoint_signature.parameters
dependant = Dependant(call=call, name=name)
for param_name in signature_params:
param = signature_params[param_name]
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 in signature_params:
param = signature_params[param_name]
if (
(param.default == param.empty) or isinstance(param.default, params.Path)
) and (param_name in path_param_names):
assert (
lenient_issubclass(param.annotation, param_supported_types)
or param.annotation == param.empty
), f"Path params must be of one of the supported types"
add_param_to_fields(
param=param,
dependant=dependant,
default_schema=params.Path,
force_type=params.ParamTypes.path,
)
elif (
param.default == param.empty
or param.default is None
or isinstance(param.default, param_supported_types)
) and (
param.annotation == param.empty
or lenient_issubclass(param.annotation, param_supported_types)
):
add_param_to_fields(
param=param, dependant=dependant, default_schema=params.Query
)
elif isinstance(param.default, params.Param):
if param.annotation != param.empty:
origin = getattr(param.annotation, "__origin__", None)
param_all_types = param_supported_types + (list, tuple, set)
if isinstance(param.default, (params.Query, params.Header)):
assert lenient_issubclass(
param.annotation, param_all_types
) or lenient_issubclass(
origin, param_all_types
), f"Parameters for Query and Header must be of type str, int, float, bool, list, tuple or set: {param}"
else:
assert lenient_issubclass(
param.annotation, param_supported_types
), f"Parameters for Path and Cookies must be of type str, int, float, bool: {param}"
add_param_to_fields(
param=param, dependant=dependant, default_schema=params.Query
)
elif lenient_issubclass(param.annotation, Request):
dependant.request_param_name = param_name
elif lenient_issubclass(param.annotation, WebSocket):
dependant.websocket_param_name = param_name
elif lenient_issubclass(param.annotation, BackgroundTasks):
dependant.background_tasks_param_name = param_name
elif lenient_issubclass(param.annotation, SecurityScopes):
dependant.security_scopes_param_name = param_name
elif not isinstance(param.default, params.Depends):
add_param_to_body_fields(param=param, dependant=dependant)
return dependant
def add_param_to_fields(
*,
param: inspect.Parameter,
dependant: Dependant,
default_schema: Type[Schema] = params.Param,
force_type: params.ParamTypes = None,
) -> None:
default_value = Required
if not param.default == param.empty:
default_value = param.default
if isinstance(default_value, params.Param):
schema = default_value
default_value = schema.default
if getattr(schema, "in_", None) is None:
schema.in_ = default_schema.in_
if force_type:
schema.in_ = force_type
else:
schema = default_schema(default_value)
required = default_value == Required
annotation: Any = Any
if not param.annotation == param.empty:
annotation = param.annotation
annotation = get_annotation_from_schema(annotation, schema)
if not schema.alias and getattr(schema, "convert_underscores", None):
alias = param.name.replace("_", "-")
else:
alias = schema.alias or param.name
field = Field(
name=param.name,
type_=annotation,
default=None if required else default_value,
alias=alias,
required=required,
model_config=BaseConfig,
class_validators={},
schema=schema,
)
if schema.in_ == params.ParamTypes.path:
dependant.path_params.append(field)
elif schema.in_ == params.ParamTypes.query:
dependant.query_params.append(field)
elif schema.in_ == params.ParamTypes.header:
dependant.header_params.append(field)
else:
assert (
schema.in_ == params.ParamTypes.cookie
), f"non-body parameters must be in path, query, header or cookie: {param.name}"
dependant.cookie_params.append(field)
def add_param_to_body_fields(*, param: inspect.Parameter, dependant: Dependant) -> None:
default_value = Required
if not param.default == param.empty:
default_value = param.default
if isinstance(default_value, Schema):
schema = default_value
default_value = schema.default
else:
schema = Schema(default_value)
required = default_value == Required
annotation = get_annotation_from_schema(param.annotation, schema)
field = Field(
name=param.name,
type_=annotation,
default=None if required else default_value,
alias=schema.alias or param.name,
required=required,
model_config=BaseConfig,
class_validators={},
schema=schema,
)
dependant.body_params.append(field)
def is_coroutine_callable(call: Callable) -> bool:
if inspect.isfunction(call):
return asyncio.iscoroutinefunction(call)
if inspect.isclass(call):
return False
call = getattr(call, "__call__", None)
return asyncio.iscoroutinefunction(call)
async def solve_dependencies(
*,
request: Union[Request, WebSocket],
dependant: Dependant,
body: Dict[str, Any] = None,
background_tasks: BackgroundTasks = None,
) -> Tuple[Dict[str, Any], List[ErrorWrapper], Optional[BackgroundTasks]]:
values: Dict[str, Any] = {}
errors: List[ErrorWrapper] = []
for sub_dependant in dependant.dependencies:
sub_values, sub_errors, background_tasks = await solve_dependencies(
request=request,
dependant=sub_dependant,
body=body,
background_tasks=background_tasks,
)
if sub_errors:
errors.extend(sub_errors)
continue
assert sub_dependant.call is not None, "sub_dependant.call must be a function"
if is_coroutine_callable(sub_dependant.call):
solved = await sub_dependant.call(**sub_values)
else:
solved = await run_in_threadpool(sub_dependant.call, **sub_values)
if sub_dependant.name is not None:
values[sub_dependant.name] = solved
path_values, path_errors = request_params_to_args(
dependant.path_params, request.path_params
)
query_values, query_errors = request_params_to_args(
dependant.query_params, request.query_params
)
header_values, header_errors = request_params_to_args(
dependant.header_params, request.headers
)
cookie_values, cookie_errors = request_params_to_args(
dependant.cookie_params, request.cookies
)
values.update(path_values)
values.update(query_values)
values.update(header_values)
values.update(cookie_values)
errors += path_errors + query_errors + header_errors + cookie_errors
if dependant.body_params:
body_values, body_errors = await request_body_to_args( # type: ignore # body_params checked above
dependant.body_params, body
)
values.update(body_values)
errors.extend(body_errors)
if dependant.request_param_name and isinstance(request, Request):
values[dependant.request_param_name] = request
elif dependant.websocket_param_name and isinstance(request, WebSocket):
values[dependant.websocket_param_name] = request
if dependant.background_tasks_param_name:
if background_tasks is None:
background_tasks = BackgroundTasks()
values[dependant.background_tasks_param_name] = background_tasks
if dependant.security_scopes_param_name:
values[dependant.security_scopes_param_name] = SecurityScopes(
scopes=dependant.security_scopes
)
return values, errors, background_tasks
def request_params_to_args(
required_params: Sequence[Field],
received_params: Union[Mapping[str, Any], QueryParams, Headers],
) -> Tuple[Dict[str, Any], List[ErrorWrapper]]:
values = {}
errors = []
for field in required_params:
if field.shape in sequence_shapes and isinstance(
received_params, (QueryParams, Headers)
):
value = received_params.getlist(field.alias)
else:
value = received_params.get(field.alias)
schema: params.Param = field.schema
assert isinstance(schema, params.Param), "Params must be subclasses of Param"
if value is None:
if field.required:
errors.append(
ErrorWrapper(
MissingError(),
loc=(schema.in_.value, field.alias),
config=BaseConfig,
)
)
else:
values[field.name] = deepcopy(field.default)
continue
v_, errors_ = field.validate(value, values, loc=(schema.in_.value, field.alias))
if isinstance(errors_, ErrorWrapper):
errors.append(errors_)
elif isinstance(errors_, list):
errors.extend(errors_)
else:
values[field.name] = v_
return values, errors
async def request_body_to_args(
required_params: List[Field], received_body: Dict[str, Any]
) -> Tuple[Dict[str, Any], List[ErrorWrapper]]:
values = {}
errors = []
if required_params:
field = required_params[0]
embed = getattr(field.schema, "embed", None)
if len(required_params) == 1 and not embed:
received_body = {field.alias: received_body}
for field in required_params:
if field.shape in sequence_shapes and isinstance(received_body, FormData):
value = received_body.getlist(field.alias)
else:
value = received_body.get(field.alias)
if (
value is None
or (isinstance(field.schema, params.Form) and value == "")
or (
isinstance(field.schema, params.Form)
and field.shape in sequence_shapes
and len(value) == 0
)
):
if field.required:
errors.append(
ErrorWrapper(
MissingError(), loc=("body", field.alias), config=BaseConfig
)
)
else:
values[field.name] = deepcopy(field.default)
continue
if (
isinstance(field.schema, params.File)
and lenient_issubclass(field.type_, bytes)
and isinstance(value, UploadFile)
):
value = await value.read()
elif (
field.shape in sequence_shapes
and isinstance(field.schema, params.File)
and lenient_issubclass(field.type_, bytes)
and isinstance(value, sequence_types)
):
awaitables = [sub_value.read() for sub_value in value]
contents = await asyncio.gather(*awaitables)
value = sequence_shape_to_type[field.shape](contents)
v_, errors_ = field.validate(value, values, loc=("body", field.alias))
if isinstance(errors_, ErrorWrapper):
errors.append(errors_)
elif isinstance(errors_, list):
errors.extend(errors_)
else:
values[field.name] = v_
return values, errors
def get_schema_compatible_field(*, field: Field) -> Field:
out_field = field
if lenient_issubclass(field.type_, UploadFile):
use_type: type = bytes
if field.shape in sequence_shapes:
use_type = List[bytes]
out_field = Field(
name=field.name,
type_=use_type,
class_validators=field.class_validators,
model_config=field.model_config,
default=field.default,
required=field.required,
alias=field.alias,
schema=field.schema,
)
return out_field
def get_body_field(*, dependant: Dependant, name: str) -> Optional[Field]:
flat_dependant = get_flat_dependant(dependant)
if not flat_dependant.body_params:
return None
first_param = flat_dependant.body_params[0]
embed = getattr(first_param.schema, "embed", None)
if len(flat_dependant.body_params) == 1 and not embed:
return get_schema_compatible_field(field=first_param)
model_name = "Body_" + name
BodyModel = create_model(model_name)
for f in flat_dependant.body_params:
BodyModel.__fields__[f.name] = get_schema_compatible_field(field=f)
required = any(True for f in flat_dependant.body_params if f.required)
if any(isinstance(f.schema, params.File) for f in flat_dependant.body_params):
BodySchema: Type[params.Body] = params.File
elif any(isinstance(f.schema, params.Form) for f in flat_dependant.body_params):
BodySchema = params.Form
else:
BodySchema = params.Body
field = Field(
name="body",
type_=BodyModel,
default=None,
required=required,
model_config=BaseConfig,
class_validators={},
alias="body",
schema=BodySchema(None),
)
return field
PK |N fastapi/openapi/__init__.pyPK |Nma a fastapi/openapi/constants.pyMETHODS_WITH_BODY = set(("POST", "PUT", "DELETE", "PATCH"))
REF_PREFIX = "#/components/schemas/"
PK |N
j fastapi/openapi/docs.pyfrom typing import Optional
from starlette.responses import HTMLResponse
def get_swagger_ui_html(
*,
openapi_url: str,
title: str,
swagger_js_url: str = "https://cdn.jsdelivr.net/npm/swagger-ui-dist@3/swagger-ui-bundle.js",
swagger_css_url: str = "https://cdn.jsdelivr.net/npm/swagger-ui-dist@3/swagger-ui.css",
swagger_favicon_url: str = "https://fastapi.tiangolo.com/img/favicon.png",
oauth2_redirect_url: Optional[str] = None,
) -> HTMLResponse:
html = f"""
{title}
"""
return HTMLResponse(html)
def get_redoc_html(
*,
openapi_url: str,
title: str,
redoc_js_url: str = "https://cdn.jsdelivr.net/npm/redoc@next/bundles/redoc.standalone.js",
redoc_favicon_url: str = "https://fastapi.tiangolo.com/img/favicon.png",
) -> HTMLResponse:
html = f"""
{title}
"""
return HTMLResponse(html)
def get_swagger_ui_oauth2_redirect_html() -> HTMLResponse:
html = """
"""
return HTMLResponse(content=html)
PK |N#5) ) fastapi/openapi/models.pyimport logging
from enum import Enum
from typing import Any, Dict, List, Optional, Union
from pydantic import BaseModel, Schema as PSchema
from pydantic.types import UrlStr
logger = logging.getLogger("fastapi")
try:
import email_validator
assert email_validator # make autoflake ignore the unused import
from pydantic.types import EmailStr # type: ignore
except ImportError: # pragma: no cover
logger.warning(
"email-validator not installed, email fields will be treated as str.\n"
+ "To install, run: pip install email-validator"
)
class EmailStr(str): # type: ignore
pass
class Contact(BaseModel):
name: Optional[str] = None
url: Optional[UrlStr] = None
email: Optional[EmailStr] = None
class License(BaseModel):
name: str
url: Optional[UrlStr] = None
class Info(BaseModel):
title: str
description: Optional[str] = None
termsOfService: Optional[str] = None
contact: Optional[Contact] = None
license: Optional[License] = None
version: str
class ServerVariable(BaseModel):
enum: Optional[List[str]] = None
default: str
description: Optional[str] = None
class Server(BaseModel):
url: UrlStr
description: Optional[str] = None
variables: Optional[Dict[str, ServerVariable]] = None
class Reference(BaseModel):
ref: str = PSchema(..., alias="$ref") # type: ignore
class Discriminator(BaseModel):
propertyName: str
mapping: Optional[Dict[str, str]] = None
class XML(BaseModel):
name: Optional[str] = None
namespace: Optional[str] = None
prefix: Optional[str] = None
attribute: Optional[bool] = None
wrapped: Optional[bool] = None
class ExternalDocumentation(BaseModel):
description: Optional[str] = None
url: UrlStr
class SchemaBase(BaseModel):
ref: Optional[str] = PSchema(None, alias="$ref") # type: ignore
title: Optional[str] = None
multipleOf: Optional[float] = None
maximum: Optional[float] = None
exclusiveMaximum: Optional[float] = None
minimum: Optional[float] = None
exclusiveMinimum: Optional[float] = None
maxLength: Optional[int] = PSchema(None, gte=0) # type: ignore
minLength: Optional[int] = PSchema(None, gte=0) # type: ignore
pattern: Optional[str] = None
maxItems: Optional[int] = PSchema(None, gte=0) # type: ignore
minItems: Optional[int] = PSchema(None, gte=0) # type: ignore
uniqueItems: Optional[bool] = None
maxProperties: Optional[int] = PSchema(None, gte=0) # type: ignore
minProperties: Optional[int] = PSchema(None, gte=0) # type: ignore
required: Optional[List[str]] = None
enum: Optional[List[str]] = None
type: Optional[str] = None
allOf: Optional[List[Any]] = None
oneOf: Optional[List[Any]] = None
anyOf: Optional[List[Any]] = None
not_: Optional[List[Any]] = PSchema(None, alias="not") # type: ignore
items: Optional[Any] = None
properties: Optional[Dict[str, Any]] = None
additionalProperties: Optional[Union[Dict[str, Any], bool]] = None
description: Optional[str] = None
format: Optional[str] = None
default: Optional[Any] = None
nullable: Optional[bool] = None
discriminator: Optional[Discriminator] = None
readOnly: Optional[bool] = None
writeOnly: Optional[bool] = None
xml: Optional[XML] = None
externalDocs: Optional[ExternalDocumentation] = None
example: Optional[Any] = None
deprecated: Optional[bool] = None
class Schema(SchemaBase):
allOf: Optional[List[SchemaBase]] = None
oneOf: Optional[List[SchemaBase]] = None
anyOf: Optional[List[SchemaBase]] = None
not_: Optional[List[SchemaBase]] = PSchema(None, alias="not") # type: ignore
items: Optional[SchemaBase] = None
properties: Optional[Dict[str, SchemaBase]] = None
additionalProperties: Optional[Union[SchemaBase, bool]] = None
class Example(BaseModel):
summary: Optional[str] = None
description: Optional[str] = None
value: Optional[Any] = None
externalValue: Optional[UrlStr] = None
class ParameterInType(Enum):
query = "query"
header = "header"
path = "path"
cookie = "cookie"
class Encoding(BaseModel):
contentType: Optional[str] = None
# Workaround OpenAPI recursive reference, using Any
headers: Optional[Dict[str, Union[Any, Reference]]] = None
style: Optional[str] = None
explode: Optional[bool] = None
allowReserved: Optional[bool] = None
class MediaType(BaseModel):
schema_: Optional[Union[Schema, Reference]] = PSchema(
None, alias="schema"
) # type: ignore
example: Optional[Any] = None
examples: Optional[Dict[str, Union[Example, Reference]]] = None
encoding: Optional[Dict[str, Encoding]] = None
class ParameterBase(BaseModel):
description: Optional[str] = None
required: Optional[bool] = None
deprecated: Optional[bool] = None
# Serialization rules for simple scenarios
style: Optional[str] = None
explode: Optional[bool] = None
allowReserved: Optional[bool] = None
schema_: Optional[Union[Schema, Reference]] = PSchema(
None, alias="schema"
) # type: ignore
example: Optional[Any] = None
examples: Optional[Dict[str, Union[Example, Reference]]] = None
# Serialization rules for more complex scenarios
content: Optional[Dict[str, MediaType]] = None
class Parameter(ParameterBase):
name: str
in_: ParameterInType = PSchema(..., alias="in") # type: ignore
class Header(ParameterBase):
pass
# Workaround OpenAPI recursive reference
class EncodingWithHeaders(Encoding):
headers: Optional[Dict[str, Union[Header, Reference]]] = None
class RequestBody(BaseModel):
description: Optional[str] = None
content: Dict[str, MediaType]
required: Optional[bool] = None
class Link(BaseModel):
operationRef: Optional[str] = None
operationId: Optional[str] = None
parameters: Optional[Dict[str, Union[Any, str]]] = None
requestBody: Optional[Union[Any, str]] = None
description: Optional[str] = None
server: Optional[Server] = None
class Response(BaseModel):
description: str
headers: Optional[Dict[str, Union[Header, Reference]]] = None
content: Optional[Dict[str, MediaType]] = None
links: Optional[Dict[str, Union[Link, Reference]]] = None
class Responses(BaseModel):
default: Response
class Operation(BaseModel):
tags: Optional[List[str]] = None
summary: Optional[str] = None
description: Optional[str] = None
externalDocs: Optional[ExternalDocumentation] = None
operationId: Optional[str] = None
parameters: Optional[List[Union[Parameter, Reference]]] = None
requestBody: Optional[Union[RequestBody, Reference]] = None
responses: Union[Responses, Dict[str, Response]]
# Workaround OpenAPI recursive reference
callbacks: Optional[Dict[str, Union[Dict[str, Any], Reference]]] = None
deprecated: Optional[bool] = None
security: Optional[List[Dict[str, List[str]]]] = None
servers: Optional[List[Server]] = None
class PathItem(BaseModel):
ref: Optional[str] = PSchema(None, alias="$ref") # type: ignore
summary: Optional[str] = None
description: Optional[str] = None
get: Optional[Operation] = None
put: Optional[Operation] = None
post: Optional[Operation] = None
delete: Optional[Operation] = None
options: Optional[Operation] = None
head: Optional[Operation] = None
patch: Optional[Operation] = None
trace: Optional[Operation] = None
servers: Optional[List[Server]] = None
parameters: Optional[List[Union[Parameter, Reference]]] = None
# Workaround OpenAPI recursive reference
class OperationWithCallbacks(BaseModel):
callbacks: Optional[Dict[str, Union[Dict[str, PathItem], Reference]]] = None
class SecuritySchemeType(Enum):
apiKey = "apiKey"
http = "http"
oauth2 = "oauth2"
openIdConnect = "openIdConnect"
class SecurityBase(BaseModel):
type_: SecuritySchemeType = PSchema(..., alias="type") # type: ignore
description: Optional[str] = None
class APIKeyIn(Enum):
query = "query"
header = "header"
cookie = "cookie"
class APIKey(SecurityBase):
type_ = PSchema(SecuritySchemeType.apiKey, alias="type") # type: ignore
in_: APIKeyIn = PSchema(..., alias="in") # type: ignore
name: str
class HTTPBase(SecurityBase):
type_ = PSchema(SecuritySchemeType.http, alias="type") # type: ignore
scheme: str
class HTTPBearer(HTTPBase):
scheme = "bearer"
bearerFormat: Optional[str] = None
class OAuthFlow(BaseModel):
refreshUrl: Optional[str] = None
scopes: Dict[str, str] = {}
class OAuthFlowImplicit(OAuthFlow):
authorizationUrl: str
class OAuthFlowPassword(OAuthFlow):
tokenUrl: str
class OAuthFlowClientCredentials(OAuthFlow):
tokenUrl: str
class OAuthFlowAuthorizationCode(OAuthFlow):
authorizationUrl: str
tokenUrl: str
class OAuthFlows(BaseModel):
implicit: Optional[OAuthFlowImplicit] = None
password: Optional[OAuthFlowPassword] = None
clientCredentials: Optional[OAuthFlowClientCredentials] = None
authorizationCode: Optional[OAuthFlowAuthorizationCode] = None
class OAuth2(SecurityBase):
type_ = PSchema(SecuritySchemeType.oauth2, alias="type") # type: ignore
flows: OAuthFlows
class OpenIdConnect(SecurityBase):
type_ = PSchema(SecuritySchemeType.openIdConnect, alias="type") # type: ignore
openIdConnectUrl: str
SecurityScheme = Union[APIKey, HTTPBase, OAuth2, OpenIdConnect, HTTPBearer]
class Components(BaseModel):
schemas: Optional[Dict[str, Union[Schema, Reference]]] = None
responses: Optional[Dict[str, Union[Response, Reference]]] = None
parameters: Optional[Dict[str, Union[Parameter, Reference]]] = None
examples: Optional[Dict[str, Union[Example, Reference]]] = None
requestBodies: Optional[Dict[str, Union[RequestBody, Reference]]] = None
headers: Optional[Dict[str, Union[Header, Reference]]] = None
securitySchemes: Optional[Dict[str, Union[SecurityScheme, Reference]]] = None
links: Optional[Dict[str, Union[Link, Reference]]] = None
callbacks: Optional[Dict[str, Union[Dict[str, PathItem], Reference]]] = None
class Tag(BaseModel):
name: str
description: Optional[str] = None
externalDocs: Optional[ExternalDocumentation] = None
class OpenAPI(BaseModel):
openapi: str
info: Info
servers: Optional[List[Server]] = None
paths: Dict[str, PathItem]
components: Optional[Components] = None
security: Optional[List[Dict[str, List[str]]]] = None
tags: Optional[List[Tag]] = None
externalDocs: Optional[ExternalDocumentation] = None
PK |Ny* * fastapi/openapi/utils.pyfrom typing import Any, Dict, List, Optional, Sequence, Tuple, Type
from fastapi import routing
from fastapi.dependencies.models import Dependant
from fastapi.dependencies.utils import get_flat_dependant
from fastapi.encoders import jsonable_encoder
from fastapi.openapi.constants import METHODS_WITH_BODY, REF_PREFIX
from fastapi.openapi.models import OpenAPI
from fastapi.params import Body, Param
from fastapi.utils import get_flat_models_from_routes, get_model_definitions
from pydantic.fields import Field
from pydantic.schema import Schema, field_schema, get_model_name_map
from pydantic.utils import lenient_issubclass
from starlette.responses import JSONResponse
from starlette.routing import BaseRoute
from starlette.status import HTTP_422_UNPROCESSABLE_ENTITY
validation_error_definition = {
"title": "ValidationError",
"type": "object",
"properties": {
"loc": {"title": "Location", "type": "array", "items": {"type": "string"}},
"msg": {"title": "Message", "type": "string"},
"type": {"title": "Error Type", "type": "string"},
},
"required": ["loc", "msg", "type"],
}
validation_error_response_definition = {
"title": "HTTPValidationError",
"type": "object",
"properties": {
"detail": {
"title": "Detail",
"type": "array",
"items": {"$ref": REF_PREFIX + "ValidationError"},
}
},
}
def get_openapi_params(dependant: Dependant) -> List[Field]:
flat_dependant = get_flat_dependant(dependant)
return (
flat_dependant.path_params
+ flat_dependant.query_params
+ flat_dependant.header_params
+ flat_dependant.cookie_params
)
def get_openapi_security_definitions(flat_dependant: Dependant) -> Tuple[Dict, List]:
security_definitions = {}
operation_security = []
for security_requirement in flat_dependant.security_requirements:
security_definition = jsonable_encoder(
security_requirement.security_scheme.model,
by_alias=True,
include_none=False,
)
security_name = security_requirement.security_scheme.scheme_name
security_definitions[security_name] = security_definition
operation_security.append({security_name: security_requirement.scopes})
return security_definitions, operation_security
def get_openapi_operation_parameters(
all_route_params: Sequence[Field]
) -> Tuple[Dict[str, Dict], List[Dict[str, Any]]]:
definitions: Dict[str, Dict] = {}
parameters = []
for param in all_route_params:
schema: Param = param.schema
if "ValidationError" not in definitions:
definitions["ValidationError"] = validation_error_definition
definitions["HTTPValidationError"] = validation_error_response_definition
parameter = {
"name": param.alias,
"in": schema.in_.value,
"required": param.required,
"schema": field_schema(param, model_name_map={})[0],
}
if schema.description:
parameter["description"] = schema.description
if schema.deprecated:
parameter["deprecated"] = schema.deprecated
parameters.append(parameter)
return definitions, parameters
def get_openapi_operation_request_body(
*, body_field: Field, model_name_map: Dict[Type, str]
) -> Optional[Dict]:
if not body_field:
return None
assert isinstance(body_field, Field)
body_schema, _ = field_schema(
body_field, model_name_map=model_name_map, ref_prefix=REF_PREFIX
)
schema: Schema = body_field.schema
if isinstance(schema, Body):
request_media_type = schema.media_type
else:
# Includes not declared media types (Schema)
request_media_type = "application/json"
required = body_field.required
request_body_oai: Dict[str, Any] = {}
if required:
request_body_oai["required"] = required
request_body_oai["content"] = {request_media_type: {"schema": body_schema}}
return request_body_oai
def generate_operation_id(*, route: routing.APIRoute, method: str) -> str:
if route.operation_id:
return route.operation_id
path: str = route.path_format
operation_id = route.name + path
operation_id = operation_id.replace("{", "_").replace("}", "_").replace("/", "_")
operation_id = operation_id + "_" + method.lower()
return operation_id
def generate_operation_summary(*, route: routing.APIRoute, method: str) -> str:
if route.summary:
return route.summary
return route.name.replace("_", " ").title()
def get_openapi_operation_metadata(*, route: routing.APIRoute, method: str) -> Dict:
operation: Dict[str, Any] = {}
if route.tags:
operation["tags"] = route.tags
operation["summary"] = generate_operation_summary(route=route, method=method)
if route.description:
operation["description"] = route.description
operation["operationId"] = generate_operation_id(route=route, method=method)
if route.deprecated:
operation["deprecated"] = route.deprecated
return operation
def get_openapi_path(
*, route: routing.APIRoute, model_name_map: Dict[Type, str]
) -> Tuple[Dict, Dict, Dict]:
path = {}
security_schemes: Dict[str, Any] = {}
definitions: Dict[str, Any] = {}
assert route.methods is not None, "Methods must be a list"
if route.include_in_schema:
for method in route.methods:
operation = get_openapi_operation_metadata(route=route, method=method)
parameters: List[Dict] = []
flat_dependant = get_flat_dependant(route.dependant)
security_definitions, operation_security = get_openapi_security_definitions(
flat_dependant=flat_dependant
)
if operation_security:
operation.setdefault("security", []).extend(operation_security)
if security_definitions:
security_schemes.update(security_definitions)
all_route_params = get_openapi_params(route.dependant)
validation_definitions, operation_parameters = get_openapi_operation_parameters(
all_route_params=all_route_params
)
definitions.update(validation_definitions)
parameters.extend(operation_parameters)
if parameters:
operation["parameters"] = parameters
if method in METHODS_WITH_BODY:
request_body_oai = get_openapi_operation_request_body(
body_field=route.body_field, model_name_map=model_name_map
)
if request_body_oai:
operation["requestBody"] = request_body_oai
if "ValidationError" not in definitions:
definitions["ValidationError"] = validation_error_definition
definitions[
"HTTPValidationError"
] = validation_error_response_definition
if route.responses:
for (additional_status_code, response) in route.responses.items():
assert isinstance(
response, dict
), "An additional response must be a dict"
field = route.response_fields.get(additional_status_code)
if field:
response_schema, _ = field_schema(
field, model_name_map=model_name_map, ref_prefix=REF_PREFIX
)
response.setdefault("content", {}).setdefault(
"application/json", {}
)["schema"] = response_schema
response.setdefault("description", "Additional Response")
operation.setdefault("responses", {})[
str(additional_status_code)
] = response
status_code = str(route.status_code)
response_schema = {"type": "string"}
if lenient_issubclass(route.response_class, JSONResponse):
if route.response_field:
response_schema, _ = field_schema(
route.response_field,
model_name_map=model_name_map,
ref_prefix=REF_PREFIX,
)
else:
response_schema = {}
operation.setdefault("responses", {}).setdefault(status_code, {})[
"description"
] = route.response_description
operation.setdefault("responses", {}).setdefault(
status_code, {}
).setdefault("content", {}).setdefault(route.response_class.media_type, {})[
"schema"
] = response_schema
if all_route_params or route.body_field:
operation["responses"][str(HTTP_422_UNPROCESSABLE_ENTITY)] = {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {"$ref": REF_PREFIX + "HTTPValidationError"}
}
},
}
path[method.lower()] = operation
return path, security_schemes, definitions
def get_openapi(
*,
title: str,
version: str,
openapi_version: str = "3.0.2",
description: str = None,
routes: Sequence[BaseRoute],
openapi_prefix: str = ""
) -> Dict:
info = {"title": title, "version": version}
if description:
info["description"] = description
output = {"openapi": openapi_version, "info": info}
components: Dict[str, Dict] = {}
paths: Dict[str, Dict] = {}
flat_models = get_flat_models_from_routes(routes)
model_name_map = get_model_name_map(flat_models)
definitions = get_model_definitions(
flat_models=flat_models, model_name_map=model_name_map
)
for route in routes:
if isinstance(route, routing.APIRoute):
result = get_openapi_path(route=route, model_name_map=model_name_map)
if result:
path, security_schemes, path_definitions = result
if path:
paths.setdefault(openapi_prefix + route.path_format, {}).update(
path
)
if security_schemes:
components.setdefault("securitySchemes", {}).update(
security_schemes
)
if path_definitions:
definitions.update(path_definitions)
if definitions:
components.setdefault("schemas", {}).update(definitions)
if components:
output["components"] = components
output["paths"] = paths
return jsonable_encoder(OpenAPI(**output), by_alias=True, include_none=False)
PK |N߸^ ^ 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 |N$+8 8 fastapi/security/api_key.pyfrom typing import Optional
from fastapi.openapi.models import APIKey, APIKeyIn
from fastapi.security.base import SecurityBase
from starlette.exceptions import HTTPException
from starlette.requests import Request
from starlette.status import HTTP_403_FORBIDDEN
class APIKeyBase(SecurityBase):
pass
class APIKeyQuery(APIKeyBase):
def __init__(self, *, name: str, scheme_name: str = None, auto_error: bool = True):
self.model: APIKey = APIKey(**{"in": APIKeyIn.query}, name=name)
self.scheme_name = scheme_name or self.__class__.__name__
self.auto_error = auto_error
async def __call__(self, request: Request) -> Optional[str]:
api_key: str = request.query_params.get(self.model.name)
if not api_key:
if self.auto_error:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
else:
return None
return api_key
class APIKeyHeader(APIKeyBase):
def __init__(self, *, name: str, scheme_name: str = None, auto_error: bool = True):
self.model: APIKey = APIKey(**{"in": APIKeyIn.header}, name=name)
self.scheme_name = scheme_name or self.__class__.__name__
self.auto_error = auto_error
async def __call__(self, request: Request) -> Optional[str]:
api_key: str = request.headers.get(self.model.name)
if not api_key:
if self.auto_error:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
else:
return None
return api_key
class APIKeyCookie(APIKeyBase):
def __init__(self, *, name: str, scheme_name: str = None, auto_error: bool = True):
self.model: APIKey = APIKey(**{"in": APIKeyIn.cookie}, name=name)
self.scheme_name = scheme_name or self.__class__.__name__
self.auto_error = auto_error
async def __call__(self, request: Request) -> Optional[str]:
api_key: str = request.cookies.get(self.model.name)
if not api_key:
if self.auto_error:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
else:
return None
return api_key
PK |Nr(ԍ fastapi/security/base.pyfrom fastapi.openapi.models import SecurityBase as SecurityBaseModel
class SecurityBase:
model: SecurityBaseModel
scheme_name: str
PK |Nx3 fastapi/security/http.pyimport binascii
from base64 import b64decode
from typing import Optional
from fastapi.exceptions import HTTPException
from fastapi.openapi.models import (
HTTPBase as HTTPBaseModel,
HTTPBearer as HTTPBearerModel,
)
from fastapi.security.base import SecurityBase
from fastapi.security.utils import get_authorization_scheme_param
from pydantic import BaseModel
from starlette.requests import Request
from starlette.status import HTTP_401_UNAUTHORIZED, HTTP_403_FORBIDDEN
class HTTPBasicCredentials(BaseModel):
username: str
password: str
class HTTPAuthorizationCredentials(BaseModel):
scheme: str
credentials: str
class HTTPBase(SecurityBase):
def __init__(
self, *, scheme: str, scheme_name: str = None, auto_error: bool = True
):
self.model = HTTPBaseModel(scheme=scheme)
self.scheme_name = scheme_name or self.__class__.__name__
self.auto_error = auto_error
async def __call__(
self, request: Request
) -> Optional[HTTPAuthorizationCredentials]:
authorization: str = request.headers.get("Authorization")
scheme, credentials = get_authorization_scheme_param(authorization)
if not (authorization and scheme and credentials):
if self.auto_error:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
else:
return None
return HTTPAuthorizationCredentials(scheme=scheme, credentials=credentials)
class HTTPBasic(HTTPBase):
def __init__(
self, *, scheme_name: str = None, realm: str = None, auto_error: bool = True
):
self.model = HTTPBaseModel(scheme="basic")
self.scheme_name = scheme_name or self.__class__.__name__
self.realm = realm
self.auto_error = auto_error
async def __call__(self, request: Request) -> Optional[HTTPBasicCredentials]:
authorization: str = request.headers.get("Authorization")
scheme, param = get_authorization_scheme_param(authorization)
if self.realm:
unauthorized_headers = {"WWW-Authenticate": f'Basic realm="{self.realm}"'}
else:
unauthorized_headers = {"WWW-Authenticate": "Basic"}
invalid_user_credentials_exc = HTTPException(
status_code=HTTP_401_UNAUTHORIZED,
detail="Invalid authentication credentials",
headers=unauthorized_headers,
)
if not authorization or scheme.lower() != "basic":
if self.auto_error:
raise HTTPException(
status_code=HTTP_401_UNAUTHORIZED,
detail="Not authenticated",
headers=unauthorized_headers,
)
else:
return None
try:
data = b64decode(param).decode("ascii")
except (ValueError, UnicodeDecodeError, binascii.Error):
raise invalid_user_credentials_exc
username, separator, password = data.partition(":")
if not (separator):
raise invalid_user_credentials_exc
return HTTPBasicCredentials(username=username, password=password)
class HTTPBearer(HTTPBase):
def __init__(
self,
*,
bearerFormat: str = None,
scheme_name: str = None,
auto_error: bool = True,
):
self.model = HTTPBearerModel(bearerFormat=bearerFormat)
self.scheme_name = scheme_name or self.__class__.__name__
self.auto_error = auto_error
async def __call__(
self, request: Request
) -> Optional[HTTPAuthorizationCredentials]:
authorization: str = request.headers.get("Authorization")
scheme, credentials = get_authorization_scheme_param(authorization)
if not (authorization and scheme and credentials):
if self.auto_error:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
else:
return None
if scheme.lower() != "bearer":
raise HTTPException(
status_code=HTTP_403_FORBIDDEN,
detail="Invalid authentication credentials",
)
return HTTPAuthorizationCredentials(scheme=scheme, credentials=credentials)
class HTTPDigest(HTTPBase):
def __init__(self, *, scheme_name: str = None, auto_error: bool = True):
self.model = HTTPBaseModel(scheme="digest")
self.scheme_name = scheme_name or self.__class__.__name__
self.auto_error = auto_error
async def __call__(
self, request: Request
) -> Optional[HTTPAuthorizationCredentials]:
authorization: str = request.headers.get("Authorization")
scheme, credentials = get_authorization_scheme_param(authorization)
if not (authorization and scheme and credentials):
if self.auto_error:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
else:
return None
if scheme.lower() != "digest":
raise HTTPException(
status_code=HTTP_403_FORBIDDEN,
detail="Invalid authentication credentials",
)
return HTTPAuthorizationCredentials(scheme=scheme, credentials=credentials)
PK |N% fastapi/security/oauth2.pyfrom typing import List, Optional
from fastapi.exceptions import HTTPException
from fastapi.openapi.models import OAuth2 as OAuth2Model, OAuthFlows as OAuthFlowsModel
from fastapi.params import Form
from fastapi.security.base import SecurityBase
from fastapi.security.utils import get_authorization_scheme_param
from starlette.requests import Request
from starlette.status import HTTP_401_UNAUTHORIZED, HTTP_403_FORBIDDEN
class OAuth2PasswordRequestForm:
"""
This is a dependency class, use it like:
@app.post("/login")
def login(form_data: Oauth2PasswordRequestForm = Depends()):
data = form_data.parse()
print(data.username)
print(data.password)
for scope in data.scopes:
print(scope)
if data.client_id:
print(data.client_id)
if data.client_secret:
print(data.client_secret)
return data
It creates the following Form request parameters in your endpoint:
grant_type: the OAuth2 spec says it is required and MUST be the fixed string "password".
Nevertheless, this dependency class is permissive and allows not passing it. If you want to enforce it,
use instead the OAuth2PasswordRequestFormStrict dependency.
username: username string. The OAuth2 spec requires the exact field name "username".
password: password string. The OAuth2 spec requires the exact field name "password".
scope: Optional string. Several scopes (each one a string) separated by spaces. E.g.
"items:read items:write users:read profile openid"
client_id: optional string. OAuth2 recommends sending the client_id and client_secret (if any)
using HTTP Basic auth, as: client_id:client_secret
client_secret: optional string. OAuth2 recommends sending the client_id and client_secret (if any)
using HTTP Basic auth, as: client_id:client_secret
"""
def __init__(
self,
grant_type: str = Form(None, regex="password"),
username: str = Form(...),
password: str = Form(...),
scope: str = Form(""),
client_id: Optional[str] = Form(None),
client_secret: Optional[str] = Form(None),
):
self.grant_type = grant_type
self.username = username
self.password = password
self.scopes = scope.split()
self.client_id = client_id
self.client_secret = client_secret
class OAuth2PasswordRequestFormStrict(OAuth2PasswordRequestForm):
"""
This is a dependency class, use it like:
@app.post("/login")
def login(form_data: Oauth2PasswordRequestFormStrict = Depends()):
data = form_data.parse()
print(data.username)
print(data.password)
for scope in data.scopes:
print(scope)
if data.client_id:
print(data.client_id)
if data.client_secret:
print(data.client_secret)
return data
It creates the following Form request parameters in your endpoint:
grant_type: the OAuth2 spec says it is required and MUST be the fixed string "password".
This dependency is strict about it. If you want to be permissive, use instead the
OAuth2PasswordRequestFormStrict dependency class.
username: username string. The OAuth2 spec requires the exact field name "username".
password: password string. The OAuth2 spec requires the exact field name "password".
scope: Optional string. Several scopes (each one a string) separated by spaces. E.g.
"items:read items:write users:read profile openid"
client_id: optional string. OAuth2 recommends sending the client_id and client_secret (if any)
using HTTP Basic auth, as: client_id:client_secret
client_secret: optional string. OAuth2 recommends sending the client_id and client_secret (if any)
using HTTP Basic auth, as: client_id:client_secret
"""
def __init__(
self,
grant_type: str = Form(..., regex="password"),
username: str = Form(...),
password: str = Form(...),
scope: str = Form(""),
client_id: Optional[str] = Form(None),
client_secret: Optional[str] = Form(None),
):
super().__init__(
grant_type=grant_type,
username=username,
password=password,
scope=scope,
client_id=client_id,
client_secret=client_secret,
)
class OAuth2(SecurityBase):
def __init__(
self,
*,
flows: OAuthFlowsModel = OAuthFlowsModel(),
scheme_name: str = None,
auto_error: bool = True
):
self.model = OAuth2Model(flows=flows)
self.scheme_name = scheme_name or self.__class__.__name__
self.auto_error = auto_error
async def __call__(self, request: Request) -> Optional[str]:
authorization: str = request.headers.get("Authorization")
if not authorization:
if self.auto_error:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
else:
return None
return authorization
class OAuth2PasswordBearer(OAuth2):
def __init__(
self,
tokenUrl: str,
scheme_name: str = None,
scopes: dict = None,
auto_error: bool = True,
):
if not scopes:
scopes = {}
flows = OAuthFlowsModel(password={"tokenUrl": tokenUrl, "scopes": scopes})
super().__init__(flows=flows, scheme_name=scheme_name, auto_error=auto_error)
async def __call__(self, request: Request) -> Optional[str]:
authorization: str = request.headers.get("Authorization")
scheme, param = get_authorization_scheme_param(authorization)
if not authorization or scheme.lower() != "bearer":
if self.auto_error:
raise HTTPException(
status_code=HTTP_401_UNAUTHORIZED,
detail="Not authenticated",
headers={"WWW-Authenticate": "Bearer"},
)
else:
return None
return param
class SecurityScopes:
def __init__(self, scopes: List[str] = None):
self.scopes = scopes or []
self.scope_str = " ".join(self.scopes)
PK |N ' 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 |N
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 |N{> > fastapi-0.26.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 !HP O fastapi-0.26.0.dist-info/WHEEL HM
K-*ϳR03rOK-J,/RH,szd&Y)r$[)T&Ur PK !HLa C ! fastapi-0.26.0.dist-info/METADATA\rF&Jl9ZI^[my-y][.Cr,܌D3[8?S쓜{);VT*"LOO_i,/|2*MŎc/F.L9-o羷圕q}#ߜQi_D'"(c? dOH_DHyA2HC<̥ Eay1XE*& Ҹ_(?Qگ{h{㲘8Co"˟3WƾE5֟tM;oRRofĢz'Rʈ?q2)YY C a:MǑ)-cb_ k!]:9 ϽD^(b /
(5Q
WTTZra\Qu} 4str7H֒ƳlhZ*y\J|PsGDC1DxIVdPMOW1*pD7TҐQpx/
/sa5N;1PΑ#ā/& vVq9:PXO
[n橵JD1"WT q]q9!,,i5ϊX1tlL+ x ܈1N6{b*a1$$ȉAb
p(
\a$zZ J
99RI2sn# -qp h/κ]p,ύbhr{g5Xo, ćDtjCk',`W 0a$?nik8+q|hhTIOCR4K ΄?LBlm(߮gƔSI2 k|T`yR VʦP QBykOty$RY{
E2 j7 H T\&P"܉3RK{2(Rgu OgH|6\z K C0 >E"" FÍ[+.
Z"PPEMV1~cflr;9=>EPg -KFCldRH}L$gS<&
$F&Ŝ!nqgS uQVPH5T?+@WqO?A{}3ܹ#N3h :NG ȿ@#9@)C+kOa] iW)xFϘ5~Rh* W@m<剳4|SL~Q@hJ( LMKAHժyK4힎DzyngFOE \$ |/OP'W]i2>y\͝
g
߹ V60fg'}p^ENXb)I~ 8>z h`Y Q&R%QDR̐!I:o;shyp̝`!=PɅ{c#m4KÀB$ի6\D1ooo 2wwyk l6F4dS~ O $փ%lxIhoJmkHU3[KR% '~6pⴘc(}sCԛcX=aDȩfخJ䗅 p{v/я[m^fE [p g2eQCf; .Rg
X 0iTлIJ sLdAYP dQB8Ihqp`3qKd*#gu O oih {
gNع˼w-׆l p.S[Xw!/ C.s ]KKTzdZZ!_|y ZKv^ʳb*ـ>oTaee(TL(:%q iqX]nlbS[ 8;N(G .4-669u%\"y֞Le ̼$OdI;T_=ϿBY kuEM6BG9CG>K ÌpHKs2"0ǯAG
"RPΞ:spmx/ J J5+OQlO }.:92g;0a Cܪ f.h{R{(l|YV-5K~
2}NI:<T;Y+o v֞\wQ(R,& t*r'dg1C*ʰXbZ~ ZЀ.e[$?-19R :Yî\,ǬkY / Mx{R.Ӯ[z&Z1P-2,:KlFӶ{:
34QяlFl"0.2H9i0*;@;bзfrq= I ڵ.S]d{\F&=WY/OP_NW^,NL"b7PẐ"BX kpˑ-@\DXNZ*y
yN8%FzS;ea9j T9#
_2Cm .%rIV ;L $8Xbnm\Vpf1>pMJ y.2gbe⦑7 b>Y n$̗_+ 3/J2~(ūlY$s1`Ôΰb Y1),+E#lU@2#R@F>+h
5C,I.\wHU
/шZ:iY7)) \6)]H"sz.pb#y+715..l e&,di
.WA
#&'>o
3wlPZ 嵵m֙6 `DN~/[ ,\F
eAu9*j4JQ#ڈx':j)>dހC|k,n<A{5_dPc%A|l)QP {$ \fTu4E媈x&pe Qv]1{DYE om\$y+~q23
jnpFN{E `
q[9 ^EOćЪD_
bj*PqL3 tL9/Xt.ٻVݱ 14HT*X=WSjGAXPΩ(""ha @Hgik \7 s-hE
[P)JJ z :NlIuq,Bc$c]0Z&Y/R Vf.%̛l}q^*Y_>譧PVO,]mEQf0HJd(3gbSʞ aH?3HiMnVk PA ɻw'O>ƺ֒ ly|0v|xڛVΞٳ
ಂhՉ)Ԏ}K1m4nIZ5 [")Gp-=Mn=[3YɈ`^Mxc
:G$7*^ۍ/,5[V] CT`ÚjN4mW\_±6t?&z+FQ*cUe&7']^7bCFZl&"9smмI9=V͐A$ΠlG7Sn$PMhe?SwDJƸM7]x uTjC~^Χ
, "Os{C g%JgY
D
LҴqѼ¡xLXxBзnbl)h6,bU'LN ,v) jDz/PMMP'Q5\5
ӧSzMSm"ɸ (gr,p)lxv[bu6qsQǽO[yL rtb;GOdF_% Z
LeE"P6l*~=OvpXͥLA#_n&{EWIaA$MQ`(֛ޛ% .ݦ/#i 5unٔ{97|`hWMna153p+
yڂUM5nm\آmI=l$Y-bôگ#̧F3ݽr?
SM~;Iu<-:tE汮 %,
F?PK !HD' h fastapi-0.26.0.dist-info/RECORD˖H< TsInY $w 朩*zwȈLU[3a]P XҶg \^bV㚼~+
Quսp*k/5lE{vtB"Wгjc)>PhWn{o3-gq,\9좩f:5N63A
QܾRy &PSS7n? a{UU#pD)hOkDqo~TQj?V'ޑ/RJy-{N##uS+xY ܬ%#XTAq6?
'
*P" g
S%'Q|6@ތ0SgUb YVʹјBy>1<Փcdh%OM C)xeeo7g4)~ Fs/: F|(m*} 5\}P<54'\ 4 1\-:hϳ_4P,K^8
0iCݵK;WN)EH M\EˬqRRo 7Sd9Gky3Ǜ;7NQG @0,3s"{_Ff= ;fY Z&5*7ic.E֢W/M aj5k]Н&S(Y|γa }Ks^ RoGq$>iDN6)*zCՌ 97+(,Dp1,yN3Q=vP&Qѣ $5 `RfwG/<> \~k<~MR _ #9A>=)NV)F7ft B ?EKn e](dϑ<]*ԝ}MȤ x#a4Ӎ5J-`yI)"{Y _:xB>)}AHPg
EiH3-ߣ:
#*Fhzexzs]s/fr
"KI8s*ؑNF|մOU:Kuѱ4&B b˾8Fo[w=WIcJpog_].ٚm7 @iRV!#ռ+$}y k;%a>_-8mQm] =%vNcx˳z7Ύr4G>B8?PK |N fastapi/__init__.pyPK |NyU U fastapi/applications.pyPK |N X fastapi/datastructures.pyPK |N Ŀ 6Z fastapi/encoders.pyPK |NG G i fastapi/exception_handlers.pyPK |N l fastapi/exceptions.pyPK |NhS S n fastapi/param_functions.pyPK |NVPC + fastapi/params.pyPK |N fastapi/py.typedPK |NYL7v v fastapi/routing.pyPK |NaD D ; fastapi/utils.pyPK |N fastapi/dependencies/__init__.pyPK |N u fastapi/dependencies/models.pyPK |N+0H H % fastapi/dependencies/utils.pyPK |N n fastapi/openapi/__init__.pyPK |Nma a &o fastapi/openapi/constants.pyPK |N
j o fastapi/openapi/docs.pyPK |N#5) ) Ƀ fastapi/openapi/models.pyPK |Ny* * խ fastapi/openapi/utils.pyPK |N߸^ ^ fastapi/security/__init__.pyPK |N$+8 8 ~ fastapi/security/api_key.pyPK |Nr(ԍ fastapi/security/base.pyPK |Nx3 fastapi/security/http.pyPK |N% fastapi/security/oauth2.pyPK |N ' U fastapi/security/open_id_connect_url.pyPK |N
fastapi/security/utils.pyPK |N{> > fastapi-0.26.0.dist-info/LICENSEPK !HP O L fastapi-0.26.0.dist-info/WHEELPK !HLa C ! fastapi-0.26.0.dist-info/METADATAPK !HD' h (5 fastapi-0.26.0.dist-info/RECORDPK b :