我正在嘗試創建一個燒瓶應用程式，該應用程式具有每 10 分鐘運行一次的后臺任務。
我找到了這個模塊：https ://medium.com/greedygame-engineering/an-elegant-way-to-run-periodic-tasks-in-python-61b7c477b679
所以我將它添加到我的專案中：
poetry add timeloop
我使用Flasky代碼示例創建了層次結構（我沒有這本書）：https ://github.com/miguelgrinberg/flasky
# app/__init__.py
from flask import Flask
from timeloop import Timeloop
from app.utils.json_utils import JSON_Custom
from flask_bootstrap import Bootstrap
from config import config
bootstrap = Bootstrap() : Bootstrap
def create_app(config_name): -> Flask
app = Flask(__name__) : Flask
app.config.from_object(config[config_name])
app.json_encoder = JSON_Custom
config[config_name].init_app(app)
bootstrap.init_app(app)
tl = Timeloop() : Timeloop
tl.start()
with app.app_context():
app.config['TIME_LOOP'] = tl : Timeloop
from .utils import utils as utils_blueprint
app.register_blueprint(utils_blueprint)
from .api import api as api_blueprint
app.register_blueprint(api_blueprint, url_prefix='/api/v1')
from .passwd import passwd as passwd_blueprint
app.register_blueprint(passwd_blueprint)
return app
如您所見，我正在嘗試將 timeloop 類實體添加到背景關系中，以便稍后參考它。
現在在我的另一個檔案中
#app/utils/connection_utils.py
@current_app.config['TIME_LOOP'].job(interval=timedelta(minutes=10))
def synchronize_applications():
try:
#code
except:
#code
finally:
current_app.config['TIME_LOOP'].stop()
現在的問題是我已經嘗試了很多方法，每次我得到不同的錯誤。有了這個設定，我得到
raise RuntimeError(unbound_message) from None
RuntimeError: Working outside of application context.
This typically means that you attempted to use functionality that needed
the current application. To solve this, set up an application context
with app.app_context(). See the documentation for more information.
編輯：
我也嘗試了以下方法。
#app/models.py
from timeloop import Timeloop
tl = Timeloop()
#app/__init__.py
def create_app(config_name):
...
from app.models import tl
tl.start()
...
#app/utils/connection_utils.py
from app.models import tl
@tl.job(interval=timedelta(minutes=10))
def synchronize_application():
with current_app.app_context():
...
現在flask APP啟動了，但是遇到TL就崩潰了。
[2022-10-26 12:54:11,626] [timeloop] [INFO] Starting Timeloop..
[2022-10-26 12:54:11,626] [timeloop] [INFO] Registered job <function synchronize_applications at 0x7fa1640984c0>
[2022-10-26 12:54:11,626] [timeloop] [INFO] Timeloop now started. Jobs will run based on the interval set
* Serving Flask app 'main.py'
* Debug mode: off
WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead.
* Running on http://127.0.0.1:5000
Press CTRL C to quit
Exception in thread Thread-1:
Traceback (most recent call last):
...
raise RuntimeError(unbound_message) from None
RuntimeError: Working outside of application context.
This typically means that you attempted to use functionality that needed
the current application. To solve this, set up an application context
with app.app_context(). See the documentation for more information.
我認為這是因為它在單獨的執行緒中運行？
uj5u.com熱心網友回復：
最后的解決方案是這樣的：
我創建了一個后臺任務類，它接受當前應用程式和存盤作業的時間回圈物件。
https://github.com/Ruggiero-Santo/timeloop
from app.models import timeloop
bootstrap = Bootstrap() : Bootstrap
def create_app(config_name): -> Flask
app = Flask(__name__) : Flask
# Import config
app.config.from_object(config[config_name])
app.json_encoder = JSON_Custom
config[config_name].init_app(app)
# Add bootstrap
bootstrap.init_app(app)
# Pass the App and the timeloop object created to the backgroud tasks.
background_tasks = BackgroundTasks(app, timeloop) : BackgroundTasks
# Start the background tasks.
background_tasks.start()
from .utils import utils as utils_blueprint
app.register_blueprint(utils_blueprint)
from .api import api as api_blueprint
app.register_blueprint(api_blueprint, url_prefix='/api/v1')
from .passwd import passwd as passwd_blueprint
app.register_blueprint(passwd_blueprint)
return app
class BackgroundTasks(threading.Thread):
"""Class that runs background tasks for a flask application.
Args:
threading.Thread: Creates a new thread.
"""
def __init__(self, app: Flask, timeloop: Timeloop): -> None
"""Create a background tasks object that runs periodical tasks in the background of an flask application.
Args:
app: Flask application object.
timeloop: Timeloop object.
"""
super(BackgroundTasks, self).__init__()
self.app = app : Flask
self.timeloop = timeloop : Timeloop
self.timeloop.add_job(self.synchronize_applications, interval=timedelta(minutes=10), exception=False)
def run(self): -> None
# Use the current application context and start the timeloop service.
with self.app.app_context():
self.timeloop.start()
def synchronize_applicatons(self): -> None
#do stuff
所以最后我沒有使用裝飾器，但是我學到了很多，發現了這個很棒的模塊：
https://github.com/GrahamDumpleton/wrapt
https://www.youtube.com/watch?v=W7Rv-km3ZuA&t=32s
轉載請註明出處，本文鏈接：https://www.uj5u.com/ruanti/522527.html
標籤：Python烧瓶