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

我有一个烧瓶应用程序。

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

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() 的实例。我想直接将 app 对象提供给 gunicorn,而不是通过在字符串中对其进行寻址

我尝试过的: 我打开了 gunicorn 库 main.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()
stack overflow How to run gunicorn inside python not as a command line?
原文答案

答案:

作者头像

像这样的东西对我有用。首先我实例化 BaseApplication 类。它有一个 run() 方法。详细信息在 gunicorn 文档中的 how to create a custom application 上。

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)
作者头像

在 my_file.py 中插入以下内容

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

我的目标并不完全相同:我可以将应用程序指定为字符串(与在命令行上使用 gunicorn 相同)而不是传递 python 应用程序对象。实际上,我不确定传递单个应用程序对象是否真的有意义,因为 gunicorn 不应该在它产生的每个工作人员中都有不同的应用程序对象吗?

我主要担心的是我在没有 subprocess 或类似的帮助的情况下运行 gunicorn。我还使用了 FastAPI 而不是 Flask,并且(根据当前推荐的 prod 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 而不是从 BaseApplication 继承,这导致 gunicorn 以与从命令行调用时基本相同的方式启动。

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