Guide

About ConsenSys-Utils

ConsenSys-Utils is a library including a set of utility resources used on a daily basis by ConsenSys France Engineering team.

Create a Flask Application with Factory pattern

Quickstart

ConsenSys-Utils provides multiple features to create a Flask application. In particular ConsenSys-Utils helps you implement the Application factory pattern

1. Create a app.py

   >>> from consensys_utils.flask import FlaskFactory
   >>> from consensys_utils.flask.cli import FlaskGroup
   
   # Create an application factory
   >>> app_factory = FlaskFactory(__name__)
   
   # Declares a click application using ConsenSys-Utils click group
   >>> cli = FlaskGroup(app_factory=app_factory)
   

2. Define an entry point in setup.py:

   from setuptools import setup
   
   setup(
       name='my-app',
       ...,
       entry_points={
           'console_scripts': [
               'my-app=app:cli'
           ],
       },
   )
   

3. Install the application and start the application

   $ pip install -e .
   $ my-app run --config config.yml
   

   Note that

   • config.yml is your .yml configuration file
   • you don’t need to set FLASK_APP environment variable
   • run command reads FLASK_ENV environment variable. If FLASK_ENV=production the application will be run using a gunicorn server otherwise it uses werkzeug default development server

Advanced usage

Class consensys_utils.flask.FlaskFactory allows you to

• provide a specific yaml configuration loader
• provide specifics WSGI middlewares
• initialize specifics Flask extensions
• set application hooks
• register specifics Flask blueprints

Change configuration loader

By default consensys_utils.flask.FlaskFactory uses a .yml configuration that validates against consensys_utils.config.schema.flask.ConfigSchema. If you like you can define your own configuration loader.

>>> from consensys_utils.flask import FlaskFactory
>>> from consensys_utils.flask.cli import FlaskGroup
>>> from cfg_loader import ConfigSchema, YamlConfigLoader
>>> from marshmallow import fields

# Declare you configuration schema and config loader
>>> class MySchema(ConfigSchema):
...     my_parameter = fields.Str()

>>> yaml_config_loader = YamlConfigLoader(config_schema=MySchema)

# Create an application factory
>>> app_factory = FlaskFactory(__name__, yaml_config_loader=yaml_config_loader)

# Declares a click application using ConsenSys-Utils click group
>>> cli = FlaskGroup(app_factory=app_factory)

Add WSGI Middlewares

You can define your own WSGI middlewares and have it automatically applied on your application

>>> from consensys_utils.flask import FlaskFactory
>>> from consensys_utils.flask.cli import FlaskGroup
>>> import base64

>>> class AuthMiddleware:
...     def __init__(self, wsgi):
...         self.wsgi = wsgi
...
...     @staticmethod
...     def is_authenticated(header):
...         if not header:
...             return False
...         _, encoded = header.split(None, 1)
...         decoded = base64.b64decode(encoded).decode('UTF-8')
...         username, password = decoded.split(':', 1)
...         return username == password
...
...     def __call__(self, environ, start_response):
...         if self.is_authenticated(environ.get('HTTP_AUTHORIZATION')):
...             return self.wsgi(environ, start_response)
...         start_response('401 Authentication Required',
...             [('Content-Type', 'text/html'),
...              ('WWW-Authenticate', 'Basic realm="Login"')])
...         return [b'Login']

>>> middlewares = [AuthMiddleware]

# Create an application factory
>>> app_factory = FlaskFactory(__name__, middlewares=middlewares)

# Declares a click application using ConsenSys-Utils click group
>>> cli = FlaskGroup(app_factory=app_factory)

Add Flasks Extension

You can declare your own flask extensions

>>> from consensys_utils.flask import FlaskFactory
>>> from consensys_utils.flask.cli import FlaskGroup
>>> from flasgger import Swagger

>>> swag = Swagger(template={'version': '0.3.4-dev'})

>>> my_extensions = [swag]

# Create an application factory
>>> createapp_factory_app = FlaskFactory(__name__, extensions=my_extensions)

# Declares a click application using ConsenSys-Utils click group
>>> cli = FlaskGroup(app_factory=app_factory)

consensys_utils.flask.FlaskFactory also extensions given as a function taking a flask.Flask application as an argument

>>> from consensys_utils.flask import FlaskFactory
>>> from consensys_utils.flask.cli import FlaskGroup

>>> def init_login_extension(app):
...     if app.config.get('LOGIN'):
...         from flask_login import LoginManager
...
...         login_manager = LoginManager()
...         login_manager.init_app(app)

>>> my_extensions = [init_login_extension]

# Create an application factory
>>> app_factory = FlaskFactory(__name__, extensions=my_extensions)

# Declares a click application using ConsenSys-Utils click group
>>> cli = FlaskGroup(app_factory=app_factory)

It allows you to implement advanced extension initialization based on application configuration. In particular in the example above it allows to allows user having ‘Flask-Login’ installed on option, only users having activated a LOGIN configuration need to have ‘Flask-Login’ installed.

Set Application Hooks

>>> from consensys_utils.flask import FlaskFactory
>>> from consensys_utils.flask.cli import FlaskGroup

>>> def set_log_request_hook(app):
...     @app.before_request
...     def log_request():
...         current_app.logger.debug(request)

>>> my_hook_setters = [set_log_request_hook]

# Create an application factory
>>> app_factory = FlaskFactory(__name__, hook_setters=my_hook_setters)

# Declares a click application using ConsenSys-Utils click group
>>> cli = FlaskGroup(app_factory=app_factory)

Register Blueprints

>>> from flask import Blueprint
>>> from consensys_utils.flask import FlaskFactory
>>> from consensys_utils.flask.cli import FlaskGroup


>>> my_bp1 = Blueprint('my-bp1', __name__)
>>> my_bp2 = Blueprint('my-bp2', __name__)

>>> blueprints = [
...        my_bp1,
...     lambda app: app.register_blueprint(my_bp2),
... ]

# Create an application factory
>>> app_factory = FlaskFactory(__name__, blueprints=blueprints)

# Declares a click application using ConsenSys-Utils click group
>>> cli = FlaskGroup(app_factory=app_factory)

Declare custom CLI commands

It is highly recommended that you declare custom CLI commands directly on the consensys_utils.flask.cli.FlaskGroup object. It automatically injects a --config option to the command for configuration file.

>>> from flask import Blueprint
>>> from flask.cli import with_appcontext
>>> from consensys_utils.flask import FlaskFactory
>>> from consensys_utils.flask.cli import FlaskGroup

# Create an application factory
>>> app_factory = FlaskFactory(__name__)

# Declares a click application using ConsenSys-Utils click group
>>> cli = FlaskGroup(app_factory=app_factory)

>>> @cli.command('test')
... @with_appcontext
... def custom_command():
...    click.echo('Test Command on %s' % current_app.import_name)

Properly manage process to execute an iterator

Quickstart

ConsenSys-Utils provides some resources to properly maintain the execution of an iterator. In particular it allows to

1. Run the iterator with a Gunicorn worker in a properly maintained process
2. Connect a Flask application to the iterator enabling external control on iterator state

It relies on two main resources

• consensys_utils.flask.extensions.iterable.FlaskIterable() that allows to transform a Flask application into an Iterable
• consensys_utils.gunicorn.workers.SyncIterableWorker() that allows to properly maintain a loop on an iterable WSGI object

1. Create a app.py

   >>> from flask import Flask
   >>> from consensys_utils.flask.extensions.iterable import FlaskIterable
   >>> from consensys_utils.flask import FlaskFactory
   >>> from consensys_utils.flask.cli import FlaskGroup
   
   # Create an iterator
   >>> iterator = iter(range(3))
   
   # Create an app factory and extend it to make it with a FlaskIterable extension
   >>> iterable = FlaskIterable(iterator)
   >>> app_factory = FlaskFactory(__name__, extensions=[iterable])
   
   # Declares a click application using ConsenSys-Utils click group
   >>> cli = FlaskGroup(app_factory=app_factory)
   

2. Set a config.yml choosing a consensys_utils.gunicorn.workers.SyncIterableWorker() Gunicorn worker allowing to iterate ono the

   flask:
     base:
       APP_NAME: Iterating-App
   gunicorn:
     worker-processes:
       worker_class: consensys_utils.gunicorn.workers.SyncIteratingWorker
   

3. Define application entry point and start application as described in Create Flask Application Quickstart

Advanced usage

For an advance use-case you can refer to the next example

"""
    examples.iterable
    ~~~~~~~~~~~~~~~~~

    Implement an example of properly managing an iterator using Flask-Iterable and Gunicorn

    :copyright: Copyright 2017 by ConsenSys France.
    :license: BSD, see LICENSE for more details.
"""

import logging
import os

from cfg_loader.utils import parse_yaml_file
from flask import current_app, jsonify
from gunicorn.app.base import BaseApplication

from consensys_utils.flask import Flask
from consensys_utils.flask.extensions.iterable import FlaskIterable
from consensys_utils.gunicorn.workers import PauseIteration

logger = logging.getLogger('examples.iterable')
LOGGING_FILE = os.path.join(os.path.dirname(__file__), 'logging.yml')


# Declare an iterator that we want to properly managed using Gunicorn
class Iterator:
    def __init__(self):
        self.meter = 0

    def set_config(self, config):
        self.meter = config['meter']

    def __iter__(self):
        return self

    def __next__(self):
        logger.info('Iterator.__next__ meter=%s' % self.meter)
        self.meter += 1
        if self.meter % 2 == 0:
            # Indicating the running loop to pause iteration for 2 secs
            raise PauseIteration(2)

        if self.meter >= 100:
            raise StopIteration()


# We declare a Flask application and extend it to make it iterable
iterable_app = Flask(__name__)
iterable_app.config['meter'] = 10
FlaskIterable(Iterator, iterable_app)


# We declare routes on Flask application to interact with the iterator
@iterable_app.route('/get')
def get():
    """Get current value of the iterator meter"""
    logger.info('app.get meter=%s' % current_app.iterator.meter)
    return jsonify({'data': current_app.iterator.meter})


@iterable_app.route('/set/<int:meter>')
def set(meter=0):
    """Set current value of the iterator meter"""
    current_app.iterator.meter = rv = meter
    logger.info('app.set meter=%s' % current_app.iterator.meter)
    return jsonify({'data': rv})


# We declare a custom Gunicorn application for the only matter of the example
class Application(BaseApplication):
    def load(self):
        return iterable_app

    def load_config(self):
        self.cfg.set('logconfig_dict', parse_yaml_file(LOGGING_FILE))
        # We use specific ConsenSys-Utils worker class
        self.cfg.set('worker_class', 'consensys_utils.gunicorn.workers.SyncIteratingWorker')


if __name__ == "__main__":
    # Run iterator
    app = Application()
    app.run()