Flask で時間のかかる処理（メール送信、画像処理、外部 API 呼び出しなど）をリクエスト内で実行すると、レスポンスが遅くなる。Celery を使えば、これらの処理をバックグラウンドで非同期実行できる。
Celery のセットアップ
pip install celery redis
# app/celery_app.py from celery import Celery def make_celery(app): celery = Celery( app.import_name, backend=app.config['CELERY_RESULT_BACKEND'], broker=app.config['CELERY_BROKER_URL'] ) celery.conf.update(app.config) class ContextTask(celery.Task): def __call__(self, *args, **kwargs): with app.app_context(): return self.run(*args, **kwargs) celery.Task = ContextTask return celery
Flask アプリとの統合
# app/__init__.py from flask import Flask from app.celery_app import make_celery def create_app(config_name='default'): app = Flask(__name__) app.config.from_object(f'config.{config_name}') return app app = create_app() celery = make_celery(app)
# config.py class Config: CELERY_BROKER_URL = 'redis://localhost:6379/0' CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
タスクの定義
# app/tasks.py from app import celery from app.services.email_service import send_email_sync @celery.task def send_email_task(to, subject, body): send_email_sync(to, subject, body) @celery.task(bind=True, max_retries=3) def process_image_task(self, image_id): try: # 画像処理 process_image(image_id) except Exception as e: # リトライ self.retry(exc=e, countdown=60)
ビューからタスクを呼び出す
from app.tasks import send_email_task @app.route('/register', methods=['POST']) def register(): user = create_user(request.form) # 非同期でメール送信 send_email_task.delay( to=user.email, subject='Welcome!', body='Thank you for registering.' ) return 'Registration complete'
.delay() を呼ぶと、タスクがキューに追加され、すぐにレスポンスを返せる。
Celery ワーカーの起動
celery -A app.celery worker --loglevel=info
タスクの状態確認
from app.tasks import process_image_task @app.route('/process/<int:image_id>') def process(image_id): task = process_image_task.delay(image_id) return {'task_id': task.id} @app.route('/status/<task_id>') def task_status(task_id): task = process_image_task.AsyncResult(task_id) return { 'state': task.state, 'result': task.result if task.ready() else None }
定期実行（Celery Beat）
# config.py from celery.schedules import crontab CELERY_BEAT_SCHEDULE = { 'cleanup-every-hour': { 'task': 'app.tasks.cleanup_task', 'schedule': crontab(minute=0), # 毎時0分 }, }
celery -A app.celery beat --loglevel=info
本番環境での注意点
| Redis/RabbitMQ | メッセージブローカーの可用性を確保 |
| ワーカー数 | CPU コア数に応じて調整 |
| 監視 | Flower などでタスクを監視 |
Celery を導入することで、Flask アプリケーションのレスポンス性能を維持しつつ、重い処理をバックグラウンドで実行できる。
