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.

class Config[source]

Bases: object

env_prefix = 'CONTRIB_'
secrets_dir = None
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]

Bases: fastapi_contrib.exceptions.HTTPException

exception fastapi_contrib.exceptions.ForbiddenError(error_code: int = 403, detail: Any = 'Forbidden.', fields: Optional[List[Dict]] = None)[source]

Bases: fastapi_contrib.exceptions.HTTPException

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]

Bases: fastapi_contrib.exceptions.HTTPException

exception fastapi_contrib.exceptions.NotFoundError(error_code: int = 404, detail: Any = 'Not found.', fields: Optional[List[Dict]] = None)[source]

Bases: fastapi_contrib.exceptions.HTTPException

exception fastapi_contrib.exceptions.UnauthorizedError(error_code: int = 401, detail: Any = 'Unauthorized.', fields: Optional[List[Dict]] = None)[source]

Bases: fastapi_contrib.exceptions.HTTPException

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

class fastapi_contrib.pagination.PaginationMeta(name, bases, namespace, *args, **kwargs)[source]

Bases: type

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.'
abstract has_required_permissions(request: starlette.requests.Request) → bool[source]
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

get_route_handler() → Callable[source]

Module contents

Top-level package for FastAPI Contrib.