fastapi_contrib package¶
Subpackages¶
Submodules¶
fastapi_contrib.conf module¶
- class fastapi_contrib.conf.Settings(_env_file: Optional[Union[pathlib.Path, str]] = '<object object>', _env_file_encoding: Optional[str] = None, _secrets_dir: Optional[Union[pathlib.Path, str]] = None, *, logger: str = 'logging', log_level: str = 'INFO', debug_timing: bool = False, request_id_header: str = 'Request-ID', service_name: str = 'fastapi_contrib', trace_id_header: str = 'X-TRACE-ID', jaeger_host: str = 'jaeger', jaeger_port: int = 5775, jaeger_sampler_type: str = 'probabilistic', jaeger_sampler_rate: float = 1.0, mongodb_dsn: str = 'mongodb://example:pwd@localhost:27017', mongodb_dbname: str = 'default', mongodb_min_pool_size: int = 0, mongodb_max_pool_size: int = 100, mongodb_id_generator: str = 'fastapi_contrib.db.utils.default_id_generator', now_function: str = None, TZ: str = 'UTC', fastapi_app: str = None, user_model: str = 'fastapi_contrib.auth.models.User', token_model: str = 'fastapi_contrib.auth.models.Token', token_generator: str = 'fastapi_contrib.auth.utils.default_token_generator', apps: List[str] = [], apps_folder_name: str = 'apps')[source]¶
Bases:
pydantic.env_settings.BaseSettings
Configuration settings for this library.
For now you could only change the settings via CONTRIB_<ATTRIBUTE_NAME> environment variables.
- Parameters
logger – Dotted path to the logger (using this attribute, standard logging methods will be used: logging.debug(), .info(), etc.
log_level – Standard LEVEL for logging (DEBUG/INFO/WARNING/etc.)
debug_timing – Whether to enable time logging for decorated functions
request_id_header – String name for header, that is expected to have unique request id for tracing purposes. Might go away when we add opentracing here.
mongodb_dsn – DSN connection string to MongoDB
mongodb_dbname – String name of a database to connect to in MongoDB
mongodb_id_generator – Dotted path to the function, which will be used when assigning IDs for MongoDB records
now_function – Dotted path to the function, which will be used when assigning created field for MongoDB records. Should be used throughout the code for consistency.
fastapi_app – Dotted path to the instance of FastAPI main app.
user_model – Dotted path to the class, which will be used as the main user model in a project.
token_model – Dotted path to the class, which will be used as the main token model in a project.
token_generator – Dotted path to the function, which will be used when assigning key attribute of a token model.
apps – List of app names. For now only needed to detect models inside them and generate indexes upon startup (see: create_indexes)
apps_folder_name – Name of the folders which contains dirs with apps.
- TZ: str¶
- apps: List[str]¶
- apps_folder_name: str¶
- debug_timing: bool¶
- fastapi_app: str¶
- jaeger_host: str¶
- jaeger_port: int¶
- jaeger_sampler_rate: float¶
- jaeger_sampler_type: str¶
- log_level: str¶
- logger: str¶
- mongodb_dbname: str¶
- mongodb_dsn: str¶
- mongodb_id_generator: str¶
- mongodb_max_pool_size: int¶
- mongodb_min_pool_size: int¶
- now_function: str¶
- request_id_header: str¶
- service_name: str¶
- token_generator: str¶
- token_model: str¶
- trace_id_header: str¶
- user_model: str¶
fastapi_contrib.exception_handlers module¶
- async fastapi_contrib.exception_handlers.http_exception_handler(request: starlette.requests.Request, exc: starlette.exceptions.HTTPException) → fastapi_contrib.common.responses.UJSONResponse[source]¶
- Handles StarletteHTTPException, translating it into flat dict error data:
code - unique code of the error in the system
detail - general description of the error
fields - list of dicts with description of the error in each field
- Parameters
request – Starlette Request instance
exc – StarletteHTTPException instance
- Returns
UJSONResponse with newly formatted error data
- async fastapi_contrib.exception_handlers.internal_server_error_handler(request: starlette.requests.Request, exc: fastapi.exceptions.RequestValidationError) → fastapi_contrib.common.responses.UJSONResponse[source]¶
- async fastapi_contrib.exception_handlers.not_found_error_handler(request: starlette.requests.Request, exc: fastapi.exceptions.RequestValidationError) → fastapi_contrib.common.responses.UJSONResponse[source]¶
- fastapi_contrib.exception_handlers.parse_error(err: Any, field_names: List, raw: bool = True) → Optional[dict][source]¶
Parse single error object (such as pydantic-based or fastapi-based) to dict
- Parameters
err – Error object
field_names – List of names of the field that are already processed
raw – Whether this is a raw error or wrapped pydantic error
- Returns
dict with name of the field (or “__all__”) and actual message
- fastapi_contrib.exception_handlers.raw_errors_to_fields(raw_errors: List) → List[dict][source]¶
Translates list of raw errors (instances) into list of dicts with name/msg
- Parameters
raw_errors – List with instances of raw error
- Returns
List of dicts (1 dict for every raw error)
- fastapi_contrib.exception_handlers.setup_exception_handlers(app: fastapi.applications.FastAPI) → None[source]¶
Helper function to setup exception handlers for app. Use during app startup as follows:
app = FastAPI() @app.on_event('startup') async def startup(): setup_exception_handlers(app)
- Parameters
app – app object, instance of FastAPI
- Returns
None
- async fastapi_contrib.exception_handlers.validation_exception_handler(request: starlette.requests.Request, exc: fastapi.exceptions.RequestValidationError) → fastapi_contrib.common.responses.UJSONResponse[source]¶
- Handles ValidationError, translating it into flat dict error data:
code - unique code of the error in the system
detail - general description of the error
fields - list of dicts with description of the error in each field
- Parameters
request – Starlette Request instance
exc – StarletteHTTPException instance
- Returns
UJSONResponse with newly formatted error data
fastapi_contrib.exceptions module¶
- exception fastapi_contrib.exceptions.BadRequestError(error_code: int, detail: Any, fields: Optional[List[Dict]] = None)[source]¶
- exception fastapi_contrib.exceptions.ForbiddenError(error_code: int = 403, detail: Any = 'Forbidden.', fields: Optional[List[Dict]] = None)[source]¶
- exception fastapi_contrib.exceptions.HTTPException(status_code: int, error_code: int, detail: Optional[Any] = None, fields: Optional[List[Dict]] = None)[source]¶
Bases:
starlette.exceptions.HTTPException
- exception fastapi_contrib.exceptions.InternalServerError(error_code: int = 500, detail: Any = 'Internal Server Error.', fields: Optional[List[Dict]] = None)[source]¶
- exception fastapi_contrib.exceptions.NotFoundError(error_code: int = 404, detail: Any = 'Not found.', fields: Optional[List[Dict]] = None)[source]¶
fastapi_contrib.pagination module¶
- class fastapi_contrib.pagination.Pagination(request: starlette.requests.Request, offset: int = Query(0), limit: int = Query(100))[source]¶
Bases:
object
Query params parser and db collection paginator in one.
Use it as dependency in route, then invoke paginate with serializer:
app = FastAPI() class SomeSerializer(ModelSerializer): class Meta: model = SomeModel @app.get("/") async def list(pagination: Pagination = Depends()): filter_kwargs = {} return await pagination.paginate( serializer_class=SomeSerializer, **filter_kwargs )
Subclass this pagination to define custom default & maximum values for offset & limit:
class CustomPagination(Pagination): default_offset = 90 default_limit = 1 max_offset = 100` max_limit = 2000
- Parameters
request – starlette Request object
offset – query param of how many records to skip
limit – query param of how many records to show
- default_limit = 100¶
- default_offset = 0¶
- async get_count(**kwargs) → int[source]¶
Retrieves counts for query list, filtered by kwargs.
- Parameters
kwargs – filters that are proxied in db query
- Returns
number of found records
- async get_list(_sort=None, **kwargs) → list[source]¶
Retrieves actual list of records. It comes raw, which means it retrieves dict from DB, instead of making conversion for every object in list into Model.
- Parameters
kwargs – filters that are proxied in db query
- Returns
list of dicts from DB, filtered by kwargs
- get_next_url() → str[source]¶
Constructs next parameter in resulting JSON, produces URL for next “page” of paginated results.
- Returns
URL for next “page” of paginated results.
- get_previous_url() → str[source]¶
Constructs previous parameter in resulting JSON, produces URL for previous “page” of paginated results.
- Returns
URL for previous “page” of paginated results.
- max_limit = 1000¶
- max_offset = None¶
- async paginate(serializer_class: fastapi_contrib.serializers.common.Serializer, _sort=None, **kwargs) → dict[source]¶
Actual pagination function, takes serializer class, filter options as kwargs and returns dict with the following fields:
count - counts for query list, filtered by kwargs
next - URL for next “page” of paginated results
previous - URL for previous “page” of paginated results
result - actual list of records (dicts)
- Parameters
serializer_class – needed to get Model & sanitize list from DB
kwargs – filters that are proxied in db query
- Returns
dict that should be returned as a response
fastapi_contrib.permissions module¶
- class fastapi_contrib.permissions.BasePermission(request: starlette.requests.Request)[source]¶
Bases:
abc.ABC
Abstract permission that all other Permissions must be inherited from.
Defines basic error message, status & error codes.
Upon initialization, calls abstract method has_required_permissions which will be specific to concrete implementation of Permission class.
You would write your permissions like this:
class TeapotUserAgentPermission(BasePermission): def has_required_permissions(self, request: Request) -> bool: return request.headers.get('User-Agent') == "Teapot v1.0"
- error_code = 403¶
- error_msg = 'Forbidden.'¶
- status_code = 403¶
- class fastapi_contrib.permissions.PermissionsDependency(permissions_classes: list)[source]¶
Bases:
object
Permission dependency that is used to define and check all the permission classes from one place inside route definition.
Use it as an argument to FastAPI’s Depends as follows:
app = FastAPI() @app.get( "/teapot/", dependencies=[Depends( PermissionsDependency([TeapotUserAgentPermission]))] ) async def teapot() -> dict: return {"teapot": True}
fastapi_contrib.routes module¶
- class fastapi_contrib.routes.ValidationErrorLoggingRoute(path: str, endpoint: Callable[[...], Any], *, response_model: Optional[Type[Any]] = None, status_code: int = 200, tags: Optional[List[str]] = None, dependencies: Optional[Sequence[fastapi.params.Depends]] = None, summary: Optional[str] = None, description: Optional[str] = None, response_description: str = 'Successful Response', responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None, deprecated: Optional[bool] = None, name: Optional[str] = None, methods: Optional[Union[Set[str], List[str]]] = None, operation_id: Optional[str] = None, response_model_include: Optional[Union[Set[Union[int, str]], Dict[Union[int, str], Any]]] = None, response_model_exclude: Optional[Union[Set[Union[int, str]], Dict[Union[int, str], Any]]] = None, response_model_by_alias: bool = True, response_model_exclude_unset: bool = False, response_model_exclude_defaults: bool = False, response_model_exclude_none: bool = False, include_in_schema: bool = True, response_class: Union[Type[starlette.responses.Response], fastapi.datastructures.DefaultPlaceholder] = <fastapi.datastructures.DefaultPlaceholder object>, dependency_overrides_provider: Optional[Any] = None, callbacks: Optional[List[starlette.routing.BaseRoute]] = None)[source]¶
Bases:
fastapi.routing.APIRoute
Module contents¶
Top-level package for FastAPI Contrib.