API

This part of the documentation lists the full API reference of all classes and functions.

WSGI

class example.wsgi.ApplicationLoader(application, overrides=None)

Bases: BaseApplication

Define gunicorn interface for any given web framework.

Parameters:

  • application (Any) – Any given web framework application object instance.

  • overrides (Optional[Dict[str, Any]]) – Map of gunicorn settings to override.

_application

Any given web framework application object instance.

Type:

Any

_overrides

Map of gunicorn settings to override.

Type:

Optional[Dict[str, Any]]

_set_cfg(cfg)

Set gunicorn config given map of setting names to their values.

Parameters:

cfg (Dict[str, Any]) – Map of gunicorn setting names to their values.

Raises:

Exception – Raised on config error.

Return type:

None

init(parser, opts, args)

Patch required but not needed base class method.

Parameters:

  • parser (Any) –

  • opts (Any) –

  • args (Any) –

Return type:

None

load()

Load WSGI application.

Return type:

Any

load_config()

Load gunicorn configuration.

Return type:

None

Config

Application configuration.

The config submodule defines configuration for your application, router, gunicorn, and more.

Resources:

  1. Pydantic documentation

  2. Gunicorn documentation

class example.config.application.Application(_env_file='<object object>', _env_file_encoding=None, _env_nested_delimiter=None, _secrets_dir=None, *, DEBUG=True, PROJECT_NAME='example', VERSION='0.1.0', DOCS_URL='/', USE_REDIS=False)

Bases: BaseSettings

Define application configuration model.

Constructor will attempt to determine the values of any fields not passed as keyword arguments by reading from the environment. Default values will still be used if the matching environment variable is not set.

Environment variables:

  • FASTAPI_DEBUG

  • FASTAPI_PROJECT_NAME

  • FASTAPI_VERSION

  • FASTAPI_DOCS_URL

  • FASTAPI_USE_REDIS

Parameters:
DEBUG

FastAPI logging level. You should disable this for production.

Type:

bool

PROJECT_NAME

FastAPI project name.

Type:

str

VERSION

Application version.

Type:

str

DOCS_URL

Path where swagger ui will be served at.

Type:

str

USE_REDIS

Whether or not to use Redis.

Type:

bool

class Config

Bases: object

Config sub-class needed to customize BaseSettings settings.

case_sensitive

When case_sensitive is True, the environment variable names must match field names (optionally with a prefix)

Type:

bool

env_prefix

The prefix for environment variable.

Type:

str

Resources:

https://pydantic-docs.helpmanual.io/usage/settings/

class example.config.redis.Redis(_env_file='<object object>', _env_file_encoding=None, _env_nested_delimiter=None, _secrets_dir=None, *, REDIS_HOST='127.0.0.1', REDIS_PORT=6379, REDIS_USERNAME=None, REDIS_PASSWORD=None, REDIS_USE_SENTINEL=False)

Bases: BaseSettings

Define Redis configuration model.

Constructor will attempt to determine the values of any fields not passed as keyword arguments by reading from the environment. Default values will still be used if the matching environment variable is not set.

Environment variables:

  • FASTAPI_REDIS_HOTS

  • FASTAPI_REDIS_PORT

  • FASTAPI_REDIS_USERNAME

  • FASTAPI_REDIS_PASSWORD

  • FASTAPI_REDIS_USE_SENTINEL

Parameters:
REDIS_HOTS

Redis host.

Type:

str

REDIS_PORT

Redis port.

Type:

int

REDIS_USERNAME

Redis username.

Type:

Optional[str]

REDIS_PASSWORD

Redis password.

Type:

Optional[str]

REDIS_USE_SENTINEL

If provided Redis config is for Sentinel.

Type:

bool

class Config

Bases: object

Config sub-class needed to customize BaseSettings settings.

case_sensitive

When case_sensitive is True, the environment variable names must match field names (optionally with a prefix)

Type:

bool

env_prefix

The prefix for environment variable.

Type:

str

Resources:

https://pydantic-docs.helpmanual.io/usage/settings/

Gunicorn configuration file.

Resources:

  1. https://docs.gunicorn.org/en/20.1.0/settings.html

CLI

Command-line interface.

The cli submodule defines Click command-line interface root and its commands.

Resources:

  1. Click documentation

example.cli.cli.cli(*args, **kwargs)

Define command-line interface root.

Parameters:

  • options (Dict[str, Any]) – Map of command option names to their parsed values.

  • args (Any) –

  • kwargs (Any) –

Return type:

Any

example.cli.utils.validate_directory(ctx, param, value)

Verify if given path value is writable and parent directory exists.

Parameters:

  • ctx (click.Context) – Click Context object instance.

  • param (click.Option) – Click Option object instance.

  • value (str) – Click Option value.

Returns:

Original value.

Return type:

str

Raises:

click.BadParameter – If given path value is not writable or parent directory does not exist.

example.cli.serve.serve(*args, **kwargs)

Define command-line interface serve command.

Parameters:

  • ctx (click.Context) – Click Context class object instance.

  • options (Dict[str, Any]) – Map of command option names to their parsed values.

  • args (Any) –

  • kwargs (Any) –

Return type:

Any

App

Application implementation.

The app submodule defines application controllers, models, views, utils, exceptions, and all other implementations for the need of your application.

Resources:

  1. FastAPI documentation

  2. Pydantic documentation

async example.app.asgi.on_startup()

Define FastAPI startup event handler.

Resources:

  1. https://fastapi.tiangolo.com/advanced/events/#startup-event

Return type:

None

async example.app.asgi.on_shutdown()

Define FastAPI shutdown event handler.

Resources:

  1. https://fastapi.tiangolo.com/advanced/events/#shutdown-event

Return type:

None

example.app.asgi.get_application()

Initialize FastAPI application.

Returns:

Application object instance.

Return type:

FastAPI

Application configuration - root APIRouter.

Defines all FastAPI application endpoints.

Resources:

  1. https://fastapi.tiangolo.com/tutorial/bigger-applications

Controllers

Application implementation - controllers.

async example.app.controllers.ready.readiness_check()

Run basic application health check.

If the application is up and running then this endpoint will return simple response with status ok. Moreover, if it has Redis enabled then connection to it will be tested. If Redis ping fails, then this endpoint will return 502 HTTP error.

Returns:

ReadyResponse model object instance.

Return type:

response (ReadyResponse)

Raises:

HTTPException – If applications has enabled Redis and can not connect to it. NOTE! This is the custom exception, not to be mistaken with FastAPI.HTTPException class.

Models

Application implementation - models.

Views

Application implementation - views.

class example.app.views.error.ErrorModel(*, code, message, details=None)

Bases: BaseModel

Define base error model for the response.

code

HTTP error status code.

Type:

int

message

Detail on HTTP error.

Type:

str

status

HTTP error reason-phrase as per in RFC7235. NOTE! Set automatically based on HTTP error status code.

Type:

str

Raises:

pydantic.error_wrappers.ValidationError – If any of provided attribute doesn’t pass type validation.

Parameters:
class Config

Bases: object

Config sub-class needed to extend/override the generated JSON schema.

More details can be found in pydantic documentation: https://pydantic-docs.helpmanual.io/usage/schema/#schema-customization

static schema_extra(schema)

Post-process the generated schema.

Method can have one or two positional arguments. The first will be the schema dictionary. The second, if accepted, will be the model class. The callable is expected to mutate the schema dictionary in-place; the return value is not used.

Parameters:

schema (Dict[str, Any]) – The schema dictionary.

Return type:

None

classmethod _set_status(values)

Set the status field value based on the code attribute value.

Parameters:

values (Dict[str, Any]) – Stores the attributes of the ErrorModel object.

Returns:

The attributes of the ErrorModel object

with the status field.

Return type:

Dict[str, Any]

class example.app.views.error.ErrorResponse(*, error)

Bases: BaseModel

Define error response model.

error

ErrorModel class object instance.

Type:

ErrorModel

Raises:

pydantic.error_wrappers.ValidationError – If any of provided attribute doesn’t pass type validation.

Parameters:

error (ErrorModel) –

class Config

Bases: object

Config sub-class needed to extend/override the generated JSON schema.

More details can be found in pydantic documentation: https://pydantic-docs.helpmanual.io/usage/schema/#schema-customization

static schema_extra(schema)

Post-process the generated schema.

Method can have one or two positional arguments. The first will be the schema dictionary. The second, if accepted, will be the model class. The callable is expected to mutate the schema dictionary in-place; the return value is not used.

Parameters:

schema (Dict[str, Any]) – The schema dictionary.

Return type:

None

Exceptions

Application implementation - exceptions.

class example.app.exceptions.http.HTTPException(status_code, content=None, headers=None)

Bases: Exception

Define custom HTTPException class definition.

This exception combined with exception_handler method allows you to use it the same manner as you’d use FastAPI.HTTPException with one difference. You have freedom to define returned response body, whereas in FastAPI.HTTPException content is returned under “detail” JSON key.

FastAPI.HTTPException source: https://github.com/tiangolo/fastapi/blob/master/fastapi/exceptions.py

Parameters:
async example.app.exceptions.http.http_exception_handler(request, exception)

Define custom HTTPException handler.

In this application custom handler is added in asgi.py while initializing FastAPI application. This is needed in order to handle custom HTTException globally.

More details: https://fastapi.tiangolo.com/tutorial/handling-errors/#install-custom-exception-handlers

Parameters:
Returns:

FastAPI.response.JSONResponse class object instance initialized with

kwargs from custom HTTPException.

Return type:

JSONResponse

Utils

Application implementation - utilities.

Resources:

  1. https://aioredis.readthedocs.io/en/latest/

class example.app.utils.aiohttp_client.AiohttpClient

Bases: object

Aiohttp session client utility.

Utility class for handling HTTP async request for whole FastAPI application scope.

sem

Semaphore value.

Type:

asyncio.Semaphore, optional

aiohttp_client

Aiohttp client session object instance.

Type:

aiohttp.ClientSession, optional

async classmethod close_aiohttp_client()

Close aiohttp client session.

Return type:

None

async classmethod delete(url, headers=None, raise_for_status=False)

Execute HTTP DELETE request.

Parameters:

  • url (str) – HTTP DELETE request endpoint.

  • headers (Dict[str, AnyStr]) – Optional HTTP Headers to send with the request.

  • raise_for_status (bool) – Automatically call ClientResponse.raise_for_status() for response if set to True.

Returns:

HTTP DELETE request response - aiohttp.ClientResponse

object instance.

Return type:

response

async classmethod get(url, headers=None, raise_for_status=False)

Execute HTTP GET request.

Parameters:

  • url (str) – HTTP GET request endpoint.

  • headers (Dict[str, AnyStr]) – Optional HTTP Headers to send with the request.

  • raise_for_status (bool) – Automatically call ClientResponse.raise_for_status() for response if set to True.

Returns:

HTTP GET request response - aiohttp.ClientResponse

object instance.

Return type:

response

classmethod get_aiohttp_client()

Create aiohttp client session object instance.

Returns:

ClientSession object instance.

Return type:

aiohttp.ClientSession

async classmethod patch(url, data=None, headers=None, raise_for_status=False)

Execute HTTP PATCH request.

Parameters:

  • url (str) – HTTP PATCH request endpoint.

  • data (Optional[Any]) – The data to send in the body of the request. This can be a FormData object or anything that can be passed into FormData, e.g. a dictionary, bytes, or file-like object.

  • headers (Dict[str, AnyStr]) – Optional HTTP Headers to send with the request.

  • raise_for_status (bool) – Automatically call ClientResponse.raise_for_status() for response if set to True.

Returns:

HTTP PATCH request response - aiohttp.ClientResponse

object instance.

Return type:

response

async classmethod post(url, data=None, headers=None, raise_for_status=False)

Execute HTTP POST request.

Parameters:

  • url (str) – HTTP POST request endpoint.

  • data (Optional[Any]) – The data to send in the body of the request. This can be a FormData object or anything that can be passed into FormData, e.g. a dictionary, bytes, or file-like object.

  • headers (Dict[str, AnyStr]) – Optional HTTP Headers to send with the request.

  • raise_for_status (bool) – Automatically call ClientResponse.raise_for_status() for response if set to True.

Returns:

HTTP POST request response - aiohttp.ClientResponse

object instance.

Return type:

response

async classmethod put(url, data=None, headers=None, raise_for_status=False)

Execute HTTP PUT request.

Parameters:

  • url (str) – HTTP PUT request endpoint.

  • data (Optional[Any]) – The data to send in the body of the request. This can be a FormData object or anything that can be passed into FormData, e.g. a dictionary, bytes, or file-like object.

  • headers (Dict[str, AnyStr]) – Optional HTTP Headers to send with the request.

  • raise_for_status (bool) – Automatically call ClientResponse.raise_for_status() for response if set to True.

Returns:

HTTP PUT request response - aiohttp.ClientResponse

object instance.

Return type:

response

class example.app.utils.redis.RedisClient

Bases: object

Define Redis utility.

Utility class for handling Redis database connection and operations.

redis_client

Redis client object instance.

Type:

aioredis.Redis, optional

log

Logging handler for this class.

Type:

logging.Logger

base_redis_init_kwargs

Common kwargs regardless other Redis configuration

Type:

Dict[str, Union[str, int]]

connection_kwargs

Extra kwargs for Redis object init.

Type:

Optional[Dict[str, str]]

async classmethod close_redis_client()

Close Redis client.

Return type:

None

async classmethod exists(key)

Execute Redis EXISTS command.

Returns if key exists.

Parameters:

key (str) – Redis db key.

Returns:

Boolean whether key exists in Redis db.

Return type:

bool

Raises:

aioredis.RedisError – If Redis client failed while executing command.

async classmethod get(key)

Execute Redis GET command.

Get the value of key. If the key does not exist the special value None is returned. An error is returned if the value stored at key is not a string, because GET only handles string values.

Parameters:

key (str) – Redis db key.

Returns:

Value of key.

Return type:

str

Raises:

aioredis.RedisError – If Redis client failed while executing command.

async classmethod lrange(key, start, end)

Execute Redis LRANGE command.

Returns the specified elements of the list stored at key. The offsets start and stop are zero-based indexes, with 0 being the first element of the list (the head of the list), 1 being the next element and so on. These offsets can also be negative numbers indicating offsets starting at the end of the list. For example, -1 is the last element of the list, -2 the penultimate, and so on.

Parameters:

  • key (str) – Redis db key.

  • start (int) – Start offset value.

  • end (int) – End offset value.

Returns:

Returns the specified elements of the list stored at key.

Return type:

str

Raises:

aioredis.RedisError – If Redis client failed while executing command.

classmethod open_redis_client()

Create Redis client session object instance.

Based on configuration create either Redis client or Redis Sentinel.

Returns:

Redis object instance.

Return type:

aioredis.Redis

async classmethod ping()

Execute Redis PING command.

Ping the Redis server.

Returns:

Boolean, whether Redis client could ping Redis server.

Return type:

bool

Raises:

aioredis.RedisError – If Redis client failed while executing command.

async classmethod rpush(key, value)

Execute Redis RPUSH command.

Insert all the specified values at the tail of the list stored at key. If key does not exist, it is created as empty list before performing the push operation. When key holds a value that is not a list, an error is returned.

Parameters:

  • key (str) – Redis db key.

  • value (Union[str, List[AnyStr]]) – Single or multiple values to append.

Returns:

Length of the list after the push operation.

Return type:

int

Raises:

aioredis.RedisError – If Redis client failed while executing command.

async classmethod set(key, value)

Execute Redis SET command.

Set key to hold the string value. If key already holds a value, it is overwritten, regardless of its type.

Parameters:

  • key (str) – Redis db key.

  • value (str) – Value to be set.

Returns:

Redis SET command response, for more info

look: https://redis.io/commands/set#return-value

Return type:

response

Raises:

aioredis.RedisError – If Redis client failed while executing command.