Skip to main content

Plugins for FastAPI framework

Project description

Plugins for FastAPI framework, high performance, easy to learn, fast to code, ready for production

Build Status Coverage Package version Join the chat at


Aioredis is about to merge into redis-py. So waiting for release and first benchmarks. Version 1.3.1 remains the fastest.


FastAPI framework plugins - simple way to share fastapi code and utilities across applications.

The concept is plugin - plug a functional utility into your application without or with minimal effort.


See release notes


pip install fastapi-plugins
pip install fastapi-plugins[memcached]
pip install fastapi-plugins[all]

Quick start


Add information about plugin system.

Application settings

Add information about settings.

Application configuration

Add information about configuration of an application

Complete example

import fastapi
import fastapi_plugins

from fastapi_plugins.memcached import MemcachedSettings
from fastapi_plugins.memcached import memcached_plugin, TMemcachedPlugin

import asyncio
import aiojobs
import aioredis
import contextlib
import logging

class AppSettings(
    api_name: str = str(__name__)
    logging_level: int = logging.DEBUG
    logging_style: fastapi_plugins.LoggingStyle = fastapi_plugins.LoggingStyle.logjson

class AppSettingsSentinel(AppSettings):
    redis_type = fastapi_plugins.RedisType.sentinel
    redis_sentinels = 'localhost:26379'

async def lifespan(app: fastapi.FastAPI):
    config = fastapi_plugins.get_config()
    await fastapi_plugins.config_plugin.init_app(app, config)
    await fastapi_plugins.config_plugin.init()
    await fastapi_plugins.log_plugin.init_app(app, config, name=__name__)
    await fastapi_plugins.log_plugin.init()
    await memcached_plugin.init_app(app, config)
    await memcached_plugin.init()
    await fastapi_plugins.redis_plugin.init_app(app, config=config)
    await fastapi_plugins.redis_plugin.init()
    await fastapi_plugins.scheduler_plugin.init_app(app=app, config=config)
    await fastapi_plugins.scheduler_plugin.init()
    await fastapi_plugins.control_plugin.init_app(
    await fastapi_plugins.control_plugin.init()
    await fastapi_plugins.control_plugin.terminate()
    await fastapi_plugins.scheduler_plugin.terminate()
    await fastapi_plugins.redis_plugin.terminate()
    await memcached_plugin.terminate()
    await fastapi_plugins.log_plugin.terminate()
    await fastapi_plugins.config_plugin.terminate()

app = fastapi_plugins.register_middleware(fastapi.FastAPI(lifespan=lifespan))

async def root_get(
        cache: fastapi_plugins.TRedisPlugin,
        conf: fastapi_plugins.TConfigPlugin,
        logger: fastapi_plugins.TLoggerPlugin
) -> typing.Dict:
    ping = await
    logger.debug('root_get', extra=dict(ping=ping, api_name=conf.api_name))
    return dict(ping=ping, api_name=conf.api_name)"/jobs/schedule/<timeout>")
async def job_post(
    timeout: int=fastapi.Query(..., title='the job sleep time'),
    cache: fastapi_plugins.TRedisPlugin,
    scheduler: fastapi_plugins.TSchedulerPlugin,
    logger: fastapi_plugins.TLoggerPlugin
) -> str:
    async def coro(job_id, timeout, cache):
        await cache.set(job_id, 'processing')
            await asyncio.sleep(timeout)
            if timeout == 8:
                logger.critical('Ugly erred job %s' % job_id)
                raise Exception('ugly error')
        except asyncio.CancelledError:
            await cache.set(job_id, 'canceled')
            logger.warning('Cancel job %s' % job_id)
        except Exception:
            await cache.set(job_id, 'erred')
            logger.error('Erred job %s' % job_id)
            await cache.set(job_id, 'success')
  'Done job %s' % job_id)

    job_id = str(uuid.uuid4()).replace('-', '')
    logger = await fastapi_plugins.log_adapter(logger, extra=dict(job_id=job_id, timeout=timeout))    # noqa E501'New job %s' % job_id)
    await cache.set(job_id, 'pending')
    logger.debug('Pending job %s' % job_id)
    await scheduler.spawn(coro(job_id, timeout, cache))
    return job_id

async def job_get(
    job_id: str=fastapi.Query(..., title='the job id'),
    cache: fastapi_plugins.TRedisPlugin,
) -> typing.Dict:
    status = await cache.get(job_id)
    if status is None:
        raise fastapi.HTTPException(
            detail='Job %s not found' % job_id
    return dict(job_id=job_id, status=status)"/memcached/demo/<key>")
async def memcached_demo_post(
    key: str=fastapi.Query(..., title='the job id'),
    cache: fastapi_plugins.TMemcachedPlugin,
) -> typing.Dict:
    await cache.set(key.encode(), str(key + '_value').encode())
    value = await cache.get(key.encode())
    return dict(ping=(await, key=key, value=value)


Issues and suggestions are welcome through issues


This project is licensed under the terms of the MIT license.


0.12.0 (2023-03-24)

  • [feature] Annotated support

0.11.0 (2022-09-19)

  • [feature] redis-py replaces aioredis

0.10.0 (2022-07-07)

  • [feature] Update aioredis to 2.x.x
  • [feature] Add fakeredis optionally for development purpose

0.9.1 (2022-06-16)

  • [fix] Fix empty router prefix for control plugin

0.9.0 (2021-09-27)

  • [feature] Logging plugin
  • [feature] Middleware interface - register middleware at application

0.8.2 (2021-09-23)

  • [fix] Fix dependency for aioredis

0.8.1 (2021-03-31)

  • [fix] Fix settings for Python 3.7

0.8.0 (2021-03-31)

  • [feature] Settings plugin

0.7.0 (2021-03-29)

  • [feature] Control plugin with Health, Heartbeat, Environment and Version

0.6.1 (2021-03-24)

  • [fix] Bump aiojobsto get rid of not required dependencies

0.6.0 (2020-11-26)

  • [feature] Memcached

0.5.0 (2020-11-25)

  • [bug] remove __all__ since no API as such (#6).
  • [typo] Fix typos in README (#7).
  • [feature] Add Redis TTL (#8).

0.4.2 (2020-11-24)

  • [bug] Fix Redis URL (#4).

0.4.1 (2020-06-16)

  • Refactor requirements

0.4.0 (2020-04-09)

  • structure and split dependencies to extra

0.3.0 (2020-04-07)

  • Scheduler: tasks scheduler based on aiojobs

0.2.1 (2020-04-06)

  • Redis: pre-start

0.2.0 (2019-12-11)

  • Redis: sentinels

0.1.0 (2019-11-20)

  • Initial release: simple redis pool client

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

fastapi-plugins-0.12.0.tar.gz (26.0 kB view hashes)

Uploaded source

Built Distribution

fastapi_plugins-0.12.0-py3-none-any.whl (22.4 kB view hashes)

Uploaded py3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page