Sanic

This documents the various Sanic specific functionality but doesn’t cover internals of the extension.

Extension

class dockerflow.sanic.app.Dockerflow(app=None, redis=None, silenced_checks=None, version_path='.', *args, **kwargs)[source]

The Dockerflow Sanic extension. Set it up like this:

myproject.py
from sanic import Sanic
from dockerflow.sanic import Dockerflow

app = Sanic(__name__)
dockerflow = Dockerflow(app)

Or if you use the Sanic application factory pattern, in an own module set up Dockerflow first:

myproject/deployment.py
from dockerflow.sanic import Dockerflow

dockerflow = Dockerflow()

and then import and initialize it with the Sanic application object when you create the application:

myproject/app.py
def create_app(config_filename):
    app = Sanic(__name__)
    app.config.from_pyfile(config_filename)

    from myproject.deployment import dockerflow
    dockerflow.init_app(app)

    from myproject.views.admin import admin
    from myproject.views.frontend import frontend
    app.register_blueprint(admin)
    app.register_blueprint(frontend)

    return app

See the parameters for a more detailed list of optional features when initializing the extension.

Parameters:
  • app (Sanic or None) – The Sanic app that this Dockerflow extension should be initialized with.
  • redis – A SanicRedis instance to be used by the built-in Dockerflow check for the sanic_redis connection.
  • silenced_checks (list) – Dockerflow check IDs to ignore when running through the list of configured checks.
  • version_path – The filesystem path where the version.json can be found. Defaults to ..
check(func=None, name=None)[source]

A decorator to register a new Dockerflow check to be run when the /__heartbeat__ endpoint is called., e.g.:

from dockerflow.sanic import checks

@dockerflow.check
async def storage_reachable():
    try:
        acme.storage.ping()
    except SlowConnectionException as exc:
        return [checks.Warning(exc.msg, id='acme.health.0002')]
    except StorageException as exc:
        return [checks.Error(exc.msg, id='acme.health.0001')]

also works without async:

@dockerflow.check
def storage_reachable():
    # ...

or using a custom name:

@dockerflow.check(name='acme-storage-check')
async def storage_reachable():
    # ...
init_app(app)[source]

Add the Dockerflow views and middleware to app.

init_check(check, obj)[source]

Adds a given check callback with the provided object to the list of checks. Useful for built-ins but also advanced custom checks.

summary_extra(request)[source]

Build the extra data for the summary logger.

version_callback(func)[source]

A decorator to optionally register a new Dockerflow version callback and use that instead of the default of dockerflow.version.get_version().

The callback will be passed the value of the version_path parameter to the Dockerflow extension object, which defaults to the parent directory of the Sanic app’s root path.

The callback should return a dictionary with the version information as defined in the Dockerflow spec, or None if no version information could be loaded.

E.g.:

import aiofiles

app = Sanic(__name__)
dockerflow = Dockerflow(app)

@dockerflow.version_callback
async def my_version(root):
    path = os.path.join(root, 'acme_version.json')
    async with aiofiles.open(path, mode='r') as f:
        raw = await f.read()
    return json.loads(raw)

Checks

dockerflow.sanic.checks.check_redis_connected(redis)[source]

A built-in check to connect to Redis using the given client and see if it responds to the PING command.

It’s automatically added to the list of Dockerflow checks if a SanicRedis instance is passed to the Dockerflow class during instantiation, e.g.:

import redis as redislib
from sanic import Sanic
from dockerflow.sanic import Dockerflow

app = Sanic(__name__)
redis = redislib.from_url("redis://:password@localhost:6379/0")
dockerflow = Dockerflow(app, redis=redis)

An alternative approach to instantiating a Redis client directly would be using the Sanic-Redis Sanic extension:

from sanic import Sanic
from sanic_redis import SanicRedis
from dockerflow.sanic import Dockerflow

app = Sanic(__name__)
app.config['REDIS'] = {'address': 'redis://:password@localhost:6379/0'}
redis = SanicRedis(app)
dockerflow = Dockerflow(app, redis=redis)