我是 python 生態系統和 Web 開發的新手，我想用Flask框架構建一個應用程式。
此應用程式必須執行后臺任務。為此，我選擇使用Huey任務佇列。
后臺任務必須對資料庫執行一些查詢。為此，我選擇了Flask-SQLAlchemy。
我設法在休伊工人身上執行了我的任務：
NFO:huey.consumer:MainThread:The following commands are available:
app.tasks.my_task
INFO:huey:Worker-1:Executing app.tasks.my_task: c5dd18bc-df2e-4380-9c1f-b597d2924ba2
但是會出現以下錯誤：
/huey/api.py", line 379, in _execute task_value = task.execute()
...
...
flask_sqlalchemy/__init__.py", line 1042, in get_app raise RuntimeError(
RuntimeError: No application found. Either work inside a view function or push an application context. See http://flask-sqlalchemy.pocoo.org/contexts/.
這是我的專案結構：
app/
├── config.py
├── __init__.py
├── models.py
├── tasks.py
├── views.py
└─── foo.db
這是我的代碼：
#__init__.py
from flask import Flask
from app.config import db, Config, huey
from app.tasks import my_task
def create_app():
app = Flask(__name__)
app.config.from_object(Config)
db.init_app(app)
# register blueprints
from app.views import main as main_blueprint
app.register_blueprint(main_blueprint)
return app
#config.py
from flask_sqlalchemy import SQLAlchemy
from huey import RedisHuey
huey = RedisHuey(__name__, host="localhost")
db = SQLAlchemy()
class Config:
SQLALCHEMY_DATABASE_URI = "sqlite:///foo.db"
SQLALCHEMY_TRACK_MODIFICATIONS = False
#tasks.py
from app.config import huey
from app.models import User
@huey.task()
#@huey.context_task(??)
def background_task():
print("Database query:")
User.query.get(1) # Here is the problem
return 1
#view.py
from flask import Blueprint
from app.tasks import my_task
main = Blueprint("main", __name__)
@main.route("/")
def index():
background_task() # running the registered background task
return "hello view"
#models.py
from app.config import db
class User(db.Model):
def __init__(self, username: str):
self.username = username
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(64), unique=True, nullable=False)
我瀏覽了燒瓶應用程式背景關系檔案：
https://flask.palletsprojects.com/en/2.1.x/appcontext/
還有關于分片資源的huey檔案：
https://huey.readthedocs.io/en/latest/shared_resources.html
我知道我必須以某種方式向作業人員提供應用程式背景關系，但我無法將這些部分連接在一起。
我也試過這個
from flask import current_app
@huey.task()
def my_task():
with current_app.app_context():
print("Database query:")
User.query.get(1)
return 1
它給了我這個錯誤：
/flask/globals.py", line 47, in _find_app raise RuntimeError(_app_ctx_err_msg)
RuntimeError: Working outside of application context.
uj5u.com熱心網友回復：
最好讓 Huey 創建一個 Flask 應用程式供其使用。按如下方式組織您的代碼：
添加第二種方法，為 Huey 的特定用途創建 Flask 應用程式，類似于create_app
#__init__.py
from flask import Flask
from app.config import db, Config, huey
from app.tasks import my_task
def create_app():
# ...
return app
def create_huey_app():
app = Flask('HUEY APP')
app.config.from_object(Config)
# only initialize stuff here that is needed by Huey, eg DB connection
db.init_app(app)
# register any blueprints needed
# e.g. maybe it needs a blueprint to work with urls in email generation
return app
使您的所有任務都有一個可呼叫的第一個引數：
#tasks.py
from app.config import huey
from app.models import User
# every task takes an app_factory parameter
@huey.task()
def background_task(app_factory):
app = app_factory()
with app.app_context():
User.query.get(1)
return 1
然后create_huey_app作為可呼叫物件傳遞給每個任務實體：
#view.py
from flask import Blueprint
from app.tasks import my_task
main = Blueprint("main", __name__)
@main.route("/")
def index():
# pass the Huey app factory to the task
background_task(create_huey_app) # running the registered background task
return "hello view"
如果要在除錯時在本地運行任務：
@main.route("/")
def index():
# pass the Huey app factory to the task
if current_app.debug:
background_task.call_local(create_huey_app) # running the task directly
else:
background_task(create_huey_app) # running the task in Huey
return "hello view"
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/478376.html
標籤：Python 烧瓶 烧瓶-sqlalchemy 蟒蛇休伊
上一篇：如何在燒瓶中使用自定義名稱將影像保存在上傳檔案夾中？
下一篇：限制用戶訪問燒瓶中的路由