Class Executor

flask_executor/executor.py:59–271

An executor interface for :py:mod:`concurrent.futures` designed for working with Flask applications. :param app: A Flask application instance. :param name: An optional name for the executor. This can be used to configure multiple executors. Named executors will look for

class Executor(InstanceProxy, concurrent.futures._base.Executor):
    """An executor interface for :py:mod:`concurrent.futures` designed for
    working with Flask applications.

    :param app: A Flask application instance.
    :param name: An optional name for the executor. This can be used to
                 configure multiple executors. Named executors will look for
                 environment variables prefixed with the name in uppercase,
                 e.g. ``CUSTOM_EXECUTOR_TYPE``.
    """

    def __init__(self, app=None, name=''):
        self.app = app
        self._default_done_callbacks = []
        self.futures = FutureCollection()
        if re.match(r'^(\w+)?$', name) is None:
            raise ValueError(
                "Executor names may only contain letters, numbers or underscores"
            )
        self.name = name
        if len(name) > 0:
            name = name.upper() + '_'
        self.EXECUTOR_TYPE = name + 'EXECUTOR_TYPE'
        self.EXECUTOR_MAX_WORKERS = name + 'EXECUTOR_MAX_WORKERS'
        self.EXECUTOR_FUTURES_MAX_LENGTH = name + 'EXECUTOR_FUTURES_MAX_LENGTH'
        self.EXECUTOR_PROPAGATE_EXCEPTIONS = name + 'EXECUTOR_PROPAGATE_EXCEPTIONS'
        if app is not None:
            self.init_app(app)

    def init_app(self, app):
        """Initialise application. This will also intialise the configured
        executor type:

            * :class:`concurrent.futures.ThreadPoolExecutor`
            * :class:`concurrent.futures.ProcessPoolExecutor`
        """
        app.config.setdefault(self.EXECUTOR_TYPE, 'thread')
        executor_type = app.config[self.EXECUTOR_TYPE]
        executor_max_workers = default_workers(executor_type)
        app.config.setdefault(self.EXECUTOR_MAX_WORKERS, executor_max_workers)
        futures_max_length = app.config.setdefault(self.EXECUTOR_FUTURES_MAX_LENGTH, None)
        propagate_exceptions = app.config.setdefault(self.EXECUTOR_PROPAGATE_EXCEPTIONS, False)
        if futures_max_length is not None:
            self.futures.max_length = int(futures_max_length)
        if str2bool(propagate_exceptions):
            self.add_default_done_callback(propagate_exceptions_callback)
        self._self = self._make_executor(app)
        app.extensions[self.name + 'executor'] = self

    def _make_executor(self, app):
        executor_type = app.config[self.EXECUTOR_TYPE]
        executor_max_workers = app.config[self.EXECUTOR_MAX_WORKERS]
        if executor_max_workers is not None:
            executor_max_workers = int(executor_max_workers)
        if executor_type == 'thread':
            _executor = concurrent.futures.ThreadPoolExecutor
        elif executor_type == 'process':
            _executor = concurrent.futures.ProcessPoolExecutor