How to use the gunicorn.app.base.BaseApplication function in gunicorn

)


def on_exit(server: Arbiter):
    """Gunicorn shutdown tasks."""

    log.critical("Stopping hyperglass {}", __version__)

    async def runner():
        if not params.developer_mode:
            await clear_cache()

    aiorun(runner())


class HyperglassWSGI(BaseApplication):
    """Custom gunicorn app."""

    def __init__(self, app, options):
        """Initialize custom WSGI."""
        self.application = app
        self.options = options or {}
        super().__init__()

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

'user': config.get('user'),
    'group': config.get('group'),

    'raw_env': 'APP_MODE={}'.format(app_mode),

    'preload_app': preload_app,
    'daemon': app_mode=='prod',
    'reload': app_mode=='dev' and not preload_app,

    'pidfile': '{}/fcrepo.pid'.format(run_dir),
    'accesslog': '{}/gunicorn-access.log'.format(log_dir),
    'errorlog': '{}/gunicorn-error.log'.format(log_dir),
}

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

    def __init__(self, app, options={}):
        self.options = options
        self.application = app
        super(WsgiApp, self).__init__()

    def load_config(self):
        for key, value in self.options.items():
            self.cfg.set(key.lower(), value)

    def load(self):
        return self.application


def run():
    WsgiApp(fcrepo, options).run()

from gunicorn.app.base import BaseApplication


class GunicornApp(BaseApplication):
    """Convert a Flask application to a Gunicorn one.
    """

    def __init__(self, flask_app, settings=None):
        """Initialize GunicornApp.

        If no settings are provided the class is initialized using the
        documented default parameters in
        http://docs.gunicorn.org/en/stable/settings.html#config-file.

        Args:
            flask_app (flask.app.Flask): Application to be wrapped by
                gunicorn.
            settings (dict): Settings defining the configuration to use
                when lounching the gunicorn application. If any setting
                is missing, the corresponding the default value is used.

port = app.read_config('Server', 'port', 'AUGUR_PORT', '5000')
    workers = int(app.read_config('Server', 'workers', 'AUGUR_WORKERS', mp.cpu_count()))
    options = {
        'bind': '%s:%s' % (host, port),
        'workers': workers,
        'accesslog': '-',
        'access_log_format': '%(h)s - %(t)s - %(r)s',
    }
    logger.info('Starting server...')
    master = Arbiter(AugurGunicornApp(options, manager=manager, broker=broker, housekeeper=housekeeper)).run()

def worker_start(worker_name=None, instance_number=0):
    time.sleep(45 * instance_number)
    process = subprocess.Popen("cd workers/{} && {}_start".format(worker_name,worker_name), shell=True)

class AugurGunicornApp(gunicorn.app.base.BaseApplication):
    """
    Loads configurations, initializes Gunicorn, loads server
    """

    def __init__(self, options=None, manager=None, broker=None, housekeeper=None):
        self.options = options or {}
        self.manager = manager
        self.broker = broker
        self.housekeeper = housekeeper
        super(AugurGunicornApp, self).__init__()
        # self.cfg.pre_request.set(pre_request)

    def load_config(self):
        """
        Sets the values for configurations
        """

# source: http://docs.gunicorn.org/en/stable/custom.html
from __future__ import unicode_literals

import multiprocessing

import gunicorn.app.base
from twootfeed import create_app, param


class StandaloneApplication(gunicorn.app.base.BaseApplication):
    def __init__(self, current_app, options=None):
        self.options = options or {}
        self.application = current_app
        super().__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):
        return self.application

import gunicorn.app.base
from gunicorn.six import iteritems


class FalconGunicorn(gunicorn.app.base.BaseApplication):
    '''Utility to deploy falcon.API on gunicorn'''

    def __init__(self, app, options=None):
        '''Constructor
        Args:
            app (falcon.API): a routed falcon API
            options (dict): a set of options for gunicorn (e.g. workers, port, etc)
        '''
        self.options = options or {}
        self.application = app
        super(FalconGunicorn, 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):

@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

#
# You should have received a copy of the GNU Lesser General Public
# License along with this library.

from gunicorn.app.base import BaseApplication
from gunicorn.glogging import Logger
from werkzeug.wrappers import Request, Response
from werkzeug.utils import escape
from werkzeug.exceptions import HTTPException, InternalServerError

from oio.common.utils import CPU_COUNT
from oio.common.configuration import read_conf
from oio.common.logger import get_logger


class Application(BaseApplication):
    access_log_fmt = '%(bind0)s %(h)s:%({remote}p)s %(m)s %(s)s %(D)s ' + \
                     '%(B)s %(l)s %({x-oio-req-id}i)s %(U)s?%(q)s'

    def __init__(self, app, conf, logger_class=None):
        self.conf = conf
        self.application = app
        self.logger_class = logger_class
        super(Application, self).__init__()

    def load_config(self):
        bind = '%s:%s' % (self.conf.get('bind_addr', '127.0.0.1'),
                          self.conf.get('bind_port', '8000'))
        self.cfg.set('bind', bind)
        self.cfg.set('backlog', self.conf.get('backlog', 2048))
        self.cfg.set('workers', self.conf.get('workers', CPU_COUNT))
        self.cfg.set('worker_class', self.conf.get('worker_class', 'eventlet'))