我有一个烧瓶应用程序。
我使用以下命令在生产中运行它:
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库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()
这样的东西对我有用。 首先,我实例化 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)
我的目标并不完全相同:我可以将应用程序指定为字符串(与命令行上的 Gunicorn 相同),而不是传递 python 应用程序对象。实际上,我不确定传递单个应用程序对象是否真的有意义,因为gunicorn 不应该在它生成的每个工作人员中拥有不同的应用程序对象吗?
我主要担心的是我在没有
subprocess
或类似工具的帮助下运行 Gunicorn。我还使用 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
而不是 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。
我确认@Jaza 答案对于 FlaskApp 来说效果非常好。 我也 试图找到一个较短的版本。我很快发现
gunicorn.app.base.Application
使用标准 Python ArgumentParser
直接读取 sys.argv
。 所以是的,更短的版本是可能的,但这似乎需要分配给 sys.argv
,我觉得这是一个应该避免的丑陋的黑客行为。以防万一您好奇。
import gunicorn.app.wsgiapp
import multiprocessing
import sys
def run():
sys.argv = [
"-b", "0.0.0.0:8000",
"-w", str((multiprocessing.cpu_count() * 2) + 1),
"-k", "uvicorn.workers.UvicornWorker",
"myproject.main:app"
]
gunicorn.app.wsgiapp.run()
在 my_file.py 中插入以下内容
from subprocess import run
run("gunicorn -w 1 -b 0.0.0.0:5000 'path.to.wsgi:return_app()'".split(' '))