Flask

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

Extension

class dockerflow.flask.app.Dockerflow(app=None, db=None, redis=None, migrate=None, silenced_checks=None, version_path=None, *args, **kwargs)

The Dockerflow Flask extension. Set it up like this:

myproject.py

from flask import Flask
from dockerflow.flask import Dockerflow

app = Flask(__name__)
dockerflow = Dockerflow(app)

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

myproject/deployment.py

from dockerflow.flask import Dockerflow

dockerflow = Dockerflow()

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

myproject/app.py

def create_app(config_filename):
    app = Flask(__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 (Flask or None) – The Flask app that this Dockerflow extension should be initialized with.

• db – A Flask-SQLAlchemy extension instance to be used by the built-in Dockerflow check for the database connection.

• redis – A Redis connection to be used by the built-in Dockerflow check for the Redis connection.

• migrate – A Flask-Migrate extension instance to be used by the built-in Dockerflow check for Alembic migrations.

• 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 the parent directory of the Flask app’s root path.

check(func=None, name=None)

Backwards compatibility method.

property checks

Backwards compatibility alias.

init_app(app)

Initializes the extension with the given app, registers the built-in views with an own blueprint and hooks up our signal callbacks.

init_check(check, obj)

Backwards compatibility method.

summary_extra()

Build the extra data for the summary logger.

user_id()

Return the ID of the current request’s user

version_callback(func)

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 Flask 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.:

app = Flask(__name__)
dockerflow = Dockerflow(app)

@dockerflow.version_callback
def my_version(root):
    return json.loads(os.path.join(root, 'acme_version.json'))

exception dockerflow.flask.app.HeartbeatFailure(description: str | None = None, response: Response | None = None, original_exception: BaseException | None = None)

Checks

dockerflow.flask.checks.check_database_connected(db)

A built-in check to see if connecting to the configured default database backend succeeds.

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

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from dockerflow.flask import Dockerflow

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:////tmp/test.db'
db = SQLAlchemy(app)

dockerflow = Dockerflow(app, db=db)

dockerflow.flask.checks.check_migrations_applied(migrate)

A built-in check to see if all migrations have been applied correctly.

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

from flask import Flask
from flask_migrate import Migrate
from flask_sqlalchemy import SQLAlchemy
from dockerflow.flask import Dockerflow

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:////tmp/test.db'
db = SQLAlchemy(app)
migrate = Migrate(app, db)

dockerflow = Dockerflow(app, db=db, migrate=migrate)

dockerflow.flask.checks.check_redis_connected(client)

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 StrictRedis instances is passed to the Dockerflow class during instantiation, e.g.:

import redis
from flask import Flask
from dockerflow.flask import Dockerflow

app = Flask(__name__)
redis_client = redis.StrictRedis(host='localhost', port=6379, db=0)

dockerflow = Dockerflow(app, redis=redis)

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

from flask import Flask
from flask_redis import FlaskRedis
from dockerflow.flask import Dockerflow

app = Flask(__name__)
app.config['REDIS_URL'] = 'redis://:password@localhost:6379/0'
redis_store = FlaskRedis(app)

dockerflow = Dockerflow(app, redis=redis_store)

Signals

During the rendering of the /__heartbeat__ Flask view two signals are being sent to hook into the result of the checks:

dockerflow.flask.signals.heartbeat_passed

The signal that is sent when the heartbeat checks pass successfully.

dockerflow.flask.signals.heartbeat_failed

The signal that is sent when the heartbeat checks raise either a warning or worse (error, critical)

Both signals receive an additional level parameter that indicates the maximum check level that failed during the rendering.

E.g. to hook into those signals to send data to statsd, do this:

from dockerflow.flask.signals import heartbeat_passed, heartbeat_failed
from myproject.stats import statsd

@heartbeat_passed.connect_via(app)
def heartbeat_passed_handler(sender, level, **extra):
    statsd.incr('heartbeat.pass')

@heartbeat_failed.connect_via(app)
def heartbeat_failed_handler(sender, level, **extra):
    statsd.incr('heartbeat.fail')