如何在 python 中运行gunicorn而不是作为命令行？

Question

我有一个烧瓶应用程序。

我使用以下命令在生产中运行它：

python -m gunicorn -w 1 -b 0.0.0.0:5000 "path.to.wsgi:return_app()"

相反，我想在 my_file.py 中运行它

我需要一个函数来运行，它应该接受应用程序对象和端口绑定以及工作人员数量

我怎样才能做到这一点？

我需要这样的伪代码：

import gunicorn

app = return_app()

gunicorn(workers=1, ip="0.0.0.0", port=5000, app=app)

对我来说最重要的部分app=app是

要点是我想使用 app 对象作为 Flask() 的实例。我想直接将应用程序对象提供给gunicorn，而不是通过在字符串中寻址它

我尝试过的： 我已经打开了gunicorn库主.py文件

from gunicorn.app.wsgiapp import run
run()

看看它是如何工作的，但无法弄清楚

def run():
    """\
    The ``gunicorn`` command line runner for launching Gunicorn with
    generic WSGI applications.
    """
    from gunicorn.app.wsgiapp import WSGIApplication
    WSGIApplication("%(prog)s [OPTIONS] [APP_MODULE]").run()

Answer 1

像这样的东西对我有用。首先，我实例化 BaseApplication 类。它有一个run()方法。Gunicorn 文档中详细介绍了如何创建自定义应用程序。

if platform.uname().system.lower()=='linux':
    print("Detected Linux, Preparing gunicorn")        
    import gunicorn.app.base
    class StandaloneApplication(gunicorn.app.base.BaseApplication):

        def __init__(self, app, options=None):
            self.options = options or {}
            self.application = 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

if __name__ == "__main__":
    # Use a debugging session in port 5001    
    if platform.uname().system.lower()=='linux':
        print("Detected Linux, Running Gunicorn")
        options = {
            'bind': '%s:%s' % ('0.0.0.0', '5001'),
            'workers': number_of_workers(),
            # 'threads': number_of_workers(),
            'timeout': 120,
        }
        initialize()
        StandaloneApplication(app, options).run()
    else:
        print("Detected non Linux, Running in pure Flask")
        initialize()
        app.run(debug=True, host=socket.gethostbyname(socket.gethostname()), port=5001)

Answer 2

我的目标并不完全相同：我可以将应用程序指定为字符串（与命令行上的 Gunicorn 相同），而不是传递 python 应用程序对象。实际上，我不确定传递单个应用程序对象是否真的有意义，因为gunicorn 不应该在它生成的每个工作人员中拥有不同的应用程序对象吗？

我主要担心的是我在没有或类似的帮助下运行gunicorn subprocess。我还使用 FastAPI 而不是 Flask，并且（根据当前推荐的产品 FastAPI 设置）告诉 Gunicorn 生成 uvicorn 工人。

这就是我最终的结果（在myproject/web.py）：

import multiprocessing

from gunicorn.app.wsgiapp import WSGIApplication


class StandaloneApplication(WSGIApplication):
    def __init__(self, app_uri, options=None):
        self.options = options or {}
        self.app_uri = app_uri
        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 run():
    options = {
        "bind": "0.0.0.0:8000",
        "workers": (multiprocessing.cpu_count() * 2) + 1,
        "worker_class": "uvicorn.workers.UvicornWorker",
    }
    StandaloneApplication("myproject.main:app", options).run()

我的StandaloneApplication答案与 Zaero Divide 在本线程中的答案非常相似（并且基于）。但是，我传递了app_uri，并且继承自WsgiApplication而不是 from BaseApplication，这导致gunicorn 的启动方式与从命令行调用它时的方式基本相同。

注意：在 中myproject/main.py，我有app = FastAPI(). 在pyproject.toml下面[tool.poetry.scripts]，我有web = "myproject.web:run"- 所以我可以启动gunicorn poetry run web。我还在用 构建一个工件shiv -c web myproject.whl -o web.pyz，这样我就可以运行/path/to/web.pyz来启动gunicorn。

Answer 3

在 my_file.py 中插入以下内容

from subprocess import run
run("gunicorn -w 1 -b 0.0.0.0:5000 'path.to.wsgi:return_app()'".split(' '))