我正在执行一项任务，需要使用 Celery Beat 计划安排我的应用程序定期执行。
我面临的主要问题是 Celery 没有在指定的时间间隔触发我的应用程序。
有人可以帮我吗？
from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime
import requests
import xmltodict
import json
import logging
import os
from celery.schedules import crontab
from celery import Celery
from .celery import make_celery
from datetime import timedelta
db = SQLAlchemy()
class Event(db.Model):
id = db.Column(db.Integer, primary_key=True)
provider_id = db.Column(db.Integer, unique=False, nullable=False)
name = db.Column(db.String(255), nullable=False)
started_at = db.Column(db.DateTime, nullable=False)
end_at = db.Column(db.DateTime, nullable=False)
sell_mode = db.Column(db.String(50), nullable=False)
created_at = db.Column(db.DateTime, nullable=False, default=db.func.current_timestamp())
def as_dict(self):
"""
Converts the Event object to a dictionary representation.
Returns:
dict: Dictionary representation of the Event object.
"""
return {
"provider_id": self.provider_id,
"name": self.name,
"datetime": self.started_at.isoformat(),
"sell_mode": self.sell_mode
}
def remove_at_sign(obj):
"""
Recursively removes '@' sign from keys
"""
if isinstance(obj, dict):
return {key.lstrip('@'): remove_at_sign(val) for key, val in obj.items()}
elif isinstance(obj, list):
return [remove_at_sign(elem) for elem in obj]
else:
return obj
def fetch_events():
"""
Fetches events data from an external API and processes it.
Returns:
list: List of Event objects fetched from the API.
"""
PROVIDER_URL = 'some-url'
response = requests.get(PROVIDER_URL)
events = []
if response.status_code == 200:
json_resp = xmltodict.parse(response.text)
final_json = remove_at_sign(json_resp)
with open('events.json', 'a') as f1:
json.dump(final_json, f1, indent=4)
final_data = final_json['eventList']['output']['base_event']
for ind, data in enumerate(final_data):
if 'sell_mode' in data and final_data[ind]['sell_mode'] == 'online':
sell_mode = final_data[ind]['sell_mode']
title = data['title'] if 'title' in data else str()
start_date = datetime.fromisoformat(data['event']['event_start_date'])
end_date = datetime.fromisoformat(data['event']['event_end_date'])
event_id = data['event']['event_id']
events.append(Event(provider_id=event_id, name=title, end_at=end_date, sell_mode=sell_mode, created_at=datetime.now(), started_at=start_date))
print([sell_mode, title, start_date, end_date, event_id])
return events
def store_events(events, app):
"""
Stores events in the database.
Args:
events (list): List of Event objects to be stored.
app (Flask): Flask application instance.
"""
with app.app_context():
try:
db.session.add_all(events)
db.session.commit()
logging.info(f"Stored {len(events)} new events in the database.")
except Exception as e:
db.session.rollback()
logging.error(f"Failed to store events: {str(e)}")
raise
def fetch_and_store(app):
with app.app_context():
db.create_all()
events = fetch_events()
store_events(events, app)
def create_app():
# import ipdb;ipdb.set_trace()
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///events.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config.update(
CELERY_BROKER_URL='redis://127.0.0.1:6379/0',
CELERY_RESULT_BACKEND='redis://127.0.0.1:6379/0'
)
db.init_app(app)
celery = make_celery(app)
celery_worker = 'worker1@hostname'
# Configure logging
logging.basicConfig(level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
log_file = os.path.join(app.root_path, 'app.log')
file_handler = logging.FileHandler(log_file)
file_handler.setLevel(logging.INFO)
file_handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s'))
logging.getLogger().addHandler(file_handler)
logger = logging.getLogger(__name__)
with app.app_context():
db.create_all()
@celery.task(name='app.scheduled_fetch_and_store')
def scheduled_fetch_and_store():
fetch_and_store(app)
celery.conf.beat_schedule = {
'fetch-events-every-10-seconds': {
'task': 'app.scheduled_fetch_and_store',
'schedule': timedelta(seconds=10),
},
}
@app.route('/events', methods=['GET'])
def get_events():
"""
Retrieves events filtered by start and end dates.
Returns:
jsonify: JSON response containing events data.
"""
starts_at = request.args.get('starts_at')
ends_at = request.args.get('ends_at')
logger.info(f"Fetching events between {starts_at} and {ends_at}")
if not starts_at or not ends_at:
logger.error("Both 'starts_at' and 'ends_at' parameters are required")
return jsonify({"error": "Both 'starts_at' and 'ends_at' parameters are required"}), 400
try:
starts_at = datetime.fromisoformat(starts_at)
ends_at = datetime.fromisoformat(ends_at)
except ValueError:
logger.error("'starts_at' and 'ends_at' must be in ISO format")
return jsonify({"error": "'starts_at' and 'ends_at' must be in ISO format"}), 400
events = Event.query.filter(Event.started_at >= starts_at, Event.end_at <= ends_at).all()
result = [
{
"id": event.provider_id,
"name": event.name,
"start_time": event.started_at.isoformat(),
"end_time": event.end_at.isoformat(),
"sell_mode": event.sell_mode
} for event in events
]
with open('sample.json', 'w') as json_file:
json.dump(result, json_file, indent=4)
return jsonify(result)
@app.route('/routes', methods=['GET'])
def list_routes():
import urllib.parse
output = []
for rule in app.url_map.iter_rules():
options = {}
for arg in rule.arguments:
options[arg] = f"[{arg}]"
methods = ','.join(rule.methods)
url = urllib.parse.unquote(f"{rule}")
line = f"{rule.endpoint:50s} {methods:20s} {url}"
output.append(line)
return "<pre>" + "\n".join(sorted(output)) + "</pre>"
return app
if __name__ == "__main__":
app = create_app()
app.run(debug=True)
`我已经使用 Celery 设置了一个计划，每 10 秒执行一次我的应用程序，但它不起作用。
我在终端中使用了这些命令
它向我抛出此类警告：
\[2024-07-10 16:55:53,498: INFO/MainProcess\] Connected to amqp://guest:\*\*@127.0.0.1:5672//
\[2024-07-10 16:55:53,499: WARNING/MainProcess\] /home/bhupesh/.local/lib/python3.10/site-packages/celery/worker/consumer/consumer.py:507: CPendingDeprecationWarning: The broker_connection_retry configuration setting will no longer determine
whether broker connection retries are made during startup in Celery 6.0 and above.
If you wish to retain the existing behavior for retrying connections on startup,
you should set broker_connection_retry_on_startup to True.
warnings.warn(
\[2024-07-10 16:55:53,534: INFO/MainProcess\] mingle: searching for neighbors
\[2024-07-10 16:55:53,543: WARNING/MainProcess\] No hostname was supplied. Reverting to default 'localhost'
\[2024-07-10 16:55:54,571: INFO/MainProcess\] mingle: all alone
\[2024-07-10 16:55:54,624: INFO/MainProcess\] celery@vijay-latitude-7480 ready.
- celery -芹菜工人 --beat --loglevel=info
你不应该像这样运行 Celery 吗：
celery -A celery beat -l infohttps://docs.celeryq.dev/en/stable/userguide/periodic-tasks.html#starting-the-scheduler