How to use the gunicorn.app function in gunicorn

#
# This file is part of gunicorn released under the MIT license.
# See the NOTICE for more information.

import os

try:
    import unittest.mock as mock
except ImportError:
    import mock

import gunicorn.app.base
import gunicorn.arbiter


class DummyApplication(gunicorn.app.base.BaseApplication):
    """
    Dummy application that has an default configuration.
    """

    def init(self, parser, opts, args):
        """No-op"""

    def load(self):
        """No-op"""

    def load_config(self):
        """No-op"""


def test_arbiter_shutdown_closes_listeners():
    arbiter = gunicorn.arbiter.Arbiter(DummyApplication())

def run_server(self, app, host, port):
        import gunicorn.app.base

        class FlaskGUnicornApp(gunicorn.app.base.BaseApplication):
            options = {
                'bind': '{}:{}'.format(host, port),
                'workers': self.processes
            }

            def load_config(self):
                for k, v in self.options.items():
                    self.cfg.set(k.lower(), v)

            def load(self):
                return app

        FlaskGUnicornApp().run()

from __future__ import unicode_literals, absolute_import

# Based on http://docs.gunicorn.org/en/19.6.0/custom.html

import multiprocessing

import gunicorn.app.base
from six import iteritems
from fireworks.flask_site.app import app as handler_app


def number_of_workers():
    return (multiprocessing.cpu_count() * 2) + 1


class StandaloneApplication(gunicorn.app.base.BaseApplication):

    def __init__(self, app, options=None):
        self.options = options or {}
        self.application = app
        super(StandaloneApplication, self).__init__()

    def load_config(self):
        config = dict([
            (key, value) for key, value in iteritems(self.options)
            if key in self.cfg.settings and value is not None
        ])
        for key, value in iteritems(config):
            self.cfg.set(key.lower(), value)

    def load(self):
        return self.application

import copy
import multiprocessing

from flask import Flask
from flask_cors import CORS
import gunicorn.app.base

from bigchaindb import utils
from bigchaindb import BigchainDB
from bigchaindb.web.routes import add_routes
from bigchaindb.web.strip_content_type_middleware import StripContentTypeMiddleware


# TODO: Figure out if we do we need all this boilerplate.
class StandaloneApplication(gunicorn.app.base.BaseApplication):
    """Run a **wsgi** app wrapping it in a Gunicorn Base Application.

    Adapted from:
     - http://docs.gunicorn.org/en/latest/custom.html
    """

    def __init__(self, app, *, options=None):
        """Initialize a new standalone application.

        Args:
            app: A wsgi Python application.
            options (dict): the configuration.

        """
        self.options = options or {}
        self.application = app

from flask_app.models import db
    import flask_migrate
    import gunicorn.app.base

    _ensure_conf()

    app = create_app(config={'PROPAGATE_EXCEPTIONS': True})

    flask_migrate.Migrate(app, db)

    with app.app_context():
        flask_migrate.upgrade()

    workers_count = (multiprocessing.cpu_count() * 2) + 1

    class StandaloneApplication(gunicorn.app.base.BaseApplication):

        def __init__(self, app, options=None):
            self.options = options or {}
            self.application = app
            super(StandaloneApplication, self).__init__()

        def load_config(self):
            config = dict([(key, value) for key, value in self.options.items()
                           if key in self.cfg.settings and value is not None])
            for key, value in config.items():
                self.cfg.set(key.lower(), value)

        def load(self):
            return self.application

    options = {

tasks))
                if len(matched) > 0:
                    worker_components.append(matched[0]['Containers'][0]['Arn'])
    except:
        pass

    ## fallback to IP
    if len(worker_components) == 0:
        return socket.gethostbyname(socket.gethostname())
    else:
        return '-'.join(worker_components)

def number_of_workers():
    return (multiprocessing.cpu_count() * 2) + 1

class StandaloneApplication(gunicorn.app.base.BaseApplication):

    def __init__(self, app, options=None):
        self.options = options or {}
        self.application = app
        super(StandaloneApplication, self).__init__()

    def load_config(self):
        config = dict([(key, value) for key, value in iteritems(self.options)
                       if key in self.cfg.settings and value is not None])
        for key, value in iteritems(config):
            self.cfg.set(key.lower(), value)

    def load(self):
        return self.application

@click.group()

#!/usr/bin/env python

# Copyright (c) LinkedIn Corporation. All rights reserved. Licensed under the BSD-2 Clause license.
# See LICENSE in the project root for license information.

import sys
import yaml
import multiprocessing
import gunicorn.app.base


class StandaloneApplication(gunicorn.app.base.BaseApplication):

    def __init__(self, options=None, skip_build_assets=False):
        self.options = options or {}
        self.skip_build_assets = skip_build_assets
        super(StandaloneApplication, self).__init__()

    def load_config(self):
        config = {key: value for key, value in self.options.items()
                  if key in self.cfg.settings and value is not None}
        for key, value in config.items():
            self.cfg.set(key.lower(), value)

    def load(self):
        from iris_relay.app import get_relay_app
        with open(sys.argv[1]) as config_file:
            config = yaml.safe_load(config_file)

from __future__ import unicode_literals

import os
import multiprocessing
import sys

import gunicorn.app.base
from gunicorn.six import iteritems
from pyramid.paster import get_app


def number_of_workers():
    return (multiprocessing.cpu_count() * 2) + 1


class StandaloneApplication(gunicorn.app.base.BaseApplication):

    def __init__(self, conf, options=None):
        self.options = options or {}
        self.application = get_app(conf)
        super(StandaloneApplication, self).__init__()

    def load_config(self):
        config = dict([(key, value) for key, value in iteritems(self.options)
                       if key in self.cfg.settings and value is not None])
        for key, value in iteritems(config):
            self.cfg.set(key.lower(), value)

    def load(self):
        return self.application

@app.route("/", defaults={"path": ""})
@app.route("/")
def root(path):
    return flask.render_template("index.html")

try:
    import gunicorn.app.base
    from gunicorn.six import iteritems
    import gunicorn.glogging

    class BypassGunicornLogging(gunicorn.glogging.Logger):
        def setup(self, cfg):
            self.error_log.handlers = logging.root.handlers
            self.access_log.handlers = logging.root.handlers

    class GunicornBrozzlerDashboard(gunicorn.app.base.BaseApplication):
        def __init__(self, app, options=None):
            self.options = options or {}
            self.application = app
            super(GunicornBrozzlerDashboard, self).__init__()

        def load_config(self):
            config = dict(
                    [(key, value) for key, value in iteritems(self.options)
                        if key in self.cfg.settings and value is not None])
            for key, value in iteritems(config):
                self.cfg.set(key.lower(), value)
            self.cfg.set("logger_class", BypassGunicornLogging)
            self.cfg.set("accesslog", "dummy-value")

        def load(self):
            return self.application

except subprocess.TimeoutExpired as e:
                odict['exit_code'] = 1
                odict['output'] = e.output
                self.logger.warn("envoy configuration validation timed out after {} seconds{}\n{}",
                    timeout, ', retrying...' if retry < retries - 1 else '', e.output)
                continue

        if odict['exit_code'] == 0:
            self.logger.info("successfully validated the resulting envoy configuration, continuing...")
            return True

        self.logger.error("{}\ncould not validate the envoy configuration above after {} retries, failed with error \n{}\nAborting update...".format(config_json, retries, odict['output']))
        return False


class StandaloneApplication(gunicorn.app.base.BaseApplication):
    def __init__(self, app, options=None):
        self.options = options or {}
        self.application = app
        super(StandaloneApplication, self).__init__()

        # Boot chime. This is basically the earliest point at which we can consider an Ambassador
        # to be "running".
        scout_result = self.application.scout.report(mode="boot", action="boot1", no_cache=True)
        self.application.logger.info(f'BOOT: Scout result {json.dumps(scout_result)}')

    def load_config(self):
        config = dict([(key, value) for key, value in iteritems(self.options)
                       if key in self.cfg.settings and value is not None])
        for key, value in iteritems(config):
            self.cfg.set(key.lower(), value)