Flaskでの静的ファイルの取り扱い
はじめに FlaskはPythonでWebアプリケーションを開発するための軽量で柔軟なフレームワークです。Webサイトを作成する際には、HTMLだけでなくCSS […]
非同期処理とバックグラウンドタスク
2026-03-01はじめに
Webアプリケーションにおいて、時間のかかる処理（画像処理、メール送信、データ分析など）を同期的に実行すると、ユーザーは応答が返るまで待たされることになります。これではユーザー体験が損なわれ、特に大量のリクエストがある場合にはサーバーのパフォーマンスにも悪影響を及ぼします。本章では、非同期処理とバックグラウンドタスクの実装方法について学びます。具体的には、Celeryを使ったタスクキューの導入、メール送信機能の非同期化、そして効率的なバックグラウンド処理システムの構築について詳しく解説します。
Celeryの導入
Celeryとは
CeleryはPython製の分散タスクキューシステムです。複数のワーカーによるタスクの並行処理、RedisやRabbitMQなど多様なブローカーへの対応、タスクスケジューリングを標準機能として備えます。また、タスク失敗時のリトライ機構やエラーハンドリング、実行結果の保存と追跡機能も充実しており、非同期処理の実装を効率化します。
インストールと基本設定
まずは必要なパッケージをインストールします。
# 必要なパッケージのインストール
pip install celery redis flower
# メール送信用
pip install flask-mail
# 開発用（オプション）
pip install celery[redis]基本的なCeleryアプリケーションの構成
このコードはCeleryアプリケーションを生成するファクトリ関数です。環境変数からRedisのURLを取得し、ブローカーとバックエンドに設定します。メール、画像処理、データ処理などタスク種別ごとにキューを振り分けるルーティング定義が特徴です。
# app/celery_worker.py
from celery import Celery
import os
def make_celery(app_name=__name__):
"""
Celeryアプリケーションを作成するファクトリ関数
Args:
app_name: アプリケーション名
Returns:
Celery: Celeryアプリケーションインスタンス
"""
# Redisをブローカーとして使用
broker_url = os.environ.get('CELERY_BROKER_URL', 'redis://localhost:6379/0')
result_backend = os.environ.get('CELERY_RESULT_BACKEND', 'redis://localhost:6379/0')
celery_app = Celery(
app_name,
broker=broker_url,
backend=result_backend,
include=[
'app.tasks.email_tasks',
'app.tasks.image_tasks',
'app.tasks.data_processing_tasks'
]
)
# 設定
celery_app.conf.update(
# タスクの設定
task_serializer='json',
accept_content=['json'],
result_serializer='json',
timezone='Asia/Tokyo',
enable_utc=True,
# ワーカーの設定
worker_prefetch_multiplier=1, # 同時に処理するタスク数
worker_max_tasks_per_child=100, # 子プロセスの最大タスク数
worker_hijack_root_logger=False,
# タスクキューの設定
task_routes={
'app.tasks.email_tasks.*': {'queue': 'emails'},
'app.tasks.image_tasks.*': {'queue': 'images'},
'app.tasks.data_processing_tasks.*': {'queue': 'data'},
'app.tasks.urgent_tasks.*': {'queue': 'urgent'},
},
# タスク追跡の設定
task_track_started=True,
task_always_eager=False, # 開発時はTrueにすると同期実行
# 結果の有効期限
result_expires=3600, # 1時間
)
return celery_app
# Celeryアプリケーションのインスタンス化
celery = make_celery('flask_app')ワーカー設定では同時処理タスク数を1に制限し、子プロセスは100タスクで再起動します。結果は1時間で期限切れとなり、タスク開始状態の追跡が有効化されています。シリアライザはJSON、タイムゾーンは日本時間に設定され、最後にflask_appという名前のCeleryインスタンスを生成しています。
Flaskアプリケーションファクトリで、CeleryとFlaskの統合処理が中心です。create_app関数内で、まずCeleryインスタンスにFlaskの設定情報を取り込みます。
次にContextTaskクラスを定義し、Celeryタスク実行時にFlaskアプリケーションコンテキストが自動的に有効になるよう拡張しています。これにより、タスク内部でFlaskの設定値やデータベース接続などにアクセス可能になります。
# app/__init__.py
from flask import Flask
from config import Config
from app.celery_worker import celery
def create_app(config_class=Config):
app = Flask(__name__)
app.config.from_object(config_class)
# Celeryの設定をFlaskアプリから読み込む
celery.conf.update(app.config)
# コンテキストを設定（Flaskアプリケーションコンテキスト）
class ContextTask(celery.Task):
def __call__(self, *args, **kwargs):
with app.app_context():
return self.run(*args, **kwargs)
celery.Task = ContextTask
# ブループリントの登録
from app.routes.main import bp as main_bp
from app.routes.tasks import bp as tasks_bp
app.register_blueprint(main_bp)
app.register_blueprint(tasks_bp, url_prefix='/tasks')
return app最後にmainとtasksのブループリントを登録し、/tasksプレフィックスでタスク関連のエンドポイントを提供します。この実装により、CeleryタスクがFlaskアプリケーションのコンテキスト内で適切に動作するようになります。
Redisの設定と起動
このDocker Compose設定はFlask+Celeryの開発環境を定義しています。Redisをメッセージブローカーとバックエンドに使用し、パスワード認証を設定、永続化のためデータボリュームをマウントしています。
Celeryワーカーはapp.celery_workerのceleryインスタンスを起動し、Redisのヘルスチェック完了後に開始します。ソースコードはホストにマウントされ、変更が即時反映されます。Celery Beatは定期タスクのスケジューラー、Flowerは5555番ポートでWebベースの監視機能を提供します。
# docker-compose.yml（開発環境用）
version: '3.8'
services:
redis:
image: redis:7-alpine
container_name: flask_redis
ports:
- "6379:6379"
volumes:
- redis_data:/data
command: redis-server --appendonly yes --requirepass ${REDIS_PASSWORD:-password}
healthcheck:
test: ["CMD", "redis-cli", "ping"]
interval: 10s
timeout: 5s
retries: 3
celery_worker:
build: .
container_name: celery_worker
command: celery -A app.celery_worker.celery worker --loglevel=info
environment:
- FLASK_APP=run.py
- FLASK_ENV=development
- CELERY_BROKER_URL=redis://:${REDIS_PASSWORD:-password}@redis:6379/0
- CELERY_RESULT_BACKEND=redis://:${REDIS_PASSWORD:-password}@redis:6379/0
depends_on:
redis:
condition: service_healthy
volumes:
- .:/app
restart: unless-stopped
celery_beat:
build: .
container_name: celery_beat
command: celery -A app.celery_worker.celery beat --loglevel=info
environment:
- FLASK_APP=run.py
- FLASK_ENV=development
- CELERY_BROKER_URL=redis://:${REDIS_PASSWORD:-password}@redis:6379/0
- CELERY_RESULT_BACKEND=redis://:${REDIS_PASSWORD:-password}@redis:6379/0
depends_on:
redis:
condition: service_healthy
volumes:
- .:/app
restart: unless-stopped
flower:
image: mher/flower:1.0
container_name: flower_monitor
command: celery flower --broker=redis://:${REDIS_PASSWORD:-password}@redis:6379/0 --port=5555
ports:
- "5555:5555"
depends_on:
- redis
restart: unless-stopped
volumes:
redis_data:環境変数でブローカーURLを指定し、Redisパスワードはデフォルト値を設定可能です。開発効率を重視した構成となっています。
次にDockerfileはCeleryワーカー実行環境を構築します。軽量なPython 3.11-slimイメージをベースに、コンパイル依存関係としてgccとg++をインストール後、クリーンアップしています。
# Dockerfile
FROM python:3.11-slim
WORKDIR /app
# システム依存関係のインストール
RUN apt-get update && apt-get install -y \
gcc \
g++ \
&& rm -rf /var/lib/apt/lists/*
# Python依存関係のインストール
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# アプリケーションコードのコピー
COPY . .
# 非rootユーザーの作成
RUN useradd -m -u 1000 celeryuser && chown -R celeryuser:celeryuser /app
USER celeryuser
CMD ["celery", "-A", "app.celery_worker.celery", "worker", "--loglevel=info"]requirements.txtから依存パッケージをインストールし、アプリケーションコードをコピーします。セキュリティ向上のため、UID 1000の非rootユーザーを作成し、所有者を変更した上で実行ユーザーを切り替えています。
デフォルトコマンドはCeleryワーカーの起動で、app.celery_workerのceleryインスタンスを指定しています。マルチステージビルドではないシンプルな構成で、開発用途に適しています。
バックグラウンドタスク
基本的なタスクの作成
long_running_taskは進捗状況をupdate_stateで逐次更新しながら実行する長時間処理です。periodic_task_exampleは定期実行用の雛形、retryable_taskは最大3回・30秒間隔のリトライ機構を実装し、ランダムエラーで再試行をシミュレートします。
# app/tasks/base_tasks.py
from app.celery_worker import celery
import time
from datetime import datetime, timedelta
import logging
# ロガーの設定
logger = logging.getLogger(__name__)
@celery.task(bind=True)
def long_running_task(self, duration=10):
"""
長時間実行されるタスクの例
Args:
duration: 実行時間（秒）
Returns:
dict: タスク実行結果
"""
task_id = self.request.id
logger.info(f"タスク開始: {task_id}, 実行時間: {duration}秒")
try:
# タスクの進捗状況を更新
self.update_state(
state='PROGRESS',
meta={
'current': 0,
'total': 100,
'status': '処理を開始しました'
}
)
# シミュレートされた長時間処理
for i in range(duration):
time.sleep(1)
# 進捗状況を更新
progress = int((i + 1) / duration * 100)
self.update_state(
state='PROGRESS',
meta={
'current': progress,
'total': 100,
'status': f'処理中... {i+1}/{duration}秒'
}
)
logger.debug(f"タスク {task_id}: 進捗 {progress}%")
# タスク完了
result = {
'task_id': task_id,
'status': '完了',
'execution_time': duration,
'completed_at': datetime.now().isoformat(),
'message': 'タスクが正常に完了しました'
}
logger.info(f"タスク完了: {task_id}")
return result
except Exception as e:
logger.error(f"タスクエラー {task_id}: {e}")
# エラー情報を返す
return {
'task_id': task_id,
'status': '失敗',
'error': str(e),
'failed_at': datetime.now().isoformat()
}
@celery.task
def periodic_task_example():
"""
定期的に実行されるタスクの例
Returns:
dict: 実行結果
"""
try:
current_time = datetime.now().isoformat()
logger.info(f"定期タスク実行: {current_time}")
# ここに定期的に実行したい処理を記述
# 例: データベースのクリーニング、ログのローテーションなど
return {
'status': 'success',
'executed_at': current_time,
'message': '定期タスクが実行されました'
}
except Exception as e:
logger.error(f"定期タスクエラー: {e}")
return {
'status': 'error',
'error': str(e)
}
@celery.task(bind=True, max_retries=3, default_retry_delay=30)
def retryable_task(self, data):
"""
リトライ可能なタスクの例
Args:
data: 処理するデータ
Returns:
dict: 処理結果
"""
task_id = self.request.id
logger.info(f"リトライ可能タスク開始: {task_id}")
try:
# ここで何らかの処理を実行
# 例: 外部APIの呼び出し、ファイルの処理など
# エラーをシミュレート（50%の確率でエラー）
import random
if random.random() < 0.5:
raise Exception("ランダムなエラーが発生しました")
# 成功時の処理
return {
'task_id': task_id,
'status': 'success',
'data_processed': len(str(data)),
'message': '処理が成功しました'
}
except Exception as exc:
logger.warning(f"タスク {task_id} 失敗、リトライします: {exc}")
# リトライを実行
try:
self.retry(exc=exc, countdown=self.default_retry_delay)
except Exception as retry_exc:
logger.error(f"リトライ失敗: {retry_exc}")
return {
'task_id': task_id,
'status': 'failed',
'error': str(exc),
'retries_exhausted': True
}
@celery.task
def task_with_priority(priority='medium', data=None):
"""
優先度付きタスクの例
Args:
priority: 優先度 ('low', 'medium', 'high', 'urgent')
data: 処理するデータ
Returns:
dict: 処理結果
"""
logger.info(f"優先度付きタスク実行: 優先度={priority}")
# 優先度に応じた処理時間
priority_delays = {
'low': 5,
'medium': 3,
'high': 1,
'urgent': 0.5
}
delay = priority_delays.get(priority, 3)
time.sleep(delay)
return {
'priority': priority,
'processing_time': delay,
'data_received': data,
'completed_at': datetime.now().isoformat()
}
@celery.task
def chain_tasks():
"""
タスクチェーンの例（複数のタスクを順次実行）
"""
# このタスク内で複数の処理を実行
results = []
# ステップ1
time.sleep(1)
results.append({'step': 1, 'result': 'ステップ1完了'})
# ステップ2
time.sleep(2)
results.append({'step': 2, 'result': 'ステップ2完了'})
# ステップ3
time.sleep(1)
results.append({'step': 3, 'result': 'ステップ3完了'})
return {
'chain_completed': True,
'steps': len(results),
'results': results,
'total_time': 4 # 1+2+1秒
}task_with_priorityは優先度に応じた処理遅延の切替、chain_tasksは単一タスク内での段階的処理を例示しています。bind=Trueでタスクインスタンスへのアクセス、loggerで実行状況を記録する共通設計です。
タスクの監視と管理
monitor_system_statusはpsutilでCPU、メモリ、ディスク使用率を取得し、閾値超過時にアラートを生成します。cleanup_old_tasksはRedis上の古いタスク結果を削除し、check_task_healthはRedis接続、ワーカー、キュー長を包括的に診断してシステム健全性を評価します。
# app/tasks/monitor_tasks.py
from app.celery_worker import celery
from celery.result import AsyncResult
import time
from datetime import datetime, timedelta
import logging
import psutil
import redis
logger = logging.getLogger(__name__)
def get_redis_connection():
"""Redis接続を取得"""
import os
from urllib.parse import urlparse
redis_url = os.environ.get('CELERY_BROKER_URL', 'redis://localhost:6379/0')
parsed = urlparse(redis_url)
return redis.Redis(
host=parsed.hostname or 'localhost',
port=parsed.port or 6379,
password=parsed.password,
decode_responses=True
)
@celery.task
def monitor_system_status():
"""
システム状態を監視するタスク
Returns:
dict: システム状態情報
"""
try:
# システムリソース情報
cpu_percent = psutil.cpu_percent(interval=1)
memory_info = psutil.virtual_memory()
disk_usage = psutil.disk_usage('/')
# プロセス情報
current_process = psutil.Process()
# Redis接続の確認
redis_conn = None
redis_status = 'unknown'
try:
redis_conn = get_redis_connection()
redis_conn.ping()
redis_status = 'healthy'
except Exception as e:
redis_status = f'error: {str(e)}'
# Celeryワーカーの情報（簡易的なチェック）
worker_count = 0
if redis_conn:
try:
# Redisからワーカー情報を取得
workers = redis_conn.smembers('celery:workers')
worker_count = len(workers)
except:
worker_count = 0
status_info = {
'timestamp': datetime.now().isoformat(),
'system': {
'cpu_percent': cpu_percent,
'memory': {
'total': memory_info.total,
'available': memory_info.available,
'percent': memory_info.percent,
'used': memory_info.used
},
'disk': {
'total': disk_usage.total,
'used': disk_usage.used,
'free': disk_usage.free,
'percent': disk_usage.percent
}
},
'process': {
'pid': current_process.pid,
'cpu_percent': current_process.cpu_percent(),
'memory_percent': current_process.memory_percent(),
'create_time': datetime.fromtimestamp(current_process.create_time()).isoformat()
},
'celery': {
'worker_count': worker_count,
'active_queues': list(celery.conf.task_routes.keys()) if hasattr(celery.conf, 'task_routes') else []
},
'redis': {
'status': redis_status
}
}
logger.info(f"システム状態監視: CPU={cpu_percent}%, Memory={memory_info.percent}%")
# アラート条件のチェック
alerts = []
if cpu_percent > 80:
alerts.append(f"CPU使用率が高い: {cpu_percent}%")
if memory_info.percent > 85:
alerts.append(f"メモリ使用率が高い: {memory_info.percent}%")
if disk_usage.percent > 90:
alerts.append(f"ディスク使用率が高い: {disk_usage.percent}%")
if alerts:
status_info['alerts'] = alerts
logger.warning(f"システムアラート: {', '.join(alerts)}")
return status_info
except Exception as e:
logger.error(f"システム監視エラー: {e}")
return {
'timestamp': datetime.now().isoformat(),
'status': 'error',
'error': str(e)
}
@celery.task
def cleanup_old_tasks(days_old=7):
"""
古いタスク結果をクリーンアップ
Args:
days_old: 削除対象の日数
Returns:
dict: クリーンアップ結果
"""
try:
redis_conn = get_redis_connection()
# 現在の日時
cutoff_time = datetime.now() - timedelta(days=days_old)
cutoff_timestamp = cutoff_time.timestamp()
deleted_count = 0
# タスク結果のキーパターン
result_patterns = [
'celery-task-meta-*',
'celery:task:*',
]
for pattern in result_patterns:
keys = redis_conn.keys(pattern)
for key in keys:
try:
# タスク結果を取得
task_data = redis_conn.get(key)
if task_data:
# 簡易的なタイムスタンプチェック
# 実際の実装ではタスクデータから日時を解析する必要がある
redis_conn.delete(key)
deleted_count += 1
except Exception as e:
logger.warning(f"キー削除エラー {key}: {e}")
logger.info(f"古いタスクをクリーンアップ: {deleted_count}件削除")
return {
'status': 'success',
'deleted_count': deleted_count,
'cutoff_date': cutoff_time.isoformat(),
'cleanup_time': datetime.now().isoformat()
}
except Exception as e:
logger.error(f"タスククリーンアップエラー: {e}")
return {
'status': 'error',
'error': str(e)
}
@celery.task
def check_task_health():
"""
タスクシステムの健全性チェック
Returns:
dict: 健全性チェック結果
"""
health_checks = []
try:
# 1. Redis接続チェック
redis_conn = get_redis_connection()
redis_conn.ping()
health_checks.append({'component': 'redis', 'status': 'healthy'})
except Exception as e:
health_checks.append({'component': 'redis', 'status': 'unhealthy', 'error': str(e)})
try:
# 2. Celeryワーカーチェック
workers = redis_conn.smembers('celery:workers')
if workers:
health_checks.append({
'component': 'celery_workers',
'status': 'healthy',
'count': len(workers)
})
else:
health_checks.append({
'component': 'celery_workers',
'status': 'warning',
'message': 'ワーカーが実行されていません'
})
except Exception as e:
health_checks.append({
'component': 'celery_workers',
'status': 'unhealthy',
'error': str(e)
})
# 3. キューのチェック
try:
queues = ['default', 'emails', 'images', 'data', 'urgent']
queue_lengths = {}
for queue in queues:
length = redis_conn.llen(f'celery:{queue}')
queue_lengths[queue] = length
health_checks.append({
'component': 'task_queues',
'status': 'healthy',
'queue_lengths': queue_lengths
})
except Exception as e:
health_checks.append({
'component': 'task_queues',
'status': 'unhealthy',
'error': str(e)
})
# 総合評価
unhealthy_checks = [h for h in health_checks if h['status'] in ['unhealthy', 'warning']]
overall_status = 'healthy'
if any(h['status'] == 'unhealthy' for h in health_checks):
overall_status = 'unhealthy'
elif any(h['status'] == 'warning' for h in health_checks):
overall_status = 'warning'
result = {
'timestamp': datetime.now().isoformat(),
'overall_status': overall_status,
'health_checks': health_checks,
'unhealthy_count': len([h for h in health_checks if h['status'] == 'unhealthy']),
'warning_count': len([h for h in health_checks if h['status'] == 'warning'])
}
if overall_status != 'healthy':
logger.warning(f"タスクシステム健全性チェック: 状態={overall_status}")
return result
class TaskManager:
"""タスク管理ユーティリティクラス"""
@staticmethod
def get_task_status(task_id):
"""
タスクの状態を取得
Args:
task_id: タスクID
Returns:
dict: タスク状態情報
"""
try:
task_result = AsyncResult(task_id, app=celery)
status_info = {
'task_id': task_id,
'status': task_result.status,
'ready': task_result.ready(),
'successful': task_result.successful(),
'failed': task_result.failed()
}
if task_result.ready():
if task_result.successful():
status_info['result'] = task_result.result
else:
status_info['error'] = str(task_result.result)
# 追加情報（状態がPROGRESSの場合）
if task_result.status == 'PROGRESS':
status_info['info'] = task_result.info
return status_info
except Exception as e:
return {
'task_id': task_id,
'status': 'ERROR',
'error': f'タスク状態取得エラー: {str(e)}'
}
@staticmethod
def revoke_task(task_id, terminate=False):
"""
タスクをキャンセル
Args:
task_id: タスクID
terminate: 強制終了するか
Returns:
dict: キャンセル結果
"""
try:
celery.control.revoke(task_id, terminate=terminate)
return {
'success': True,
'task_id': task_id,
'message': 'タスクをキャンセルしました'
}
except Exception as e:
return {
'success': False,
'task_id': task_id,
'error': str(e)
}
@staticmethod
def get_queue_stats():
"""
キューの統計情報を取得
Returns:
dict: キューの統計情報
"""
try:
redis_conn = get_redis_connection()
queues = ['default', 'emails', 'images', 'data', 'urgent']
stats = {}
for queue in queues:
length = redis_conn.llen(f'celery:{queue}')
stats[queue] = {
'length': length,
'name': queue
}
# アクティブなワーカー数
workers = redis_conn.smembers('celery:workers')
return {
'queues': stats,
'worker_count': len(workers),
'workers': list(workers),
'timestamp': datetime.now().isoformat()
}
except Exception as e:
return {
'error': str(e),
'timestamp': datetime.now().isoformat()
}TaskManagerクラスはタスク状態取得、キャンセル、キュー統計を静的メソッドで提供するユーティリティです。監視タスク自体がCeleryで定期実行されることを前提とした設計になっています。
メール送信機能
メール設定とユーティリティ
Flask-Mailを使用したメール送信モジュールを作ります。init_mailでFlaskアプリケーションにメール設定を初期化し、環境変数からSMTP認証情報を読み込みます。
EmailTemplateクラスはwelcome、password_reset、notificationの3種類のHTMLテンプレートを内包し、Jinja2で動的レンダリングします。件名と本文をテンプレート変数から生成し、デフォルトでアプリ名と現在年を注入します。
<pre><code class="language-python"># app/email/__init__.py
from flask_mail import Mail, Message
import os
from threading import Thread
from flask import current_app, render_template_string
import logging
logger = logging.getLogger(__name__)
# Flask-Mailの初期化
mail = Mail()
def init_mail(app):
"""メールを初期化"""
mail.init_app(app)
# 設定のデフォルト値
app.config.setdefault('MAIL_SERVER', 'smtp.gmail.com')
app.config.setdefault('MAIL_PORT', 587)
app.config.setdefault('MAIL_USE_TLS', True)
app.config.setdefault('MAIL_USE_SSL', False)
app.config.setdefault('MAIL_USERNAME', os.environ.get('MAIL_USERNAME'))
app.config.setdefault('MAIL_PASSWORD', os.environ.get('MAIL_PASSWORD'))
app.config.setdefault('MAIL_DEFAULT_SENDER', os.environ.get('MAIL_DEFAULT_SENDER'))
app.config.setdefault('MAIL_MAX_EMAILS', None)
app.config.setdefault('MAIL_ASCII_ATTACHMENTS', False)
return mail
class EmailTemplate:
"""メールテンプレートクラス"""
@staticmethod
def get_template(template_name, **kwargs):
"""
メールテンプレートを取得
Args:
template_name: テンプレート名
**kwargs: テンプレート変数
Returns:
tuple: (件名, HTML本文, プレーンテキスト本文)
"""
templates = {
'welcome': {
'subject': '{{app_name}}へようこそ！',
'html': '''
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<style>
body { font-family: Arial, sans-serif; line-height: 1.6; color: #333; }
.container { max-width: 600px; margin: 0 auto; padding: 20px; }
.header { background-color: #4CAF50; color: white; padding: 20px; text-align: center; }
.content { padding: 30px; background-color: #f9f9f9; }
.footer { text-align: center; padding: 20px; color: #777; font-size: 12px; }
.button { display: inline-block; padding: 12px 24px; background-color: #4CAF50;
color: white; text-decoration: none; border-radius: 4px; margin: 20px 0; }
</style>
</head>
<body>
<div class="container">
<div class="header">
<h1>{{app_name}}</h1>
</div>
<div class="content">
<h2>{{user_name}}さん、ようこそ！</h2>
<p>ご登録ありがとうございます。アカウントの登録が完了しました。</p>
<p>以下のボタンをクリックして、アカウントを有効化してください：</p>
<p style="text-align: center;">
<a href="{{activation_link}}" class="button">アカウントを有効化</a>
</p>
<p>このリンクは{{expiry_hours}}時間有効です。</p>
<p>ご不明な点がございましたら、お気軽にお問い合わせください。</p>
</div>
<div class="footer">
<p>このメールは{{app_name}}から自動送信されています。</p>
<p>© {{current_year}} {{app_name}}. All rights reserved.</p>
</div>
</div>
</body>
</html>
''',
'text': '''
{{app_name}}へようこそ！
{{user_name}}さん、ご登録ありがとうございます。アカウントの登録が完了しました。
以下のリンクをクリックして、アカウントを有効化してください：
{{activation_link}}
このリンクは{{expiry_hours}}時間有効です。
ご不明な点がございましたら、お気軽にお問い合わせください。
--
このメールは{{app_name}}から自動送信されています。
© {{current_year}} {{app_name}}. All rights reserved.
'''
},
'password_reset': {
'subject': 'パスワードリセットのご案内',
'html': '''
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<style>
body { font-family: Arial, sans-serif; line-height: 1.6; color: #333; }
.container { max-width: 600px; margin: 0 auto; padding: 20px; }
.header { background-color: #f44336; color: white; padding: 20px; text-align: center; }
.content { padding: 30px; background-color: #f9f9f9; }
.footer { text-align: center; padding: 20px; color: #777; font-size: 12px; }
.button { display: inline-block; padding: 12px 24px; background-color: #f44336;
color: white; text-decoration: none; border-radius: 4px; margin: 20px 0; }
.warning { background-color: #fff3cd; border: 1px solid #ffeaa7;
padding: 15px; border-radius: 4px; margin: 20px 0; }
</style>
</head>
<body>
<div class="container">
<div class="header">
<h1>パスワードリセット</h1>
</div>
<div class="content">
<h2>パスワードのリセットをリクエストされました</h2>
<p>以下のボタンをクリックして、新しいパスワードを設定してください：</p>
<p style="text-align: center;">
<a href="{{reset_link}}" class="button">パスワードをリセット</a>
</p>
<div class="warning">
<p><strong>注意:</strong> このリンクは{{expiry_minutes}}分間のみ有効です。</p>
<p>パスワードのリセットをリクエストしていない場合は、このメールを無視してください。</p>
</div>
</div>
<div class="footer">
<p>このメールは{{app_name}}から自動送信されています。</p>
<p>© {{current_year}} {{app_name}}. All rights reserved.</p>
</div>
</div>
</body>
</html>
''',
'text': '''
パスワードリセット
パスワードのリセットをリクエストされました。
以下のリンクをクリックして、新しいパスワードを設定してください：
{{reset_link}}
注意: このリンクは{{expiry_minutes}}分間のみ有効です。
パスワードのリセットをリクエストしていない場合は、このメールを無視してください。
--
このメールは{{app_name}}から自動送信されています。
© {{current_year}} {{app_name}}. All rights reserved.
'''
},
'notification': {
'subject': '{{subject}}',
'html': '''
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<style>
body { font-family: Arial, sans-serif; line-height: 1.6; color: #333; }
.container { max-width: 600px; margin: 0 auto; padding: 20px; }
.header { background-color: #2196F3; color: white; padding: 20px; text-align: center; }
.content { padding: 30px; background-color: #f9f9f9; }
.footer { text-align: center; padding: 20px; color: #777; font-size: 12px; }
</style>
</head>
<body>
<div class="container">
<div class="header">
<h1>通知</h1>
</div>
<div class="content">
<h2>{{title}}</h2>
{{content|safe}}
{% if action_link %}
<p style="text-align: center; margin-top: 30px;">
<a href="{{action_link}}" style="display: inline-block; padding: 12px 24px;
background-color: #2196F3; color: white; text-decoration: none;
border-radius: 4px;">詳細を見る</a>
</p>
{% endif %}
</div>
<div class="footer">
<p>このメールは{{app_name}}から自動送信されています。</p>
<p>© {{current_year}} {{app_name}}. All rights reserved.</p>
</div>
</div>
</body>
</html>
''',
'text': '''
通知: {{title}}
{{content}}
{% if action_link %}
詳細: {{action_link}}
{% endif %}
--
このメールは{{app_name}}から自動送信されています。
© {{current_year}} {{app_name}}. All rights reserved.
'''
}
}
if template_name not in templates:
raise ValueError(f"テンプレート '{template_name}' は存在しません")
template = templates[template_name]
# デフォルト変数を追加
from datetime import datetime
default_vars = {
'current_year': datetime.now().year,
'app_name': current_app.config.get('APP_NAME', 'Flaskアプリケーション')
}
# ユーザー指定の変数でデフォルトを上書き
all_vars = {**default_vars, **kwargs}
# テンプレートをレンダリング
subject = render_template_string(template['subject'], **all_vars)
html_body = render_template_string(template['html'], **all_vars)
text_body = render_template_string(template['text'], **all_vars)
return subject, html_body, text_body
def send_email_sync(to, subject, html_body, text_body=None, cc=None, bcc=None,
attachments=None, reply_to=None):
"""
同期的にメールを送信
Args:
to: 送信先メールアドレス（文字列またはリスト）
subject: 件名
html_body: HTML本文
text_body: プレーンテキスト本文（オプション）
cc: CCメールアドレス
bcc: BCCメールアドレス
attachments: 添付ファイルのリスト
reply_to: 返信先メールアドレス
Returns:
bool: 送信成功かどうか
"""
try:
# メッセージオブジェクトを作成
msg = Message(
subject=subject,
recipients=[to] if isinstance(to, str) else to,
html=html_body,
body=text_body or html_body, # text_bodyがなければHTMLからテキストを抽出（簡易的）
cc=cc,
bcc=bcc,
reply_to=reply_to
)
# 添付ファイルを追加
if attachments:
for attachment in attachments:
if isinstance(attachment, dict):
msg.attach(
filename=attachment['filename'],
content_type=attachment.get('content_type'),
data=attachment['data']
)
elif isinstance(attachment, tuple) and len(attachment) == 2:
msg.attach(*attachment)
# メールを送信
mail.send(msg)
logger.info(f"メール送信成功: {to} - {subject}")
return True
except Exception as e:
logger.error(f"メール送信エラー: {e}")
return False
def send_email_async(to, subject, html_body, text_body=None, cc=None, bcc=None,
attachments=None, reply_to=None):
"""
非同期でメールを送信（スレッドを使用）
Note: 本番環境ではCeleryタスクを使用することを推奨
"""
def send_thread():
with current_app.app_context():
send_email_sync(to, subject, html_body, text_body, cc, bcc, attachments, reply_to)
thread = Thread(target=send_thread)
thread.start()
return thread
def send_template_email(to, template_name, **template_vars):
"""
テンプレートを使用してメールを送信
Args:
to: 送信先メールアドレス
template_name: テンプレート名
**template_vars: テンプレート変数
Returns:
bool: 送信成功かどうか
"""
try:
# テンプレートを取得
subject, html_body, text_body = EmailTemplate.get_template(
template_name, **template_vars
)
# メールを送信
return send_email_sync(to, subject, html_body, text_body)
except Exception as e:
logger.error(f"テンプレートメール送信エラー: {e}")
return False</code></pre>送信関数は同期的なsend_email_syncと、Threadによる簡易非同期のsend_email_asyncを提供。添付ファイル、CC/BCCに対応し、send_template_emailでテンプレート名を指定するだけで送信可能です。エラーはログに記録されます。
Celeryを使ったメール送信タスク
Celeryを使用した非同期メール送信タスク群です。send_email_taskは汎用メール送信の基盤タスクで、最大3回のリトライ機構と進捗追跡を備えています。send_welcome_emailとsend_password_reset_emailは特定用途向けのタスクで、FlaskアプリケーションコンテキストからベースURLや有効期限を取得し、テンプレートエンジンでレンダリングしたメールを送信します。
# app/tasks/email_tasks.py
from app.celery_worker import celery
from app.email import send_email_sync, send_template_email, EmailTemplate
from flask import current_app
import logging
from datetime import datetime, timedelta
import time
logger = logging.getLogger(__name__)
@celery.task(bind=True, max_retries=3, default_retry_delay=60)
def send_email_task(self, to, subject, html_body, text_body=None,
cc=None, bcc=None, attachments=None, reply_to=None):
"""
メール送信タスク（Celeryを使用）
Args:
to: 送信先メールアドレス
subject: 件名
html_body: HTML本文
text_body: プレーンテキスト本文
cc: CCメールアドレス
bcc: BCCメールアドレス
attachments: 添付ファイル
reply_to: 返信先メールアドレス
Returns:
dict: 送信結果
"""
task_id = self.request.id
logger.info(f"メール送信タスク開始: {task_id}")
try:
# タスクの進捗状況を更新
self.update_state(
state='PROGRESS',
meta={
'current': 0,
'total': 100,
'status': 'メール送信を開始します'
}
)
# メール送信
success = send_email_sync(
to=to,
subject=subject,
html_body=html_body,
text_body=text_body,
cc=cc,
bcc=bcc,
attachments=attachments,
reply_to=reply_to
)
if success:
# 送信成功
self.update_state(
state='PROGRESS',
meta={
'current': 100,
'total': 100,
'status': 'メール送信完了'
}
)
result = {
'task_id': task_id,
'status': 'success',
'to': to,
'subject': subject,
'sent_at': datetime.now().isoformat(),
'message': 'メールが正常に送信されました'
}
logger.info(f"メール送信成功: {task_id} - {to}")
return result
else:
# 送信失敗
error_msg = 'メール送信に失敗しました'
logger.error(f"メール送信失敗: {task_id} - {to}")
# リトライを実行
self.retry(exc=Exception(error_msg), countdown=self.default_retry_delay)
except Exception as e:
logger.error(f"メール送信タスクエラー {task_id}: {e}")
try:
# リトライを実行
self.retry(exc=e, countdown=self.default_retry_delay)
except Exception as retry_exc:
# リトライ回数超過
return {
'task_id': task_id,
'status': 'failed',
'to': to,
'subject': subject,
'error': str(e),
'retries_exhausted': True,
'failed_at': datetime.now().isoformat()
}
@celery.task(bind=True)
def send_welcome_email(self, user_email, user_name, activation_token):
"""
ウェルカムメール送信タスク
Args:
user_email: ユーザーのメールアドレス
user_name: ユーザー名
activation_token: アクティベーショントークン
Returns:
dict: 送信結果
"""
task_id = self.request.id
logger.info(f"ウェルカムメール送信開始: {task_id} - {user_email}")
try:
# アプリケーションコンテキストの取得
app = current_app._get_current_object()
# アクティベーションリンクの生成
activation_link = f"{app.config.get('BASE_URL', 'http://localhost:5000')}/activate/{activation_token}"
# テンプレート変数
template_vars = {
'user_name': user_name,
'activation_link': activation_link,
'expiry_hours': app.config.get('ACTIVATION_EXPIRY_HOURS', 24)
}
# メール送信
success = send_template_email(
to=user_email,
template_name='welcome',
**template_vars
)
if success:
result = {
'task_id': task_id,
'status': 'success',
'email_type': 'welcome',
'user_email': user_email,
'user_name': user_name,
'sent_at': datetime.now().isoformat()
}
logger.info(f"ウェルカムメール送信成功: {task_id}")
return result
else:
error_msg = 'ウェルカムメールの送信に失敗しました'
logger.error(f"ウェルカムメール送信失敗: {task_id}")
raise Exception(error_msg)
except Exception as e:
logger.error(f"ウェルカムメールタスクエラー {task_id}: {e}")
return {
'task_id': task_id,
'status': 'failed',
'email_type': 'welcome',
'user_email': user_email,
'error': str(e),
'failed_at': datetime.now().isoformat()
}
@celery.task(bind=True)
def send_password_reset_email(self, user_email, reset_token):
"""
パスワードリセットメール送信タスク
Args:
user_email: ユーザーのメールアドレス
reset_token: リセットトークン
Returns:
dict: 送信結果
"""
task_id = self.request.id
logger.info(f"パスワードリセットメール送信開始: {task_id} - {user_email}")
try:
app = current_app._get_current_object()
# リセットリンクの生成
reset_link = f"{app.config.get('BASE_URL', 'http://localhost:5000')}/reset-password/{reset_token}"
# テンプレート変数
template_vars = {
'reset_link': reset_link,
'expiry_minutes': app.config.get('PASSWORD_RESET_EXPIRY_MINUTES', 30)
}
# メール送信
success = send_template_email(
to=user_email,
template_name='password_reset',
**template_vars
)
if success:
result = {
'task_id': task_id,
'status': 'success',
'email_type': 'password_reset',
'user_email': user_email,
'sent_at': datetime.now().isoformat()
}
logger.info(f"パスワードリセットメール送信成功: {task_id}")
return result
else:
error_msg = 'パスワードリセットメールの送信に失敗しました'
logger.error(f"パスワードリセットメール送信失敗: {task_id}")
raise Exception(error_msg)
except Exception as e:
logger.error(f"パスワードリセットメールタスクエラー {task_id}: {e}")
return {
'task_id': task_id,
'status': 'failed',
'email_type': 'password_reset',
'user_email': user_email,
'error': str(e),
'failed_at': datetime.now().isoformat()
}
@celery.task
def send_bulk_emails(emails_data, template_name, template_vars_list,
delay_between_emails=1):
"""
バルクメール送信タスク
Args:
emails_data: メールデータのリスト [{'to': 'email', 'name': 'name'}, ...]
template_name: テンプレート名
template_vars_list: テンプレート変数のリスト（各メールごと）
delay_between_emails: メール間の遅延（秒）
Returns:
dict: バルク送信結果
"""
total_emails = len(emails_data)
sent_count = 0
failed_count = 0
failed_emails = []
logger.info(f"バルクメール送信開始: {total_emails}件")
for i, (email_data, template_vars) in enumerate(zip(emails_data, template_vars_list)):
try:
# 個々のメール送信タスクを実行
to = email_data['to']
user_name = email_data.get('name', '')
# テンプレート変数をマージ
merged_vars = {**template_vars, 'user_name': user_name}
# メール送信
success = send_template_email(
to=to,
template_name=template_name,
**merged_vars
)
if success:
sent_count += 1
logger.debug(f"バルクメール送信成功 ({i+1}/{total_emails}): {to}")
else:
failed_count += 1
failed_emails.append({'email': to, 'reason': '送信失敗'})
logger.warning(f"バルクメール送信失敗 ({i+1}/{total_emails}): {to}")
# 進捗状況をログに記録
if (i + 1) % 10 == 0 or i == total_emails - 1:
logger.info(f"バルクメール進捗: {i+1}/{total_emails}件完了")
# メール間の遅延
if delay_between_emails > 0 and i < total_emails - 1:
time.sleep(delay_between_emails)
except Exception as e:
failed_count += 1
failed_emails.append({'email': email_data.get('to', 'unknown'), 'reason': str(e)})
logger.error(f"バルクメール送信エラー ({i+1}/{total_emails}): {e}")
result = {
'total_emails': total_emails,
'sent_count': sent_count,
'failed_count': failed_count,
'success_rate': (sent_count / total_emails * 100) if total_emails > 0 else 0,
'failed_emails': failed_emails,
'started_at': datetime.now() - timedelta(seconds=total_emails * delay_between_emails),
'completed_at': datetime.now().isoformat()
}
logger.info(f"バルクメール送信完了: {sent_count}/{total_emails}件成功")
return result
@celery.task
def send_scheduled_email(to, subject, html_body, text_body=None,
send_at=None, timezone='Asia/Tokyo'):
"""
スケジュールされたメール送信タスク
Args:
to: 送信先メールアドレス
subject: 件名
html_body: HTML本文
text_body: プレーンテキスト本文
send_at: 送信日時（ISO形式文字列）
timezone: タイムゾーン
Returns:
dict: 送信結果
"""
from datetime import datetime
import pytz
logger.info(f"スケジュールメール送信タスク開始: {to} - {subject}")
try:
# 送信時間の解析
if send_at:
send_time = datetime.fromisoformat(send_at.replace('Z', '+00:00'))
# タイムゾーンの適用
tz = pytz.timezone(timezone)
send_time = send_time.astimezone(tz)
current_time = datetime.now(tz)
# 送信時間まで待機（簡易的な実装）
# 実際にはCelery Beatを使用することを推奨
time_to_wait = (send_time - current_time).total_seconds()
if time_to_wait > 0:
logger.info(f"スケジュールメール: {time_to_wait:.0f}秒待機")
time.sleep(time_to_wait)
elif time_to_wait < 0:
logger.warning(f"スケジュールメール: 送信時間が過去です ({time_to_wait:.0f}秒)")
# メール送信
success = send_email_sync(
to=to,
subject=subject,
html_body=html_body,
text_body=text_body
)
if success:
result = {
'status': 'success',
'email_type': 'scheduled',
'to': to,
'subject': subject,
'scheduled_for': send_at,
'actual_sent_at': datetime.now().isoformat()
}
logger.info(f"スケジュールメール送信成功: {to}")
return result
else:
error_msg = 'スケジュールメールの送信に失敗しました'
logger.error(f"スケジュールメール送信失敗: {to}")
raise Exception(error_msg)
except Exception as e:
logger.error(f"スケジュールメールタスクエラー: {e}")
return {
'status': 'failed',
'email_type': 'scheduled',
'to': to,
'error': str(e),
'failed_at': datetime.now().isoformat()
}
@celery.task
def process_email_queue():
"""
メールキューを処理するタスク
（実際の実装ではデータベースからメールキューを読み込む）
"""
logger.info("メールキュー処理開始")
try:
# ここでデータベースから未送信のメールを取得
# pending_emails = EmailQueue.query.filter_by(status='pending').limit(100).all()
pending_emails = [] # 仮のデータ
processed_count = 0
failed_count = 0
for email in pending_emails:
try:
# メール送信
success = send_email_sync(
to=email.recipient,
subject=email.subject,
html_body=email.html_body,
text_body=email.text_body
)
if success:
# 送信成功時の処理
# email.status = 'sent'
# email.sent_at = datetime.now()
# db.session.commit()
processed_count += 1
else:
# 送信失敗時の処理
# email.status = 'failed'
# email.error_count += 1
# db.session.commit()
failed_count += 1
except Exception as e:
logger.error(f"キュー内メール送信エラー: {e}")
failed_count += 1
result = {
'processed_count': processed_count,
'failed_count': failed_count,
'total': len(pending_emails),
'completed_at': datetime.now().isoformat()
}
logger.info(f"メールキュー処理完了: {processed_count}件成功, {failed_count}件失敗")
return result
except Exception as e:
logger.error(f"メールキュー処理エラー: {e}")
return {
'status': 'error',
'error': str(e),
'processed_count': 0,
'failed_count': 0
}send_bulk_emailsは大量メール送信用で、メール間遅延を設定可能です。send_scheduled_emailは指定時刻まで待機する簡易スケジューリング、process_email_queueはデータベースキュー処理の雛形を提供しています。各タスクは実行結果とエラー情報を構造化データで返します。
メール送信の設定とテスト
RedisをCeleryのブローカー兼バックエンドとして指定し、シリアライザやタイムゾーンなど非同期処理基盤の基本パラメータを定義しています。メール設定ではSMTP認証情報、アプリケーション設定ではベースURLやトークン有効期限を管理します。
# config.py
import os
from datetime import timedelta
class Config:
# 基本設定
SECRET_KEY = os.environ.get('SECRET_KEY') or 'dev-secret-key'
# データベース設定
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or \
'sqlite:///app.db'
SQLALCHEMY_TRACK_MODIFICATIONS = False
# Redis設定（Celery用）
REDIS_URL = os.environ.get('REDIS_URL', 'redis://localhost:6379/0')
# Celery設定
CELERY_BROKER_URL = os.environ.get('CELERY_BROKER_URL', 'redis://localhost:6379/0')
CELERY_RESULT_BACKEND = os.environ.get('CELERY_RESULT_BACKEND', 'redis://localhost:6379/0')
CELERY_TASK_SERIALIZER = 'json'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'Asia/Tokyo'
CELERY_ENABLE_UTC = True
# メール設定
MAIL_SERVER = os.environ.get('MAIL_SERVER', 'smtp.gmail.com')
MAIL_PORT = int(os.environ.get('MAIL_PORT', 587))
MAIL_USE_TLS = os.environ.get('MAIL_USE_TLS', 'true').lower() in ['true', 'on', '1']
MAIL_USE_SSL = os.environ.get('MAIL_USE_SSL', 'false').lower() in ['true', 'on', '1']
MAIL_USERNAME = os.environ.get('MAIL_USERNAME')
MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD')
MAIL_DEFAULT_SENDER = os.environ.get('MAIL_DEFAULT_SENDER', 'noreply@example.com')
# アプリケーション設定
APP_NAME = os.environ.get('APP_NAME', 'Flaskアプリケーション')
BASE_URL = os.environ.get('BASE_URL', 'http://localhost:5000')
# トークン有効期限
ACTIVATION_EXPIRY_HOURS = int(os.environ.get('ACTIVATION_EXPIRY_HOURS', 24))
PASSWORD_RESET_EXPIRY_MINUTES = int(os.environ.get('PASSWORD_RESET_EXPIRY_MINUTES', 30))
# セキュリティ設定
SECURITY_PASSWORD_SALT = os.environ.get('SECURITY_PASSWORD_SALT', 'dev-password-salt')
# スケジューリング設定
CELERYBEAT_SCHEDULE = {
# 毎分実行
'monitor-system-every-minute': {
'task': 'app.tasks.monitor_tasks.monitor_system_status',
'schedule': timedelta(seconds=60),
},
# 毎日午前3時に実行
'cleanup-old-tasks-daily': {
'task': 'app.tasks.monitor_tasks.cleanup_old_tasks',
'schedule': timedelta(days=1),
'args': (7,), # 7日以上前のタスクを削除
},
# 毎時間実行
'check-task-health-hourly': {
'task': 'app.tasks.monitor_tasks.check_task_health',
'schedule': timedelta(hours=1),
},
# 毎5分実行
'process-email-queue-every-5-minutes': {
'task': 'app.tasks.email_tasks.process_email_queue',
'schedule': timedelta(minutes=5),
},
# 毎日午前9時に実行（月曜日から金曜日のみ）
'send-daily-report-weekdays': {
'task': 'app.tasks.report_tasks.send_daily_report',
'schedule': timedelta(days=1),
'kwargs': {'weekdays_only': True},
},
}特に重要なのはCELERYBEAT_SCHEDULE辞書で、システム監視、タスククリーンアップ、健全性チェック、メールキュー処理などの定期実行タスクを宣言的にスケジュール定義しています。各タスクに実行間隔と引数を設定し、Celery Beatによる自動実行を可能にしています。
FlaskルートとAPIエンドポイント
CeleryタスクをHTTP APIとして公開するFlaskブループリントは、各エンドポイントはリクエストを受け付けると即座に202 Acceptedを返し、実際の処理はCeleryワーカーに委譲します。
メール送信系API（/send-test-email、/send-welcome-email）はリクエストデータを検証後、対応するCeleryタスクのdelayメソッドを呼び出して非同期実行します。/status/<task_id>と/monitor/<task_id>はTaskManagerユーティリティを通じてタスクの状態や進捗を取得します。
# app/routes/tasks.py
from flask import Blueprint, request, jsonify, current_app
from app.tasks.email_tasks import (
send_email_task,
send_welcome_email,
send_password_reset_email,
send_bulk_emails
)
from app.tasks.base_tasks import (
long_running_task,
retryable_task,
task_with_priority
)
from app.tasks.monitor_tasks import TaskManager
from celery.result import AsyncResult
import uuid
bp = Blueprint('tasks', __name__)
@bp.route('/send-test-email', methods=['POST'])
def send_test_email():
"""テストメール送信API"""
try:
data = request.get_json()
# 必須パラメータのチェック
if not data or 'to' not in data or 'subject' not in data:
return jsonify({
'error': '必須パラメータが不足しています',
'required': ['to', 'subject', 'html_body']
}), 400
# メール送信タスクをキューに追加
task = send_email_task.delay(
to=data['to'],
subject=data['subject'],
html_body=data.get('html_body', 'テストメールです
'),
text_body=data.get('text_body'),
cc=data.get('cc'),
bcc=data.get('bcc')
)
return jsonify({
'success': True,
'message': 'メール送信タスクを開始しました',
'task_id': task.id,
'status_url': f'/tasks/status/{task.id}'
}), 202
except Exception as e:
current_app.logger.error(f"テストメール送信エラー: {e}")
return jsonify({'error': str(e)}), 500
@bp.route('/send-welcome-email', methods=['POST'])
def api_send_welcome_email():
"""ウェルカムメール送信API"""
try:
data = request.get_json()
if not data or 'email' not in data or 'name' not in data:
return jsonify({'error': 'emailとnameは必須です'}), 400
# アクティベーショントークンの生成（実際の実装ではデータベースに保存）
activation_token = str(uuid.uuid4())
# ウェルカムメール送信タスクをキューに追加
task = send_welcome_email.delay(
user_email=data['email'],
user_name=data['name'],
activation_token=activation_token
)
return jsonify({
'success': True,
'message': 'ウェルカムメールを送信します',
'task_id': task.id,
'activation_token': activation_token,
'status_url': f'/tasks/status/{task.id}'
}), 202
except Exception as e:
current_app.logger.error(f"ウェルカムメール送信エラー: {e}")
return jsonify({'error': str(e)}), 500
@bp.route('/long-task', methods=['POST'])
def start_long_task():
"""長時間実行タスク開始API"""
try:
data = request.get_json() or {}
duration = data.get('duration', 10)
# 長時間実行タスクを開始
task = long_running_task.delay(duration=duration)
return jsonify({
'success': True,
'message': f'長時間実行タスクを開始しました（推定時間: {duration}秒）',
'task_id': task.id,
'status_url': f'/tasks/status/{task.id}',
'monitor_url': f'/tasks/monitor/{task.id}'
}), 202
except Exception as e:
current_app.logger.error(f"長時間タスク開始エラー: {e}")
return jsonify({'error': str(e)}), 500
@bp.route('/status/', methods=['GET'])
def get_task_status(task_id):
"""タスクステータス取得API"""
try:
status_info = TaskManager.get_task_status(task_id)
return jsonify(status_info), 200
except Exception as e:
current_app.logger.error(f"タスクステータス取得エラー: {e}")
return jsonify({'error': str(e)}), 500
@bp.route('/monitor/', methods=['GET'])
def monitor_task(task_id):
"""タスクモニタリングAPI（SSEまたは定期的ポーリング用）"""
try:
status_info = TaskManager.get_task_status(task_id)
# 進捗情報があれば追加
if status_info.get('status') == 'PROGRESS':
task_result = AsyncResult(task_id, app=current_app.celery)
if task_result.info:
status_info['progress'] = task_result.info
return jsonify(status_info), 200
except Exception as e:
current_app.logger.error(f"タスクモニタリングエラー: {e}")
return jsonify({'error': str(e)}), 500
@bp.route('/cancel/', methods=['POST'])
def cancel_task(task_id):
"""タスクキャンセルAPI"""
try:
data = request.get_json() or {}
terminate = data.get('terminate', False)
result = TaskManager.revoke_task(task_id, terminate=terminate)
if result['success']:
return jsonify(result), 200
else:
return jsonify(result), 400
except Exception as e:
current_app.logger.error(f"タスクキャンセルエラー: {e}")
return jsonify({'error': str(e)}), 500
@bp.route('/queue-stats', methods=['GET'])
def get_queue_stats():
"""キュースタッツ取得API"""
try:
stats = TaskManager.get_queue_stats()
return jsonify(stats), 200
except Exception as e:
current_app.logger.error(f"キュースタッツ取得エラー: {e}")
return jsonify({'error': str(e)}), 500
@bp.route('/health', methods=['GET'])
def task_system_health():
"""タスクシステム健全性チェックAPI"""
try:
from app.tasks.monitor_tasks import check_task_health
health_result = check_task_health.delay()
health_result.wait(timeout=10) # 10秒待機
if health_result.successful():
return jsonify(health_result.result), 200
else:
return jsonify({'error': '健全性チェックに失敗しました'}), 500
except Exception as e:
current_app.logger.error(f"健全性チェックエラー: {e}")
return jsonify({'error': str(e)}), 500
@bp.route('/test-priority-task', methods=['POST'])
def test_priority_task():
"""優先度付きタスクテストAPI"""
try:
data = request.get_json() or {}
priority = data.get('priority', 'medium')
message = data.get('message', 'テストメッセージ')
# 優先度付きタスクを実行
task = task_with_priority.delay(
priority=priority,
data={'message': message, 'request_time': str(request.date)}
)
return jsonify({
'success': True,
'message': '優先度付きタスクを開始しました',
'task_id': task.id,
'priority': priority,
'status_url': f'/tasks/status/{task.id}'
}), 202
except Exception as e:
current_app.logger.error(f"優先度タスクテストエラー: {e}")
return jsonify({'error': str(e)}), 500
@bp.route('/test-retry-task', methods=['POST'])
def test_retry_task():
"""リトライ可能タスクテストAPI"""
try:
data = request.get_json() or {}
test_data = data.get('data', 'テストデータ')
# リトライ可能タスクを実行
task = retryable_task.delay(test_data)
return jsonify({
'success': True,
'message': 'リトライ可能タスクを開始しました',
'task_id': task.id,
'max_retries': 3,
'retry_delay': 30,
'status_url': f'/tasks/status/{task.id}'
}), 202
except Exception as e:
current_app.logger.error(f"リトライタスクテストエラー: {e}")
return jsonify({'error': str(e)}), 500 キャンセル、キュースタッツ、健全性チェックなどの管理系APIも提供し、フロントエンドが非同期処理の状態を把握・制御できる完全なRESTインターフェースを実現しています。
管理画面の実装
CeleryタスクシステムのWeb管理インターフェースです。ヘッダーにはシステム健全性インジケーター、統計エリアにはキューごとのタスク数とワーカー数をリアルタイム表示します。
メール送信フォームと長時間タスク実行フォームから非同期処理を開始でき、開始されたタスクはテーブル一覧に即時追加されます。各タスクは進捗バー付きで表示され、詳細確認やキャンセル操作が可能です。
<!-- templates/admin/tasks.html -->
<!DOCTYPE html>
<html lang="ja">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>タスク管理パネル</title>
<link href="https://cdn.jsdelivr.net/npm/bootstrap@5.1.3/dist/css/bootstrap.min.css" rel="stylesheet">
<link href="https://cdn.jsdelivr.net/npm/bootstrap-icons@1.8.1/font/bootstrap-icons.css" rel="stylesheet">
<style>
body {
background-color: #f8f9fa;
font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif;
}
.dashboard-header {
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
color: white;
padding: 2rem 0;
margin-bottom: 2rem;
}
.task-card {
transition: transform 0.2s;
border: none;
box-shadow: 0 2px 4px rgba(0,0,0,0.1);
}
.task-card:hover {
transform: translateY(-5px);
box-shadow: 0 4px 8px rgba(0,0,0,0.15);
}
.task-status {
font-size: 0.85rem;
padding: 0.25rem 0.5rem;
border-radius: 4px;
}
.status-pending { background-color: #ffc107; color: #212529; }
.status-processing { background-color: #17a2b8; color: white; }
.status-success { background-color: #28a745; color: white; }
.status-failed { background-color: #dc3545; color: white; }
.status-revoked { background-color: #6c757d; color: white; }
.progress-container {
height: 8px;
background-color: #e9ecef;
border-radius: 4px;
overflow: hidden;
margin: 10px 0;
}
.progress-bar {
height: 100%;
background: linear-gradient(90deg, #4CAF50, #8BC34A);
transition: width 0.3s ease;
}
.queue-stats {
background: white;
border-radius: 8px;
padding: 20px;
margin-bottom: 20px;
box-shadow: 0 2px 4px rgba(0,0,0,0.1);
}
.stat-item {
text-align: center;
padding: 15px;
}
.stat-value {
font-size: 2rem;
font-weight: bold;
color: #667eea;
}
.stat-label {
font-size: 0.9rem;
color: #6c757d;
text-transform: uppercase;
letter-spacing: 1px;
}
.health-indicator {
width: 12px;
height: 12px;
border-radius: 50%;
display: inline-block;
margin-right: 8px;
}
.health-healthy { background-color: #28a745; }
.health-warning { background-color: #ffc107; }
.health-unhealthy { background-color: #dc3545; }
.monitor-container {
max-height: 400px;
overflow-y: auto;
}
.log-entry {
font-family: 'Courier New', monospace;
font-size: 0.85rem;
padding: 8px 12px;
border-bottom: 1px solid #dee2e6;
}
.log-time {
color: #6c757d;
font-size: 0.8rem;
}
.log-info { color: #17a2b8; }
.log-success { color: #28a745; }
.log-warning { color: #ffc107; }
.log-error { color: #dc3545; }
</style>
</head>
<body>
<div class="dashboard-header">
<div class="container">
<div class="row align-items-center">
<div class="col-md-8">
<h1 class="display-5 fw-bold">
<i class="bi bi-gear-wide-connected"></i>
タスク管理パネル
</h1>
<p class="lead mb-0">非同期タスクとメール送信システムの管理</p>
</div>
<div class="col-md-4 text-end">
<div id="healthStatus" class="d-inline-block">
<!-- 健康状態はJavaScriptで動的に更新 -->
</div>
<button class="btn btn-light" onclick="refreshAll()">
<i class="bi bi-arrow-clockwise"></i> 更新
</button>
</div>
</div>
</div>
</div>
<div class="container">
<!-- システム統計 -->
<div class="row mb-4">
<div class="col-12">
<div class="queue-stats">
<h4 class="mb-3">
<i class="bi bi-speedometer2"></i>
システム統計
</h4>
<div class="row" id="queueStats">
<!-- 統計情報はJavaScriptで動的に更新 -->
</div>
</div>
</div>
</div>
<!-- タスク制御パネル -->
<div class="row mb-4">
<div class="col-md-6">
<div class="card task-card">
<div class="card-header bg-primary text-white">
<h5 class="mb-0">
<i class="bi bi-envelope"></i>
メール送信テスト
</h5>
</div>
<div class="card-body">
<form id="emailTestForm">
<div class="mb-3">
<label for="testEmail" class="form-label">送信先メールアドレス</label>
<input type="email" class="form-control" id="testEmail"
placeholder="test@example.com" required>
</div>
<div class="mb-3">
<label for="testSubject" class="form-label">件名</label>
<input type="text" class="form-control" id="testSubject"
value="テストメール" required>
</div>
<div class="mb-3">
<label for="testMessage" class="form-label">メッセージ</label>
<textarea class="form-control" id="testMessage" rows="3"
placeholder="テストメッセージを入力してください"></textarea>
</div>
<button type="submit" class="btn btn-primary">
<i class="bi bi-send"></i>
テストメールを送信
</button>
</form>
</div>
</div>
</div>
<div class="col-md-6">
<div class="card task-card">
<div class="card-header bg-success text-white">
<h5 class="mb-0">
<i class="bi bi-clock-history"></i>
長時間タスクテスト
</h5>
</div>
<div class="card-body">
<form id="longTaskForm">
<div class="mb-3">
<label for="taskDuration" class="form-label">実行時間（秒）</label>
<input type="number" class="form-control" id="taskDuration"
value="10" min="1" max="60" required>
<div class="form-text">1〜60秒の間で指定してください</div>
</div>
<div class="mb-3">
<label class="form-label">タスクタイプ</label>
<div class="form-check">
<input class="form-check-input" type="radio" name="taskType"
id="taskTypeNormal" value="normal" checked>
<label class="form-check-label" for="taskTypeNormal">
通常タスク
</label>
</div>
<div class="form-check">
<input class="form-check-input" type="radio" name="taskType"
id="taskTypeRetry" value="retry">
<label class="form-check-label" for="taskTypeRetry">
リトライ可能タスク（50%の確率で失敗）
</label>
</div>
</div>
<button type="submit" class="btn btn-success">
<i class="bi bi-play-circle"></i>
タスクを開始
</button>
</form>
</div>
</div>
</div>
</div>
<!-- アクティブなタスク -->
<div class="row mb-4">
<div class="col-12">
<div class="card task-card">
<div class="card-header bg-info text-white d-flex justify-content-between align-items-center">
<h5 class="mb-0">
<i class="bi bi-list-task"></i>
アクティブなタスク
</h5>
<span class="badge bg-light text-dark" id="activeTaskCount">0</span>
</div>
<div class="card-body">
<div class="table-responsive">
<table class="table table-hover">
<thead>
<tr>
<th>タスクID</th>
<th>タイプ</th>
<th>ステータス</th>
<th>進捗状況</th>
<th>開始時間</th>
<th>操作</th>
</tr>
</thead>
<tbody id="activeTasksTable">
<!-- タスク一覧はJavaScriptで動的に更新 -->
</tbody>
</table>
</div>
<div class="text-center">
<button class="btn btn-outline-info btn-sm" onclick="loadActiveTasks()">
<i class="bi bi-arrow-clockwise"></i>
リストを更新
</button>
</div>
</div>
</div>
</div>
</div>
<!-- システムログ -->
<div class="row">
<div class="col-12">
<div class="card task-card">
<div class="card-header bg-dark text-white">
<h5 class="mb-0">
<i class="bi bi-terminal"></i>
システムログ
</h5>
</div>
<div class="card-body">
<div class="monitor-container">
<div id="systemLogs">
<!-- システムログはJavaScriptで動的に更新 -->
</div>
</div>
<div class="mt-3">
<button class="btn btn-outline-dark btn-sm" onclick="clearLogs()">
<i class="bi bi-trash"></i>
ログをクリア
</button>
<button class="btn btn-outline-dark btn-sm" onclick="startLogMonitoring()">
<i class="bi bi-play"></i>
モニタリング開始
</button>
<button class="btn btn-outline-dark btn-sm" onclick="stopLogMonitoring()">
<i class="bi bi-stop"></i>
モニタリング停止
</button>
</div>
</div>
</div>
</div>
</div>
</div>
<!-- モーダル：タスク詳細 -->
<div class="modal fade" id="taskDetailModal" tabindex="-1">
<div class="modal-dialog modal-lg">
<div class="modal-content">
<div class="modal-header">
<h5 class="modal-title">タスク詳細</h5>
<button type="button" class="btn-close" data-bs-dismiss="modal"></button>
</div>
<div class="modal-body">
<div id="taskDetailContent">
<!-- タスク詳細はJavaScriptで動的に更新 -->
</div>
</div>
<div class="modal-footer">
<button type="button" class="btn btn-secondary" data-bs-dismiss="modal">閉じる</button>
</div>
</div>
</div>
</div>
<script src="https://cdn.jsdelivr.net/npm/bootstrap@5.1.3/dist/js/bootstrap.bundle.min.js"></script>
<script>
// グローバル変数
let logMonitoringInterval = null;
let activeTasks = {};
const systemLogs = [];
// DOMの読み込み完了時に実行
document.addEventListener('DOMContentLoaded', function() {
// 初期データの読み込み
refreshAll();
// フォームのイベントリスナー
document.getElementById('emailTestForm').addEventListener('submit', sendTestEmail);
document.getElementById('longTaskForm').addEventListener('submit', startLongTask);
// 定期的な更新
setInterval(refreshQueueStats, 30000); // 30秒ごとにキュースタッツを更新
setInterval(loadActiveTasks, 10000); // 10秒ごとにタスク一覧を更新
// ログモニタリングを開始
startLogMonitoring();
});
// 全てを更新
function refreshAll() {
refreshQueueStats();
loadActiveTasks();
checkSystemHealth();
fetchSystemLogs();
}
// キュースタッツを更新
async function refreshQueueStats() {
try {
const response = await fetch('/tasks/queue-stats');
const data = await response.json();
if (response.ok) {
updateQueueStatsUI(data);
} else {
addSystemLog(`キュースタッツ取得エラー: ${data.error}`, 'error');
}
} catch (error) {
addSystemLog(`ネットワークエラー: ${error.message}`, 'error');
}
}
// キュースタッツUIを更新
function updateQueueStatsUI(stats) {
const container = document.getElementById('queueStats');
if (stats.error) {
container.innerHTML = `
<div class="col-12">
<div class="alert alert-danger">
<i class="bi bi-exclamation-triangle"></i>
統計情報の取得に失敗しました: ${stats.error}
</div>
</div>
`;
return;
}
const queues = stats.queues || {};
const workerCount = stats.worker_count || 0;
let html = '';
// ワーカー統計
html += `
<div class="col-md-3 col-6 stat-item">
<div class="stat-value">${workerCount}</div>
<div class="stat-label">ワーカー</div>
</div>
`;
// キューごとの統計
for (const [queueName, queueStats] of Object.entries(queues)) {
const length = queueStats.length || 0;
const bgClass = length > 10 ? 'text-danger' : length > 0 ? 'text-warning' : 'text-success';
html += `
<div class="col-md-2 col-4 stat-item">
<div class="stat-value ${bgClass}">${length}</div>
<div class="stat-label">${queueName}</div>
</div>
`;
}
// 総タスク数
const totalTasks = Object.values(queues).reduce((sum, q) => sum + (q.length || 0), 0);
html += `
<div class="col-md-3 col-6 stat-item">
<div class="stat-value">${totalTasks}</div>
<div class="stat-label">合計タスク</div>
</div>
`;
container.innerHTML = html;
}
// アクティブなタスクを読み込み
async function loadActiveTasks() {
try {
// 実際の実装ではAPIからアクティブなタスク一覧を取得
// ここではサンプルデータを使用
// サンプルタスクデータ
const sampleTasks = [
{
id: 'task-' + Date.now(),
type: 'メール送信',
status: 'PROCESSING',
progress: 65,
started: new Date(Date.now() - 30000).toLocaleTimeString()
},
{
id: 'task-' + (Date.now() - 1000),
type: '画像処理',
status: 'PENDING',
progress: 0,
started: new Date(Date.now() - 60000).toLocaleTimeString()
},
{
id: 'task-' + (Date.now() - 2000),
type: 'データ分析',
status: 'SUCCESS',
progress: 100,
started: new Date(Date.now() - 120000).toLocaleTimeString()
}
];
// 既存のタスクとマージ
sampleTasks.forEach(task => {
if (!activeTasks[task.id]) {
activeTasks[task.id] = task;
}
});
updateActiveTasksUI();
} catch (error) {
addSystemLog(`タスク一覧取得エラー: ${error.message}`, 'error');
}
}
// アクティブタスクUIを更新
function updateActiveTasksUI() {
const tableBody = document.getElementById('activeTasksTable');
const taskCount = document.getElementById('activeTaskCount');
const tasks = Object.values(activeTasks);
taskCount.textContent = tasks.length;
if (tasks.length === 0) {
tableBody.innerHTML = `
<tr>
<td colspan="6" class="text-center text-muted py-4">
<i class="bi bi-inbox"></i>
アクティブなタスクはありません
</td>
</tr>
`;
return;
}
let html = '';
tasks.forEach(task => {
const statusClass = getStatusClass(task.status);
const progressBar = task.progress > 0 ? `
<div class="progress-container">
<div class="progress-bar" style="width: ${task.progress}%"></div>
</div>
<small class="text-muted">${task.progress}%</small>
` : '<small class="text-muted">未開始</small>';
html += `
<tr>
<td>
<code class="small">${task.id.substring(0, 12)}...</code>
</td>
<td>${task.type}</td>
<td>
<span class="task-status ${statusClass}">
${getStatusText(task.status)}
</span>
</td>
<td>${progressBar}</td>
<td>${task.started}</td>
<td>
<button class="btn btn-sm btn-outline-primary"
onclick="showTaskDetail('${task.id}')">
<i class="bi bi-info-circle"></i>
</button>
<button class="btn btn-sm btn-outline-danger"
onclick="cancelTask('${task.id}')">
<i class="bi bi-x-circle"></i>
</button>
</td>
</tr>
`;
});
tableBody.innerHTML = html;
}
// ステータスクラスを取得
function getStatusClass(status) {
switch(status) {
case 'PENDING': return 'status-pending';
case 'PROCESSING': return 'status-processing';
case 'SUCCESS': return 'status-success';
case 'FAILED': return 'status-failed';
case 'REVOKED': return 'status-revoked';
default: return 'status-pending';
}
}
// ステータステキストを取得
function getStatusText(status) {
switch(status) {
case 'PENDING': return '待機中';
case 'PROCESSING': return '処理中';
case 'SUCCESS': return '成功';
case 'FAILED': return '失敗';
case 'REVOKED': return 'キャンセル済み';
default: return status;
}
}
// システム健全性をチェック
async function checkSystemHealth() {
try {
const response = await fetch('/tasks/health');
const data = await response.json();
const healthStatus = document.getElementById('healthStatus');
if (response.ok) {
const status = data.overall_status || 'unknown';
const statusClass = `health-${status}`;
const statusText = status === 'healthy' ? '正常' :
status === 'warning' ? '警告' : '異常';
healthStatus.innerHTML = `
<div class="d-flex align-items-center">
<span class="health-indicator ${statusClass}"></span>
<span>システム: ${statusText}</span>
</div>
`;
if (status !== 'healthy') {
addSystemLog(`システム健全性: ${statusText}`, 'warning');
}
} else {
healthStatus.innerHTML = `
<div class="d-flex align-items-center">
<span class="health-indicator health-unhealthy"></span>
<span>システム: チェック失敗</span>
</div>
`;
}
} catch (error) {
console.error('健全性チェックエラー:', error);
}
}
// テストメールを送信
async function sendTestEmail(event) {
event.preventDefault();
const email = document.getElementById('testEmail').value;
const subject = document.getElementById('testSubject').value;
const message = document.getElementById('testMessage').value || 'テストメールです';
try {
const response = await fetch('/tasks/send-test-email', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify({
to: email,
subject: subject,
html_body: `<p>${message}</p>`,
text_body: message
})
});
const data = await response.json();
if (response.ok) {
addSystemLog(`テストメール送信開始: ${data.task_id}`, 'success');
// アクティブタスクに追加
activeTasks[data.task_id] = {
id: data.task_id,
type: 'テストメール',
status: 'PENDING',
progress: 0,
started: new Date().toLocaleTimeString()
};
updateActiveTasksUI();
// フォームをリセット
document.getElementById('testEmail').value = '';
document.getElementById('testSubject').value = 'テストメール';
document.getElementById('testMessage').value = '';
// 成功メッセージ
alert('テストメールの送信を開始しました。タスク一覧で進捗を確認できます。');
} else {
addSystemLog(`テストメール送信失敗: ${data.error}`, 'error');
alert(`エラー: ${data.error}`);
}
} catch (error) {
addSystemLog(`ネットワークエラー: ${error.message}`, 'error');
alert('ネットワークエラーが発生しました');
}
}
// 長時間タスクを開始
async function startLongTask(event) {
event.preventDefault();
const duration = document.getElementById('taskDuration').value;
const taskType = document.querySelector('input[name="taskType"]:checked').value;
let endpoint = '/tasks/long-task';
if (taskType === 'retry') {
endpoint = '/tasks/test-retry-task';
}
try {
const response = await fetch(endpoint, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify({
duration: parseInt(duration),
data: 'テストデータ'
})
});
const data = await response.json();
if (response.ok) {
addSystemLog(`${taskType === 'retry' ? 'リトライ' : '長時間'}タスク開始: ${data.task_id}`, 'success');
// アクティブタスクに追加
activeTasks[data.task_id] = {
id: data.task_id,
type: taskType === 'retry' ? 'リトライタスク' : '長時間タスク',
status: 'PENDING',
progress: 0,
started: new Date().toLocaleTimeString()
};
updateActiveTasksUI();
alert('タスクを開始しました。タスク一覧で進捗を確認できます。');
} else {
addSystemLog(`タスク開始失敗: ${data.error}`, 'error');
alert(`エラー: ${data.error}`);
}
} catch (error) {
addSystemLog(`ネットワークエラー: ${error.message}`, 'error');
alert('ネットワークエラーが発生しました');
}
}
// タスク詳細を表示
async function showTaskDetail(taskId) {
try {
const response = await fetch(`/tasks/status/${taskId}`);
const data = await response.json();
const modalContent = document.getElementById('taskDetailContent');
if (response.ok) {
let html = `
<h6>タスクID</h6>
<p><code>${data.task_id}</code></p>
<h6>ステータス</h6>
<p><span class="task-status ${getStatusClass(data.status)}">
${getStatusText(data.status)}
</span></p>
`;
if (data.status === 'PROCESSING' && data.info) {
html += `
<h6>進捗状況</h6>
<p>${data.info.status || '処理中'}</p>
<div class="progress-container">
<div class="progress-bar"
style="width: ${data.info.current || 0}%"></div>
</div>
<p>${data.info.current || 0}% 完了</p>
`;
}
if (data.result) {
html += `
<h6>結果</h6>
<pre class="bg-light p-3"><code>${JSON.stringify(data.result, null, 2)}</code></pre>
`;
}
if (data.error) {
html += `
<h6>エラー</h6>
<div class="alert alert-danger">
${data.error}
</div>
`;
}
modalContent.innerHTML = html;
} else {
modalContent.innerHTML = `
<div class="alert alert-danger">
タスク詳細の取得に失敗しました: ${data.error}
</div>
`;
}
// モーダルを表示
const modal = new bootstrap.Modal(document.getElementById('taskDetailModal'));
modal.show();
} catch (error) {
console.error('タスク詳細取得エラー:', error);
}
}
// タスクをキャンセル
async function cancelTask(taskId) {
if (!confirm('このタスクをキャンセルしてもよろしいですか？')) {
return;
}
try {
const response = await fetch(`/tasks/cancel/${taskId}`, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify({ terminate: true })
});
const data = await response.json();
if (response.ok && data.success) {
addSystemLog(`タスクキャンセル: ${taskId}`, 'success');
// タスクステータスを更新
if (activeTasks[taskId]) {
activeTasks[taskId].status = 'REVOKED';
updateActiveTasksUI();
}
alert('タスクをキャンセルしました');
} else {
addSystemLog(`タスクキャンセル失敗: ${data.error || '不明なエラー'}`, 'error');
alert(`キャンセル失敗: ${data.error || '不明なエラー'}`);
}
} catch (error) {
addSystemLog(`ネットワークエラー: ${error.message}`, 'error');
alert('ネットワークエラーが発生しました');
}
}
// システムログを取得
async function fetchSystemLogs() {
// 実際の実装ではAPIからログを取得
// ここではサンプルログを追加
addSystemLog('システムログの取得を開始しました', 'info');
}
// システムログを追加
function addSystemLog(message, type = 'info') {
const timestamp = new Date().toLocaleTimeString();
const logEntry = {
timestamp: timestamp,
message: message,
type: type
};
systemLogs.unshift(logEntry); // 先頭に追加
// 最大100件まで保持
if (systemLogs.length > 100) {
systemLogs.pop();
}
updateSystemLogsUI();
}
// システムログUIを更新
function updateSystemLogsUI() {
const container = document.getElementById('systemLogs');
let html = '';
systemLogs.forEach(log => {
const typeClass = `log-${log.type}`;
html += `
<div class="log-entry">
<span class="log-time">[${log.timestamp}]</span>
<span class="${typeClass}"> ${log.message}</span>
</div>
`;
});
container.innerHTML = html;
}
// ログモニタリングを開始
function startLogMonitoring() {
if (logMonitoringInterval) {
clearInterval(logMonitoringInterval);
}
logMonitoringInterval = setInterval(() => {
// 定期的にシステム状態をチェックしてログに追加
const randomLogs = [
'システム状態をチェック中...',
'Redis接続を確認しました',
'Celeryワーカーが実行中です',
'メールキューを処理中です',
'データベース接続を確認しました'
];
const randomLog = randomLogs[Math.floor(Math.random() * randomLogs.length)];
addSystemLog(randomLog, 'info');
}, 5000); // 5秒ごとにログを追加
addSystemLog('ログモニタリングを開始しました', 'success');
}
// ログモニタリングを停止
function stopLogMonitoring() {
if (logMonitoringInterval) {
clearInterval(logMonitoringInterval);
logMonitoringInterval = null;
addSystemLog('ログモニタリングを停止しました', 'info');
}
}
// ログをクリア
function clearLogs() {
systemLogs.length = 0;
updateSystemLogsUI();
addSystemLog('ログをクリアしました', 'info');
}
</script>
</body>
</html>JavaScriptは定期的なポーリングでキュースタッツとアクティブタスクを更新し、システムログエリアには操作履歴や状態変化を時系列で出力します。管理者がシステム全体の状態を視覚的に把握し、タスクを制御するためのダッシュボードとして機能します。
まとめ
Flaskアプリケーションにおける非同期処理とバックグラウンドタスクの実装について学びました。まず、分散タスクキューシステムであるCeleryの導入と基本設定として、そのセットアップ方法を解説しました。
次に、長時間実行タスク、リトライ可能タスク、優先度付きタスクといった多様なバックグラウンドタスクの実装方法を扱いました。さらに、Celeryを活用した効率的なメール送信システムにより、メール送信機能の非同期化を実現しました。
タスク管理と監視の面では、タスクの状態追跡、キューの監視、システム健全性チェックの手法を紹介し、加えて、これらのタスクシステムを管理するためのWebインターフェースとして管理画面を作成しました。非同期処理を適切に実装することで、ユーザー体験の向上、サーバーリソースの効率的な活用、スケーラビリティの向上、そしてエラーハンドリングとリトライ機能の強化といった多くのメリットが得られます。
演習問題
初級問題（3問）
問題1：基本的な非同期タスクの作成
Celeryを使用して、引数で受け取った数値の合計を計算する非同期タスクを作成してください。タスクは進捗状況を報告できるようにしてください。
問題2：シンプルなメール送信タスク
ユーザーのメールアドレスと名前を受け取り、シンプルなウェルカムメールを送信するCeleryタスクを作成してください。メールの件名は「ようこそ、{名前}さん！」としてください。
問題3：タスクステータスの確認
指定されたタスクIDのステータスを確認する関数を作成してください。タスクが完了している場合は結果を、進行中の場合は進捗状況を返すようにしてください。
中級問題（6問）
問題4：リトライ機能付きAPI呼び出しタスク
外部APIを呼び出すタスクを作成してください。API呼び出しが失敗した場合、指数バックオフで最大3回までリトライするようにしてください。
問題5：バルクメール送信システム
CSVファイルからメールアドレスを読み込み、全ユーザーに一括でメールを送信するシステムを作成してください。メール送信の進捗状況をリアルタイムで報告できるようにしてください。
問題6：タスクチェーンとグループの実装
3つのタスクをチェーン（順次実行）で実行し、その結果を使用して別のタスクを実行するワークフローを作成してください。さらに、複数のタスクをグループ（並列実行）で実行する機能も実装してください。
問題7：優先度付きタスクキュー
高優先度、中優先度、低優先度の3つのキューを作成し、タスクの優先度に応じて適切なキューに振り分けるシステムを実装してください。
問題8：定期的なメンテナンスタスク
毎日午前3時に実行されるデータベースクリーニングタスクを作成してください。古いログレコードや期限切れのセッションを削除する機能を実装してください。
問題9：タスク結果の永続化
タスクの実行結果をデータベースに保存するシステムを作成してください。タスクID、ステータス、実行時間、結果（またはエラー情報）を記録できるようにしてください。
上級問題（3問）
問題10：分散タスク処理システム
複数のワーカーノードでタスクを分散処理するシステムを設計・実装してください。各ワーカーの負荷状況を監視し、適切にタスクを振り分けるロードバランシング機能を含めてください。
問題11：リアルタイムタスクモニタリングダッシュボード
WebSocketを使用して、タスクの実行状況をリアルタイムで表示するダッシュボードを作成してください。タスクの進捗状況、システムリソース使用率、キューの状態をリアルタイムで可視化してください。
問題12：フォールトトレラントなメール送信システム
メール送信に失敗した場合、自動的に代替のSMTPサーバーを使用するフォールトトレラントなメール送信システムを実装してください。送信履歴をデータベースに記録し、再送信機能も実装してください。
演習問題 解答例
初級問題
問題1：基本的な非同期タスクの作成
# app/tasks/exercise_tasks.py
from app.celery_worker import celery
import time
from celery.result import AsyncResult
@celery.task(bind=True)
def calculate_sum_task(self, numbers):
"""
数値の合計を計算する非同期タスク
Args:
numbers: 数値のリスト
Returns:
dict: 計算結果
"""
task_id = self.request.id
try:
total = len(numbers)
current_sum = 0
# 進捗状況の報告
for i, num in enumerate(numbers, 1):
# 数値を加算
current_sum += num
# 進捗状況を更新（25%, 50%, 75%, 100%）
progress = int((i / total) * 100)
# 中間結果を報告
self.update_state(
state='PROGRESS',
meta={
'current': progress,
'total': 100,
'status': f'計算中... {i}/{total}',
'current_sum': current_sum,
'numbers_processed': i
}
)
# 処理時間をシミュレート（0.1秒待機）
time.sleep(0.1)
# 最終結果を返す
result = {
'task_id': task_id,
'status': 'completed',
'total_numbers': total,
'sum': current_sum,
'average': current_sum / total if total > 0 else 0,
'message': '計算が完了しました'
}
return result
except Exception as e:
return {
'task_id': task_id,
'status': 'failed',
'error': str(e)
}
# 使用例
def test_calculate_sum():
"""計算タスクのテスト"""
# タスクを非同期で実行
task = calculate_sum_task.delay([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
print(f"タスクID: {task.id}")
# タスクの完了を待機
result = task.get(timeout=10)
print(f"結果: {result}")
# 進捗状況の監視（別の方法）
task_result = AsyncResult(task.id, app=celery)
if task_result.status == 'PROGRESS':
print(f"進捗状況: {task_result.info}")
return result
# タスクの状態を監視する関数
def monitor_task_progress(task_id, interval=0.5, timeout=30):
"""
タスクの進捗状況を監視
Args:
task_id: タスクID
interval: チェック間隔（秒）
timeout: タイムアウト時間（秒）
"""
import time
start_time = time.time()
while time.time() - start_time < timeout:
task_result = AsyncResult(task_id, app=celery)
if task_result.ready():
if task_result.successful():
print(f"タスク完了: {task_result.result}")
else:
print(f"タスク失敗: {task_result.result}")
break
if task_result.status == 'PROGRESS':
info = task_result.info
print(f"進捗: {info.get('current', 0)}% - {info.get('status', '')}")
print(f"現在の合計: {info.get('current_sum', 0)}")
time.sleep(interval)
if not task_result.ready():
print(f"タイムアウト: タスクは{timeout}秒以内に完了しませんでした")
return task_result.status
if __name__ == "__main__":
# テスト実行
test_result = test_calculate_sum()
# 大きなリストでテスト
large_list = list(range(1, 101)) # 1から100まで
large_task = calculate_sum_task.delay(large_list)
# 進捗状況を監視
monitor_task_progress(large_task.id)問題2：シンプルなメール送信タスク
<pre><code class="language-python"># app/tasks/email_exercises.py
from app.celery_worker import celery
from flask_mail import Message
from flask import current_app
from app.email import mail
import logging
logger = logging.getLogger(__name__)
@celery.task(bind=True)
def send_welcome_email_simple(self, email, name):
"""
シンプルなウェルカムメール送信タスク
Args:
email: ユーザーのメールアドレス
name: ユーザーの名前
Returns:
dict: 送信結果
"""
task_id = self.request.id
try:
# メールの作成
subject = f"ようこそ、{name}さん！"
# シンプルなHTMLメール
html_body = f"""
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<style>
body {{ font-family: Arial, sans-serif; line-height: 1.6; color: #333; }}
.container {{ max-width: 600px; margin: 0 auto; padding: 20px; }}
.header {{ background-color: #4CAF50; color: white; padding: 20px; text-align: center; }}
.content {{ padding: 30px; background-color: #f9f9f9; }}
.footer {{ text-align: center; padding: 20px; color: #777; font-size: 12px; }}
</style>
</head>
<body>
<div class="container">
<div class="header">
<h1>ようこそ！</h1>
</div>
<div class="content">
<h2>{name}さん、当サービスへようこそ！</h2>
<p>ご登録いただき、誠にありがとうございます。</p>
<p>以下の情報でアカウントが作成されました：</p>
<ul>
<li>メールアドレス: {email}</li>
<li>お名前: {name}</li>
</ul>
<p>サービスをご利用いただくには、ログインしてプロフィールを完成させてください。</p>
</div>
<div class="footer">
<p>このメールは自動送信されています。</p>
</div>
</div>
</body>
</html>
"""
# プレーンテキストバージョン
text_body = f"""
ようこそ、{name}さん！
当サービスへようこそ！ご登録いただき、誠にありがとうございます。
以下の情報でアカウントが作成されました：
・メールアドレス: {email}
・お名前: {name}
サービスをご利用いただくには、ログインしてプロフィールを完成させてください。
--
このメールは自動送信されています。
"""
# メッセージオブジェクトの作成
msg = Message(
subject=subject,
recipients=[email],
html=html_body,
body=text_body
)
# メール送信
with current_app.app_context():
mail.send(msg)
logger.info(f"ウェルカムメール送信成功: {email}")
return {
'task_id': task_id,
'status': 'success',
'email': email,
'name': name,
'message': 'ウェルカムメールを送信しました'
}
except Exception as e:
logger.error(f"ウェルカムメール送信エラー: {e}")
# リトライを試みる
try:
self.retry(exc=e, countdown=60, max_retries=3)
except Exception as retry_exc:
return {
'task_id': task_id,
'status': 'failed',
'email': email,
'error': str(e),
'retries_exhausted': True
}
@celery.task
def batch_welcome_emails(users_data):
"""
複数ユーザーへのウェルカムメール一括送信
Args:
users_data: ユーザーデータのリスト [{'email': '...', 'name': '...'}, ...]
Returns:
dict: バッチ送信結果
"""
import time
from datetime import datetime
total_users = len(users_data)
successful = 0
failed = []
logger.info(f"バッチウェルカムメール開始: {total_users}ユーザー")
start_time = datetime.now()
for i, user in enumerate(users_data, 1):
try:
email = user['email']
name = user['name']
# 個別のメール送信タスクを実行
result = send_welcome_email_simple.delay(email, name)
# 簡易的な完了待機（実際の運用では非同期で処理）
try:
result.get(timeout=30) # 30秒タイムアウト
successful += 1
logger.debug(f"メール送信成功 ({i}/{total_users}): {email}")
except Exception as e:
failed.append({'email': email, 'error': str(e)})
logger.warning(f"メール送信失敗 ({i}/{total_users}): {email} - {e}")
# 進捗ログ
if i % 10 == 0 or i == total_users:
logger.info(f"進捗: {i}/{total_users}ユーザー処理完了")
# メールサーバーへの負荷軽減のため少し待機
time.sleep(0.5)
except Exception as e:
failed.append({'email': user.get('email', 'unknown'), 'error': str(e)})
logger.error(f"ユーザー処理エラー ({i}/{total_users}): {e}")
end_time = datetime.now()
duration = (end_time - start_time).total_seconds()
result = {
'total_users': total_users,
'successful': successful,
'failed': len(failed),
'success_rate': (successful / total_users * 100) if total_users > 0 else 0,
'failed_details': failed,
'start_time': start_time.isoformat(),
'end_time': end_time.isoformat(),
'duration_seconds': duration,
'average_time_per_user': duration / total_users if total_users > 0 else 0
}
logger.info(f"バッチウェルカムメール完了: {successful}/{total_users}成功")
return result
# テスト用関数
def test_welcome_email():
"""ウェルカムメールのテスト"""
# 単一ユーザーへの送信テスト
test_email = "test@example.com"
test_name = "山田太郎"
print("単一ユーザーへのメール送信テスト...")
task = send_welcome_email_simple.delay(test_email, test_name)
# 結果を待機
try:
result = task.get(timeout=30)
print(f"結果: {result}")
except Exception as e:
print(f"エラー: {e}")
# 複数ユーザーへの送信テスト
test_users = [
{'email': 'user1@example.com', 'name': '鈴木一郎'},
{'email': 'user2@example.com', 'name': '佐藤花子'},
{'email': 'user3@example.com', 'name': '高橋健太'},
]
print("\n複数ユーザーへのメール送信テスト...")
batch_task = batch_welcome_emails.delay(test_users)
try:
batch_result = batch_task.get(timeout=60)
print(f"バッチ結果: {batch_result}")
except Exception as e:
print(f"バッチエラー: {e}")
# モックメール送信（テスト用）
class MockMail:
"""テスト用のモックメールクラス"""
def __init__(self):
self.sent_messages = []
def send(self, message):
self.sent_messages.append({
'subject': message.subject,
'recipients': message.recipients,
'html': message.html,
'body': message.body
})
print(f"モックメール送信: {message.subject} to {message.recipients}")
return True
if __name__ == "__main__":
# テスト環境の設定
import os
os.environ['FLASK_ENV'] = 'testing'
# モックを使用してテスト
print("=== ウェルカムメールタスクテスト ===")
# 実際のメール送信をモックに置き換え（テスト時）
test_welcome_email()</code></pre>問題3：タスクステータスの確認
# app/tasks/task_monitor.py
from app.celery_worker import celery
from celery.result import AsyncResult
from datetime import datetime
import json
class TaskStatusChecker:
"""タスクステータス確認ユーティリティ"""
@staticmethod
def get_task_status(task_id):
"""
タスクのステータスを詳細に取得
Args:
task_id: タスクID
Returns:
dict: タスクステータス情報
"""
try:
task_result = AsyncResult(task_id, app=celery)
# 基本情報
status_info = {
'task_id': task_id,
'status': task_result.status,
'ready': task_result.ready(),
'successful': task_result.successful(),
'failed': task_result.failed(),
'state': task_result.state,
'retried': getattr(task_result, 'retried', False),
'timestamp': datetime.now().isoformat()
}
# 進捗状況（進行中のタスク）
if task_result.status == 'PROGRESS':
info = task_result.info
if isinstance(info, dict):
status_info['progress'] = {
'current': info.get('current', 0),
'total': info.get('total', 100),
'percentage': info.get('current', 0),
'status': info.get('status', '処理中'),
'details': {k: v for k, v in info.items()
if k not in ['current', 'total', 'status']}
}
else:
status_info['progress'] = {
'info': str(info)
}
# 完了したタスクの結果
if task_result.ready():
if task_result.successful():
result = task_result.result
if isinstance(result, dict):
status_info['result'] = result
else:
status_info['result'] = {
'value': result,
'type': type(result).__name__
}
else:
# エラー情報
error_info = task_result.result
if isinstance(error_info, Exception):
status_info['error'] = {
'type': type(error_info).__name__,
'message': str(error_info),
'args': error_info.args if hasattr(error_info, 'args') else []
}
else:
status_info['error'] = {
'details': str(error_info)
}
# タスクメタデータ（可能な場合）
try:
from celery.backends.redis import RedisBackend
if isinstance(celery.backend, RedisBackend):
# Redisから追加情報を取得
key = celery.backend.get_key_for_task(task_id)
import redis
redis_client = redis.from_url(celery.conf.broker_url)
task_data = redis_client.get(key)
if task_data:
task_data = json.loads(task_data)
status_info['metadata'] = {
'created': task_data.get('date_done'),
'result': task_data.get('result'),
'traceback': task_data.get('traceback')
}
except:
pass # メタデータの取得に失敗しても続行
return status_info
except Exception as e:
return {
'task_id': task_id,
'status': 'ERROR',
'error': f'タスクステータス取得エラー: {str(e)}',
'timestamp': datetime.now().isoformat()
}
@staticmethod
def format_status_for_display(status_info):
"""
ステータス情報を読みやすい形式で表示
Args:
status_info: ステータス情報の辞書
Returns:
str: フォーマットされた文字列
"""
lines = []
lines.append(f"タスクID: {status_info.get('task_id', '不明')}")
lines.append(f"ステータス: {status_info.get('status', '不明')}")
if 'progress' in status_info:
prog = status_info['progress']
if isinstance(prog, dict):
lines.append(f"進捗: {prog.get('current', 0)}/{prog.get('total', 100)} "
f"({prog.get('percentage', 0)}%)")
if 'status' in prog:
lines.append(f"状態: {prog['status']}")
if 'result' in status_info:
result = status_info['result']
lines.append("\n結果:")
if isinstance(result, dict):
for key, value in result.items():
lines.append(f" {key}: {value}")
else:
lines.append(f" {result}")
if 'error' in status_info:
error = status_info['error']
lines.append("\nエラー:")
if isinstance(error, dict):
for key, value in error.items():
lines.append(f" {key}: {value}")
else:
lines.append(f" {error}")
lines.append(f"\n最終更新: {status_info.get('timestamp', '不明')}")
return "\n".join(lines)
@staticmethod
def wait_for_task_completion(task_id, timeout=300, interval=1):
"""
タスクの完了を待機
Args:
task_id: タスクID
timeout: タイムアウト時間（秒）
interval: チェック間隔（秒）
Returns:
dict: 最終ステータス情報
"""
import time
start_time = time.time()
last_progress = None
while time.time() - start_time < timeout:
status_info = TaskStatusChecker.get_task_status(task_id)
current_status = status_info.get('status')
# 進捗状況の表示（変化がある場合のみ）
if 'progress' in status_info:
prog = status_info['progress']
if isinstance(prog, dict):
current_progress = f"{prog.get('current', 0)}%"
if current_progress != last_progress:
print(f"進捗: {current_progress} - {prog.get('status', '')}")
last_progress = current_progress
# 完了または失敗したらループを抜ける
if current_status in ['SUCCESS', 'FAILURE', 'REVOKED']:
print(f"タスク完了: 状態={current_status}")
return status_info
# 指定間隔で待機
time.sleep(interval)
# タイムアウト
timeout_info = TaskStatusChecker.get_task_status(task_id)
timeout_info['timeout'] = True
timeout_info['timeout_seconds'] = timeout
print(f"タイムアウト: タスクは{timeout}秒以内に完了しませんでした")
return timeout_info
@staticmethod
def get_batch_status(task_ids):
"""
複数タスクのステータスを一括取得
Args:
task_ids: タスクIDのリスト
Returns:
dict: バッチステータス情報
"""
batch_info = {
'total_tasks': len(task_ids),
'tasks': {},
'summary': {
'pending': 0,
'progress': 0,
'success': 0,
'failure': 0,
'other': 0
},
'timestamp': datetime.now().isoformat()
}
for task_id in task_ids:
status_info = TaskStatusChecker.get_task_status(task_id)
batch_info['tasks'][task_id] = status_info
# サマリーを更新
status = status_info.get('status', 'UNKNOWN')
if status == 'PENDING':
batch_info['summary']['pending'] += 1
elif status == 'PROGRESS':
batch_info['summary']['progress'] += 1
elif status == 'SUCCESS':
batch_info['summary']['success'] += 1
elif status == 'FAILURE':
batch_info['summary']['failure'] += 1
else:
batch_info['summary']['other'] += 1
# 完了率を計算
total = batch_info['total_tasks']
if total > 0:
completed = batch_info['summary']['success'] + batch_info['summary']['failure']
batch_info['summary']['completion_rate'] = (completed / total) * 100
else:
batch_info['summary']['completion_rate'] = 0
return batch_info
# 使用例とテスト
def test_task_status_checker():
"""タスクステータスチェッカーのテスト"""
print("=== タスクステータスチェッカーテスト ===")
# テスト用のタスクを実行
from app.tasks.exercise_tasks import calculate_sum_task
# タスクを実行
test_task = calculate_sum_task.delay([1, 2, 3, 4, 5])
task_id = test_task.id
print(f"実行したタスクID: {task_id}")
# 即時にステータスを確認
print("\n1. 即時ステータス確認:")
status = TaskStatusChecker.get_task_status(task_id)
print(TaskStatusChecker.format_status_for_display(status))
# 完了を待機（短いタイムアウト）
print("\n2. 完了を待機:")
final_status = TaskStatusChecker.wait_for_task_completion(
task_id, timeout=10, interval=0.5
)
print(TaskStatusChecker.format_status_for_display(final_status))
# 複数タスクのステータス確認
print("\n3. 複数タスクのステータス確認:")
# 複数のタスクを実行
task_ids = []
for i in range(3):
task = calculate_sum_task.delay(list(range(1, (i+1)*3 + 1)))
task_ids.append(task.id)
# バッチステータスを取得
batch_status = TaskStatusChecker.get_batch_status(task_ids)
print(f"総タスク数: {batch_status['total_tasks']}")
print(f"サマリー: {batch_status['summary']}")
# 各タスクの詳細
for t_id, t_status in batch_status['tasks'].items():
print(f"\nタスク {t_id[:8]}...: {t_status.get('status')}")
if 'result' in t_status:
print(f" 結果: {t_status['result'].get('sum', 'N/A')}")
return {
'single_task': status,
'final_status': final_status,
'batch_status': batch_status
}
# コマンドラインインターフェース
def cli_task_status():
"""コマンドラインからのタスクステータス確認"""
import sys
if len(sys.argv) < 2:
print("使用方法: python task_monitor.py [timeout]")
print("例: python task_monitor.py abc123-def456 30")
sys.exit(1)
task_id = sys.argv[1]
timeout = int(sys.argv[2]) if len(sys.argv) > 2 else 30
print(f"タスク {task_id} のステータスを監視します...")
print(f"タイムアウト: {timeout}秒\n")
# タスクの完了を待機
result = TaskStatusChecker.wait_for_task_completion(task_id, timeout)
# 結果を表示
print("\n" + "="*50)
print("最終ステータス:")
print("="*50)
print(TaskStatusChecker.format_status_for_display(result))
# JSON出力も可能
if '--json' in sys.argv:
import json
print("\nJSON形式:")
print(json.dumps(result, indent=2, ensure_ascii=False))
if __name__ == "__main__":
# テスト実行
test_results = test_task_status_checker()
# コマンドラインインターフェースをテスト
# 実際の使用時は以下のように呼び出し:
# python -c "from app.tasks.task_monitor import cli_task_status; cli_task_status()" 中級問題
問題4：リトライ機能付きAPI呼び出しタスク
# app/tasks/api_tasks.py
from app.celery_worker import celery
import requests
import time
import logging
from datetime import datetime
from requests.exceptions import RequestException, Timeout, ConnectionError
logger = logging.getLogger(__name__)
class APIClient:
"""APIクライアントクラス（リトライ機能付き）"""
def __init__(self, base_url=None, default_timeout=30):
self.base_url = base_url
self.default_timeout = default_timeout
self.session = requests.Session()
# セッション設定
self.session.headers.update({
'User-Agent': 'FlaskApp/1.0',
'Accept': 'application/json',
})
def call_api_with_retry(self, method, endpoint, retries=3,
backoff_factor=1.5, **kwargs):
"""
リトライ機能付きAPI呼び出し
Args:
method: HTTPメソッド（'GET', 'POST', 'PUT', 'DELETE'）
endpoint: APIエンドポイント
retries: 最大リトライ回数
backoff_factor: バックオフ係数（指数バックオフ用）
**kwargs: requestsリクエストパラメータ
Returns:
tuple: (成功したか, レスポンスまたはエラー情報)
"""
url = f"{self.base_url}{endpoint}" if self.base_url else endpoint
# デフォルトタイムアウトの設定
if 'timeout' not in kwargs:
kwargs['timeout'] = self.default_timeout
for attempt in range(retries + 1): # 初回 + リトライ回数
try:
logger.info(f"API呼び出し試行 {attempt + 1}/{retries + 1}: {method} {url}")
# API呼び出し
response = self.session.request(method, url, **kwargs)
# ステータスコードチェック
if response.status_code >= 200 and response.status_code < 300:
logger.info(f"API呼び出し成功: {response.status_code}")
return True, {
'status_code': response.status_code,
'data': response.json() if response.content else None,
'headers': dict(response.headers),
'attempt': attempt + 1
}
else:
# ステータスコードエラー
error_msg = f"HTTPエラー: {response.status_code}"
logger.warning(f"{error_msg} - 試行 {attempt + 1}")
if attempt < retries:
# リトライ可能なエラー（5xx系）
if response.status_code >= 500:
sleep_time = backoff_factor ** attempt
logger.info(f"{sleep_time:.1f}秒待機してリトライします")
time.sleep(sleep_time)
continue
else:
# 4xx系エラーはリトライしない
return False, {
'error': error_msg,
'status_code': response.status_code,
'response_text': response.text[:500],
'attempt': attempt + 1
}
else:
# リトライ回数超過
return False, {
'error': error_msg,
'status_code': response.status_code,
'response_text': response.text[:500],
'retries_exhausted': True,
'attempt': attempt + 1
}
except Timeout as e:
error_msg = f"タイムアウトエラー: {str(e)}"
logger.warning(f"{error_msg} - 試行 {attempt + 1}")
if attempt < retries:
sleep_time = backoff_factor ** attempt
logger.info(f"{sleep_time:.1f}秒待機してリトライします")
time.sleep(sleep_time)
else:
return False, {
'error': error_msg,
'exception_type': 'Timeout',
'retries_exhausted': True,
'attempt': attempt + 1
}
except ConnectionError as e:
error_msg = f"接続エラー: {str(e)}"
logger.warning(f"{error_msg} - 試行 {attempt + 1}")
if attempt < retries:
sleep_time = backoff_factor ** attempt
logger.info(f"{sleep_time:.1f}秒待機してリトライします")
time.sleep(sleep_time)
else:
return False, {
'error': error_msg,
'exception_type': 'ConnectionError',
'retries_exhausted': True,
'attempt': attempt + 1
}
except RequestException as e:
error_msg = f"リクエストエラー: {str(e)}"
logger.warning(f"{error_msg} - 試行 {attempt + 1}")
if attempt < retries:
sleep_time = backoff_factor ** attempt
logger.info(f"{sleep_time:.1f}秒待機してリトライします")
time.sleep(sleep_time)
else:
return False, {
'error': error_msg,
'exception_type': 'RequestException',
'retries_exhausted': True,
'attempt': attempt + 1
}
except Exception as e:
error_msg = f"予期せぬエラー: {str(e)}"
logger.error(f"{error_msg} - 試行 {attempt + 1}")
# 予期せぬエラーはリトライしない
return False, {
'error': error_msg,
'exception_type': type(e).__name__,
'attempt': attempt + 1
}
# ここには通常到達しない
return False, {'error': '不明なエラー'}
@celery.task(bind=True, max_retries=3)
def call_external_api_task(self, endpoint, method='GET', data=None,
params=None, headers=None, retries=3):
"""
外部API呼び出しタスク（Celeryリトライ機能を使用）
Args:
endpoint: APIエンドポイント
method: HTTPメソッド
data: 送信データ（POST/PUT用）
params: クエリパラメータ
headers: 追加ヘッダー
retries: リトライ回数（Celeryのmax_retriesも使用）
Returns:
dict: API呼び出し結果
"""
task_id = self.request.id
logger.info(f"外部API呼び出しタスク開始: {task_id}")
try:
# APIクライアントの作成
api_client = APIClient()
# リクエストパラメータの準備
request_kwargs = {}
if data is not None:
request_kwargs['json'] = data
if params is not None:
request_kwargs['params'] = params
if headers is not None:
request_kwargs['headers'] = headers
# 進捗状況の報告
self.update_state(
state='PROGRESS',
meta={
'current': 25,
'total': 100,
'status': 'APIを呼び出しています...'
}
)
# API呼び出し（アプリケーションレベルのリトライ）
success, result = api_client.call_api_with_retry(
method=method,
endpoint=endpoint,
retries=retries,
**request_kwargs
)
if success:
# 成功時の処理
self.update_state(
state='PROGRESS',
meta={
'current': 100,
'total': 100,
'status': 'API呼び出し成功'
}
)
api_result = {
'task_id': task_id,
'status': 'success',
'api_call': {
'endpoint': endpoint,
'method': method,
'attempts': result.get('attempt', 1)
},
'response': {
'status_code': result.get('status_code'),
'data': result.get('data'),
'received_at': datetime.now().isoformat()
},
'message': 'API呼び出しが成功しました'
}
logger.info(f"API呼び出し成功: {task_id}")
return api_result
else:
# 失敗時の処理（Celeryリトライを使用）
error_msg = result.get('error', 'API呼び出しに失敗しました')
logger.warning(f"API呼び出し失敗、リトライします: {task_id} - {error_msg}")
# Celeryのリトライ機能を使用
raise self.retry(
exc=Exception(error_msg),
countdown=60, # 60秒後にリトライ
max_retries=self.max_retries
)
except Exception as e:
logger.error(f"API呼び出しタスクエラー {task_id}: {e}")
# リトライ回数超過または予期せぬエラー
return {
'task_id': task_id,
'status': 'failed',
'api_call': {
'endpoint': endpoint,
'method': method
},
'error': str(e),
'retries_exhausted': True,
'failed_at': datetime.now().isoformat()
}
@celery.task(bind=True)
def batch_api_calls_task(self, api_calls, max_concurrent=3):
"""
複数API呼び出しのバッチ処理
Args:
api_calls: API呼び出し設定のリスト
max_concurrent: 同時実行数
Returns:
dict: バッチ処理結果
"""
import concurrent.futures
from functools import partial
task_id = self.request.id
total_calls = len(api_calls)
logger.info(f"バッチAPI呼び出し開始: {task_id} - {total_calls}件")
results = {
'total_calls': total_calls,
'successful': 0,
'failed': 0,
'calls': [],
'start_time': datetime.now().isoformat()
}
try:
# 進捗状況の初期報告
self.update_state(
state='PROGRESS',
meta={
'current': 0,
'total': total_calls,
'status': f'0/{total_calls}件処理完了'
}
)
# APIクライアントの作成
api_client = APIClient()
# 各API呼び出しを実行する関数
def execute_single_api_call(api_config, index):
"""単一API呼び出しを実行"""
try:
call_info = {
'index': index,
'config': api_config,
'started_at': datetime.now().isoformat()
}
# API呼び出し
success, result = api_client.call_api_with_retry(
method=api_config.get('method', 'GET'),
endpoint=api_config['endpoint'],
retries=api_config.get('retries', 2),
data=api_config.get('data'),
params=api_config.get('params'),
headers=api_config.get('headers')
)
call_info['completed_at'] = datetime.now().isoformat()
call_info['success'] = success
call_info['result'] = result
return call_info
except Exception as e:
call_info['completed_at'] = datetime.now().isoformat()
call_info['success'] = False
call_info['error'] = str(e)
return call_info
# スレッドプールで並列実行
completed = 0
with concurrent.futures.ThreadPoolExecutor(max_workers=max_concurrent) as executor:
# 各API呼び出しをスケジュール
future_to_index = {}
for i, api_config in enumerate(api_calls):
future = executor.submit(execute_single_api_call, api_config, i)
future_to_index[future] = i
# 結果を収集
for future in concurrent.futures.as_completed(future_to_index):
index = future_to_index[future]
try:
call_result = future.result()
results['calls'].append(call_result)
if call_result['success']:
results['successful'] += 1
else:
results['failed'] += 1
completed += 1
# 進捗状況を更新
progress_percent = int((completed / total_calls) * 100)
self.update_state(
state='PROGRESS',
meta={
'current': progress_percent,
'total': 100,
'status': f'{completed}/{total_calls}件処理完了',
'successful': results['successful'],
'failed': results['failed']
}
)
logger.debug(f"API呼び出し {index} 完了: {call_result['success']}")
except Exception as e:
logger.error(f"API呼び出し {index} で例外: {e}")
results['failed'] += 1
completed += 1
# 最終結果
results['end_time'] = datetime.now().isoformat()
results['success_rate'] = (results['successful'] / total_calls * 100) if total_calls > 0 else 0
logger.info(f"バッチAPI呼び出し完了: {task_id} - "
f"{results['successful']}/{total_calls}成功")
return results
except Exception as e:
logger.error(f"バッチAPI呼び出しタスクエラー {task_id}: {e}")
return {
'task_id': task_id,
'status': 'failed',
'error': str(e),
'completed_calls': len(results['calls']),
'failed_at': datetime.now().isoformat()
}
# モックAPIサーバー（テスト用）
class MockAPIServer:
"""テスト用のモックAPIサーバー"""
def __init__(self, success_rate=0.7, average_delay=0.5):
"""
Args:
success_rate: 成功確率（0.0〜1.0）
average_delay: 平均応答遅延（秒）
"""
self.success_rate = success_rate
self.average_delay = average_delay
self.request_count = 0
self.success_count = 0
self.failure_count = 0
def mock_api_call(self, endpoint, method='GET', data=None):
"""モックAPI呼び出し"""
import time
import random
self.request_count += 1
# 応答遅延をシミュレート
delay = random.uniform(self.average_delay * 0.5, self.average_delay * 1.5)
time.sleep(delay)
# 成功/失敗をランダムに決定
if random.random() < self.success_rate:
self.success_count += 1
# 成功レスポンス
response_data = {
'status': 'success',
'endpoint': endpoint,
'method': method,
'data_received': data,
'request_id': f'req_{self.request_count:06d}',
'timestamp': datetime.now().isoformat(),
'message': 'API呼び出し成功'
}
return True, {
'status_code': 200,
'data': response_data,
'headers': {'Content-Type': 'application/json'},
'attempt': 1
}
else:
self.failure_count += 1
# 失敗レスポンス（ランダムなエラータイプ）
error_types = [
('timeout', 'リクエストがタイムアウトしました', 504),
('server_error', '内部サーバーエラー', 500),
('bad_request', '無効なリクエストです', 400),
('not_found', 'リソースが見つかりません', 404),
]
error_type, error_msg, status_code = random.choice(error_types)
return False, {
'error': error_msg,
'status_code': status_code,
'response_text': f'{{"error": "{error_type}", "message": "{error_msg}"}}',
'attempt': 1
}
# テスト関数
def test_api_tasks():
"""APIタスクのテスト"""
print("=== API呼び出しタスクテスト ===")
# モックAPIサーバーの作成
mock_server = MockAPIServer(success_rate=0.5, average_delay=0.2)
# 単一API呼び出しのテスト
print("\n1. 単一API呼び出しテスト:")
# 実際の実装では外部APIを呼び出すが、ここではモックを使用
# テスト用のエンドポイント
test_endpoint = "https://api.example.com/test"
# タスクを実行
task = call_external_api_task.delay(
endpoint=test_endpoint,
method='GET',
retries=2
)
# 結果を待機
try:
result = task.get(timeout=30)
print(f"結果: {result.get('status')}")
if result.get('status') == 'success':
print(f"ステータスコード: {result.get('response', {}).get('status_code')}")
else:
print(f"エラー: {result.get('error')}")
except Exception as e:
print(f"タスク実行エラー: {e}")
# バッチAPI呼び出しのテスト
print("\n2. バッチAPI呼び出しテスト:")
# テスト用のAPI呼び出し設定
api_calls = [
{
'endpoint': 'https://api.example.com/users/1',
'method': 'GET'
},
{
'endpoint': 'https://api.example.com/users',
'method': 'POST',
'data': {'name': 'Test User', 'email': 'test@example.com'}
},
{
'endpoint': 'https://api.example.com/products',
'method': 'GET',
'params': {'category': 'electronics', 'limit': 10}
},
{
'endpoint': 'https://api.example.com/orders/123',
'method': 'PUT',
'data': {'status': 'shipped'},
'headers': {'Authorization': 'Bearer token123'}
}
]
batch_task = batch_api_calls_task.delay(api_calls, max_concurrent=2)
try:
batch_result = batch_task.get(timeout=60)
print(f"バッチ結果:")
print(f" 総数: {batch_result.get('total_calls', 0)}")
print(f" 成功: {batch_result.get('successful', 0)}")
print(f" 失敗: {batch_result.get('failed', 0)}")
print(f" 成功率: {batch_result.get('success_rate', 0):.1f}%")
except Exception as e:
print(f"バッチタスク実行エラー: {e}")
# APIClientの直接テスト
print("\n3. APIClient直接テスト:")
client = APIClient()
# 実際のAPIを呼び出す代わりに、モックを使用したテスト
# 注意: これは実際のネットワーク呼び出しではありません
print("注: 実際のネットワーク呼び出しはスキップされます")
return {
'single_task': task.id if 'task' in locals() else None,
'batch_task': batch_task.id if 'batch_task' in locals() else None
}
if __name__ == "__main__":
test_api_tasks()問題5：バルクメール送信システム
# app/tasks/bulk_email_tasks.py
from app.celery_worker import celery
from app.email import send_email_sync, EmailTemplate
import csv
import os
import time
from datetime import datetime, timedelta
import logging
from io import StringIO
import pandas as pd
logger = logging.getLogger(__name__)
class BulkEmailSystem:
"""バルクメール送信システム"""
def __init__(self, temp_dir='./temp_emails'):
self.temp_dir = temp_dir
os.makedirs(temp_dir, exist_ok=True)
def read_email_list_from_csv(self, csv_content, has_headers=True):
"""
CSVからメールリストを読み込む
Args:
csv_content: CSV文字列またはファイルパス
has_headers: ヘッダー行があるか
Returns:
list: メールデータのリスト
"""
emails = []
try:
# CSVコンテンツが文字列の場合はStringIOでラップ
if isinstance(csv_content, str) and '\n' in csv_content:
# 文字列としてのCSVデータ
csv_io = StringIO(csv_content)
reader = csv.DictReader(csv_io) if has_headers else csv.reader(csv_io)
else:
# ファイルパスとして扱う
with open(csv_content, 'r', encoding='utf-8') as f:
reader = csv.DictReader(f) if has_headers else csv.reader(f)
# データを読み込み
for row in reader:
if has_headers:
# ヘッダーありの場合は辞書として処理
email_data = {
'email': row.get('email') or row.get('Email') or row.get('メールアドレス'),
'name': row.get('name') or row.get('Name') or row.get('名前') or '',
'company': row.get('company') or row.get('Company') or row.get('会社名') or '',
'custom_field': row # 全フィールドを含む
}
else:
# ヘッダーなしの場合は位置で処理
email_data = {
'email': row[0] if len(row) > 0 else '',
'name': row[1] if len(row) > 1 else '',
'company': row[2] if len(row) > 2 else '',
'custom_field': row
}
# メールアドレスのバリデーション
if email_data['email'] and '@' in email_data['email']:
emails.append(email_data)
else:
logger.warning(f"無効なメールアドレスをスキップ: {email_data}")
logger.info(f"CSVから {len(emails)} 件のメールアドレスを読み込みました")
except Exception as e:
logger.error(f"CSV読み込みエラー: {e}")
raise
return emails
def validate_email_list(self, emails, max_emails=10000):
"""
メールリストを検証
Args:
emails: メールデータのリスト
max_emails: 最大送信件数
Returns:
dict: 検証結果
"""
valid_emails = []
invalid_emails = []
duplicates = set()
seen_emails = set()
for i, email_data in enumerate(emails):
email = email_data.get('email', '').strip().lower()
# 基本的なバリデーション
if not email:
invalid_emails.append({
'index': i,
'data': email_data,
'reason': 'メールアドレスが空です'
})
continue
if '@' not in email or '.' not in email.split('@')[-1]:
invalid_emails.append({
'index': i,
'data': email_data,
'reason': '無効なメールアドレス形式です'
})
continue
# 重複チェック
if email in seen_emails:
duplicates.add(email)
continue
seen_emails.add(email)
valid_emails.append(email_data)
# 最大件数チェック
if len(valid_emails) > max_emails:
valid_emails = valid_emails[:max_emails]
logger.warning(f"メール数が最大件数({max_emails})を超えました。最初の{max_emails}件のみ処理します")
result = {
'total_emails': len(emails),
'valid_emails': len(valid_emails),
'invalid_emails': len(invalid_emails),
'duplicate_emails': len(duplicates),
'valid_list': valid_emails,
'invalid_list': invalid_emails[:100], # 最初の100件のみ保持
'duplicate_list': list(duplicates)[:100]
}
return result
def create_email_batches(self, emails, batch_size=100):
"""
メールをバッチに分割
Args:
emails: メールリスト
batch_size: バッチサイズ
Returns:
list: バッチのリスト
"""
batches = []
for i in range(0, len(emails), batch_size):
batch = emails[i:i + batch_size]
batches.append({
'batch_number': len(batches) + 1,
'emails': batch,
'start_index': i,
'end_index': min(i + batch_size, len(emails)) - 1
})
logger.info(f"{len(emails)}件のメールを{len(batches)}バッチに分割しました")
return batches
@celery.task(bind=True)
def send_bulk_email_task(self, csv_content, template_name, template_vars=None,
subject=None, batch_size=100, delay_between_emails=1,
max_emails=10000, has_headers=True):
"""
バルクメール送信タスク
Args:
csv_content: CSVコンテンツ（文字列またはファイルパス）
template_name: 使用するテンプレート名
template_vars: テンプレート変数（辞書）
subject: 件名（テンプレートのsubjectを上書きする場合）
batch_size: バッチサイズ
delay_between_emails: メール間の遅延（秒）
max_emails: 最大送信件数
has_headers: CSVにヘッダー行があるか
Returns:
dict: 送信結果
"""
task_id = self.request.id
logger.info(f"バルクメール送信タスク開始: {task_id}")
start_time = datetime.now()
bulk_system = BulkEmailSystem()
try:
# ステップ1: CSVからメールリストを読み込み
self.update_state(
state='PROGRESS',
meta={
'current': 10,
'total': 100,
'status': 'CSVファイルを読み込んでいます...',
'step': 'reading_csv'
}
)
emails = bulk_system.read_email_list_from_csv(
csv_content, has_headers=has_headers
)
# ステップ2: メールリストを検証
self.update_state(
state='PROGRESS',
meta={
'current': 20,
'total': 100,
'status': 'メールリストを検証しています...',
'step': 'validating',
'total_emails': len(emails)
}
)
validation_result = bulk_system.validate_email_list(
emails, max_emails=max_emails
)
valid_emails = validation_result['valid_list']
total_valid = validation_result['valid_emails']
# ステップ3: バッチに分割
self.update_state(
state='PROGRESS',
meta={
'current': 30,
'total': 100,
'status': f'メールをバッチに分割しています... ({total_valid}件)',
'step': 'batching',
'valid_emails': total_valid
}
)
batches = bulk_system.create_email_batches(valid_emails, batch_size)
# 結果を格納する変数
total_sent = 0
total_failed = 0
failed_emails = []
batch_results = []
# ステップ4: バッチごとにメール送信
for batch_index, batch in enumerate(batches):
batch_num = batch['batch_number']
batch_emails = batch['emails']
# バッチ開始
self.update_state(
state='PROGRESS',
meta={
'current': 40 + int((batch_index / len(batches)) * 50),
'total': 100,
'status': f'バッチ {batch_num}/{len(batches)} を処理中...',
'step': 'sending_batch',
'batch': batch_num,
'total_batches': len(batches),
'sent_so_far': total_sent,
'failed_so_far': total_failed
}
)
batch_start_time = datetime.now()
batch_sent = 0
batch_failed = 0
batch_failed_list = []
# バッチ内の各メールを送信
for i, email_data in enumerate(batch_emails):
email = email_data['email']
name = email_data.get('name', '')
try:
# テンプレート変数を準備
email_template_vars = (template_vars or {}).copy()
email_template_vars.update({
'user_name': name,
'user_email': email,
**email_data.get('custom_field', {})
})
# メールテンプレートを取得
mail_subject, html_body, text_body = EmailTemplate.get_template(
template_name, **email_template_vars
)
# 件名を上書きする場合
if subject:
mail_subject = subject
# メール送信
success = send_email_sync(
to=email,
subject=mail_subject,
html_body=html_body,
text_body=text_body
)
if success:
batch_sent += 1
total_sent += 1
else:
batch_failed += 1
total_failed += 1
batch_failed_list.append({
'email': email,
'reason': '送信失敗'
})
# 進捗状況を更新（バッチ内）
if (i + 1) % 10 == 0 or i == len(batch_emails) - 1:
self.update_state(
state='PROGRESS',
meta={
'current': 40 + int(((batch_index + (i + 1) / len(batch_emails)) / len(batches)) * 50),
'total': 100,
'status': f'バッチ {batch_num}: {i + 1}/{len(batch_emails)}件完了',
'step': 'sending_batch',
'batch': batch_num,
'batch_progress': f'{i + 1}/{len(batch_emails)}',
'sent_so_far': total_sent,
'failed_so_far': total_failed
}
)
# メール間の遅延
if delay_between_emails > 0 and i < len(batch_emails) - 1:
time.sleep(delay_between_emails)
except Exception as e:
batch_failed += 1
total_failed += 1
batch_failed_list.append({
'email': email,
'reason': str(e)[:200] # エラーメッセージを切り詰め
})
logger.error(f"メール送信エラー {email}: {e}")
batch_end_time = datetime.now()
batch_duration = (batch_end_time - batch_start_time).total_seconds()
# バッチ結果を保存
batch_results.append({
'batch_number': batch_num,
'total_emails': len(batch_emails),
'sent': batch_sent,
'failed': batch_failed,
'start_time': batch_start_time.isoformat(),
'end_time': batch_end_time.isoformat(),
'duration_seconds': batch_duration,
'failed_emails': batch_failed_list[:50] # 最初の50件のみ
})
# バッチ完了ログ
logger.info(f"バッチ {batch_num} 完了: {batch_sent}件成功, {batch_failed}件失敗")
# バッチ間の遅延（負荷軽減のため）
if batch_index < len(batches) - 1:
time.sleep(2)
# ステップ5: 最終結果の計算
end_time = datetime.now()
total_duration = (end_time - start_time).total_seconds()
# 失敗したメールを集計
for batch in batch_results:
failed_emails.extend(batch['failed_emails'])
# 最終結果
result = {
'task_id': task_id,
'status': 'completed',
'summary': {
'total_attempted': validation_result['total_emails'],
'total_valid': total_valid,
'total_sent': total_sent,
'total_failed': total_failed,
'success_rate': (total_sent / total_valid * 100) if total_valid > 0 else 0,
'invalid_emails': validation_result['invalid_emails'],
'duplicate_emails': validation_result['duplicate_emails']
},
'timing': {
'start_time': start_time.isoformat(),
'end_time': end_time.isoformat(),
'total_duration_seconds': total_duration,
'average_time_per_email': total_duration / total_sent if total_sent > 0 else 0
},
'batch_results': batch_results,
'failed_emails': failed_emails[:200], # 最初の200件のみ
'template_used': template_name,
'batch_size': batch_size,
'delay_between_emails': delay_between_emails
}
logger.info(f"バルクメール送信完了: {task_id} - "
f"{total_sent}/{total_valid}件成功 ({result['summary']['success_rate']:.1f}%)")
return result
except Exception as e:
logger.error(f"バルクメール送信タスクエラー {task_id}: {e}")
# 部分的な結果があれば含める
partial_result = {
'task_id': task_id,
'status': 'failed',
'error': str(e),
'partial_summary': {
'total_sent': total_sent if 'total_sent' in locals() else 0,
'total_failed': total_failed if 'total_failed' in locals() else 0,
'failed_at_step': self.current_state().get('meta', {}).get('step', 'unknown')
},
'failed_at': datetime.now().isoformat()
}
return partial_result
@celery.task
def send_scheduled_bulk_email(csv_file_path, send_datetime, template_name,
template_vars=None, timezone='Asia/Tokyo'):
"""
スケジュールされたバルクメール送信
Args:
csv_file_path: CSVファイルパス
send_datetime: 送信日時（ISO形式文字列）
template_name: テンプレート名
template_vars: テンプレート変数
timezone: タイムゾーン
Returns:
dict: 送信結果
"""
import pytz
from datetime import datetime
logger.info(f"スケジュールバルクメール設定: {send_datetime}")
try:
# 送信時間の解析
send_time = datetime.fromisoformat(send_datetime.replace('Z', '+00:00'))
tz = pytz.timezone(timezone)
send_time = send_time.astimezone(tz)
current_time = datetime.now(tz)
# 送信時間までの待機時間を計算
time_to_wait = (send_time - current_time).total_seconds()
if time_to_wait > 0:
logger.info(f"スケジュールバルクメール: {time_to_wait:.0f}秒待機")
time.sleep(time_to_wait)
elif time_to_wait < -300: # 5分以上過去
logger.error(f"スケジュールバルクメール: 送信時間が5分以上過去です")
return {
'status': 'failed',
'error': '送信時間が5分以上過去です',
'scheduled_for': send_datetime,
'current_time': current_time.isoformat()
}
# バルクメール送信タスクを実行
with open(csv_file_path, 'r', encoding='utf-8') as f:
csv_content = f.read()
task = send_bulk_email_task.delay(
csv_content=csv_content,
template_name=template_name,
template_vars=template_vars
)
# タスクの完了を待機
result = task.get(timeout=3600) # 1時間タイムアウト
# スケジュール情報を追加
result['scheduled_info'] = {
'scheduled_for': send_datetime,
'actual_start_time': current_time.isoformat(),
'delay_seconds': max(0, time_to_wait)
}
return result
except Exception as e:
logger.error(f"スケジュールバルクメールエラー: {e}")
return {
'status': 'failed',
'error': str(e),
'failed_at': datetime.now().isoformat()
}
# テスト用のCSV生成関数
def create_test_csv(num_emails=50):
"""テスト用のCSVファイルを作成"""
import tempfile
import csv
# 一時ファイルを作成
temp_file = tempfile.NamedTemporaryFile(mode='w', suffix='.csv',
encoding='utf-8', delete=False)
# CSVライターを作成
writer = csv.writer(temp_file)
# ヘッダー行
writer.writerow(['email', 'name', 'company'])
# テストデータを生成
for i in range(num_emails):
email = f'test_user_{i+1}@example.com'
name = f'テストユーザー {i+1}'
company = f'テスト会社 {(i % 5) + 1}'
writer.writerow([email, name, company])
temp_file.close()
logger.info(f"テストCSVファイルを作成しました: {temp_file.name}, {num_emails}件")
return temp_file.name
# テスト関数
def test_bulk_email_system():
"""バルクメールシステムのテスト"""
print("=== バルクメールシステムテスト ===")
# テストCSVファイルの作成
csv_file = create_test_csv(10) # 10件のテストデータ
try:
# CSVファイルの内容を読み込み
with open(csv_file, 'r', encoding='utf-8') as f:
csv_content = f.read()
print(f"CSVファイル: {csv_file}")
print(f"CSV内容（最初の200文字）: {csv_content[:200]}...")
# バルクメール送信タスクのテスト
print("\n1. バルクメール送信タスクテスト:")
# テスト用のテンプレート変数
template_vars = {
'app_name': 'テストアプリケーション',
'welcome_message': '当サービスへようこそ！',
'action_link': 'https://example.com/dashboard',
'expiry_days': 7
}
# タスクを実行（実際のメール送信はスキップ）
print("注: 実際のメール送信はテストモードのためスキップされます")
# 代わりにシステムのテスト
bulk_system = BulkEmailSystem()
# CSV読み込みテスト
emails = bulk_system.read_email_list_from_csv(csv_content, has_headers=True)
print(f"読み込まれたメール数: {len(emails)}")
# バリデーションテスト
validation = bulk_system.validate_email_list(emails, max_emails=1000)
print(f"有効なメール数: {validation['valid_emails']}")
print(f"無効なメール数: {validation['invalid_emails']}")
print(f"重複メール数: {validation['duplicate_emails']}")
# バッチ分割テスト
batches = bulk_system.create_email_batches(validation['valid_list'], batch_size=3)
print(f"作成されたバッチ数: {len(batches)}")
for i, batch in enumerate(batches[:3]): # 最初の3バッチのみ表示
print(f" バッチ {batch['batch_number']}: {len(batch['emails'])}件")
# 実際のタスク実行はコメントアウト（テスト時）
# task = send_bulk_email_task.delay(
# csv_content=csv_content,
# template_name='welcome',
# template_vars=template_vars,
# batch_size=5,
# delay_between_emails=0.1
# )
# print(f"タスクID: {task.id}")
finally:
# 一時ファイルの削除
if os.path.exists(csv_file):
os.unlink(csv_file)
print(f"\n一時ファイルを削除しました: {csv_file}")
return True
if __name__ == "__main__":
test_bulk_email_system()問題6：タスクチェーンとグループの実装
# app/tasks/chained_tasks.py
from app.celery_worker import celery
from celery import chain, group, chord
import time
from datetime import datetime
import logging
logger = logging.getLogger(__name__)
@celery.task
def fetch_data_task(source):
"""
データ取得タスク（チェーンの最初のステップ）
Args:
source: データソース
Returns:
dict: 取得したデータ
"""
logger.info(f"データ取得開始: {source}")
# データ取得のシミュレーション
time.sleep(2)
# サンプルデータ
data = {
'source': source,
'data': [
{'id': 1, 'value': 100},
{'id': 2, 'value': 200},
{'id': 3, 'value': 300},
],
'fetched_at': datetime.now().isoformat(),
'item_count': 3
}
logger.info(f"データ取得完了: {source} - {data['item_count']}件")
return data
@celery.task
def process_data_task(data):
"""
データ処理タスク（チェーンの2番目のステップ）
Args:
data: 前のタスクからのデータ
Returns:
dict: 処理されたデータ
"""
logger.info(f"データ処理開始: {data.get('source', 'unknown')}")
# データ処理のシミュレーション
time.sleep(3)
source_data = data.get('data', [])
# データを処理（例: 値を2倍にする）
processed_data = []
total_value = 0
for item in source_data:
processed_item = item.copy()
processed_item['processed_value'] = item['value'] * 2
processed_item['processed_at'] = datetime.now().isoformat()
processed_data.append(processed_item)
total_value += processed_item['processed_value']
result = {
'source': data.get('source'),
'original_data': source_data,
'processed_data': processed_data,
'total_processed_value': total_value,
'average_value': total_value / len(processed_data) if processed_data else 0,
'processed_at': datetime.now().isoformat()
}
logger.info(f"データ処理完了: {result['source']} - 合計値: {result['total_processed_value']}")
return result
@celery.task
def analyze_data_task(processed_data):
"""
データ分析タスク（チェーンの3番目のステップ）
Args:
processed_data: 処理されたデータ
Returns:
dict: 分析結果
"""
logger.info(f"データ分析開始: {processed_data.get('source', 'unknown')}")
# データ分析のシミュレーション
time.sleep(1)
data_items = processed_data.get('processed_data', [])
# 簡単な分析
if data_items:
values = [item['processed_value'] for item in data_items]
analysis = {
'source': processed_data.get('source'),
'count': len(values),
'sum': sum(values),
'average': sum(values) / len(values),
'min': min(values),
'max': max(values),
'range': max(values) - min(values),
'analysis_time': datetime.now().isoformat()
}
else:
analysis = {
'source': processed_data.get('source'),
'error': '分析するデータがありません',
'analysis_time': datetime.now().isoformat()
}
logger.info(f"データ分析完了: {analysis['source']} - 件数: {analysis.get('count', 0)}")
return analysis
@celery.task
def save_results_task(analysis_results):
"""
結果保存タスク（チェーンの最後のステップ）
Args:
analysis_results: 分析結果
Returns:
dict: 保存結果
"""
logger.info(f"結果保存開始: {analysis_results.get('source', 'unknown')}")
# 結果保存のシミュレーション
time.sleep(1)
# 実際の実装ではデータベースなどに保存
save_result = {
'source': analysis_results.get('source'),
'saved_data': analysis_results,
'saved_at': datetime.now().isoformat(),
'save_status': 'success',
'message': '結果を正常に保存しました'
}
logger.info(f"結果保存完了: {save_result['source']}")
return save_result
@celery.task
def parallel_processing_task(item):
"""
並列処理タスク（グループ用）
Args:
item: 処理するアイテム
Returns:
dict: 処理結果
"""
task_id = celery.current_task.request.id if hasattr(celery.current_task, 'request') else 'unknown'
logger.info(f"並列処理開始 [{task_id[:8]}]: {item.get('id', 'unknown')}")
# 処理のシミュレーション（ランダムな時間）
import random
process_time = random.uniform(0.5, 2.0)
time.sleep(process_time)
# アイテムを処理
result = item.copy()
result['processed'] = True
result['processed_value'] = item.get('value', 0) * random.uniform(0.8, 1.2)
result['process_time'] = process_time
result['processed_by'] = task_id[:8]
result['processed_at'] = datetime.now().isoformat()
logger.info(f"並列処理完了 [{task_id[:8]}]: {item.get('id', 'unknown')} - "
f"結果: {result['processed_value']:.1f}")
return result
@celery.task
def aggregate_results_task(parallel_results):
"""
並列処理結果の集計タスク（コード用）
Args:
parallel_results: 並列処理の結果リスト
Returns:
dict: 集計結果
"""
logger.info(f"結果集計開始: {len(parallel_results)}件の結果")
if not parallel_results:
return {
'status': 'error',
'message': '集計する結果がありません',
'aggregated_at': datetime.now().isoformat()
}
# 結果を集計
total_items = len(parallel_results)
total_value = sum(item.get('processed_value', 0) for item in parallel_results)
avg_value = total_value / total_items if total_items > 0 else 0
total_process_time = sum(item.get('process_time', 0) for item in parallel_results)
# 統計情報
values = [item.get('processed_value', 0) for item in parallel_results]
aggregation = {
'total_items': total_items,
'total_processed_value': total_value,
'average_value': avg_value,
'min_value': min(values) if values else 0,
'max_value': max(values) if values else 0,
'total_process_time': total_process_time,
'average_process_time': total_process_time / total_items if total_items > 0 else 0,
'results_sample': parallel_results[:3], # 最初の3件をサンプルとして含める
'aggregated_at': datetime.now().isoformat(),
'message': f'{total_items}件の結果を集計しました'
}
logger.info(f"結果集計完了: 総数={total_items}, 総値={total_value:.1f}")
return aggregation
class TaskWorkflowManager:
"""タスクワークフローマネージャー"""
@staticmethod
def create_chain_workflow(source_name):
"""
チェーンワークフローの作成
Args:
source_name: データソース名
Returns:
chain: Celeryチェーンオブジェクト
"""
# タスクをチェーンで連結
workflow_chain = chain(
fetch_data_task.s(source_name), # ステップ1: データ取得
process_data_task.s(), # ステップ2: データ処理
analyze_data_task.s(), # ステップ3: データ分析
save_results_task.s() # ステップ4: 結果保存
)
logger.info(f"チェーンワークフローを作成: {source_name}")
return workflow_chain
@staticmethod
def create_group_workflow(items):
"""
グループワークフローの作成
Args:
items: 並列処理するアイテムのリスト
Returns:
group: Celeryグループオブジェクト
"""
# 各アイテムを並列処理するタスクのグループを作成
task_group = group(
parallel_processing_task.s(item) for item in items
)
logger.info(f"グループワークフローを作成: {len(items)}件のアイテム")
return task_group
@staticmethod
def create_chord_workflow(items):
"""
コードワークフローの作成（グループ+コールバック）
Args:
items: 処理するアイテムのリスト
Returns:
chord: Celeryコードオブジェクト
"""
# グループ（並列処理）とコールバック（集計）を組み合わせたコード
workflow_chord = chord(
group(parallel_processing_task.s(item) for item in items), # 並列処理
aggregate_results_task.s() # 結果集計
)
logger.info(f"コードワークフローを作成: {len(items)}件のアイテム")
return workflow_chord
@staticmethod
def create_complex_workflow(sources, items_per_source=5):
"""
複合ワークフローの作成（チェーン+グループ+コード）
Args:
sources: データソースのリスト
items_per_source: ソースごとのアイテム数
Returns:
chain: 複合ワークフローチェーン
"""
workflows = []
for source in sources:
# 各ソースに対してチェーンを作成
source_chain = chain(
fetch_data_task.s(source), # データ取得
process_data_task.s(), # データ処理
analyze_data_task.s() # データ分析
)
workflows.append(source_chain)
# 各ソースの処理をグループ化（並列実行）
sources_group = group(workflows)
# すべてのソース処理が完了したら集計するコード
def aggregate_sources_results(results):
"""ソース結果を集計するコールバック関数"""
logger.info(f"ソース結果を集計: {len(results)}件の結果")
aggregation = {
'total_sources': len(results),
'sources': results,
'total_items': sum(r.get('count', 0) for r in results),
'total_value': sum(r.get('sum', 0) for r in results),
'aggregated_at': datetime.now().isoformat()
}
return aggregation
# グループとコールバックを組み合わせたコード
complex_workflow = chord(
sources_group, # 各ソースの処理（並列）
aggregate_sources_results # 結果集計（シリアル）
)
logger.info(f"複合ワークフローを作成: {len(sources)}ソース")
return complex_workflow
@staticmethod
def execute_workflow(workflow, timeout=300):
"""
ワークフローを実行
Args:
workflow: ワークフローオブジェクト
timeout: タイムアウト時間（秒）
Returns:
dict: 実行結果
"""
try:
logger.info("ワークフローを実行開始")
start_time = datetime.now()
# ワークフローを実行
result = workflow.apply_async()
# 結果を待機
workflow_result = result.get(timeout=timeout)
end_time = datetime.now()
duration = (end_time - start_time).total_seconds()
execution_result = {
'status': 'success',
'workflow_type': type(workflow).__name__,
'result': workflow_result,
'execution_time': duration,
'start_time': start_time.isoformat(),
'end_time': end_time.isoformat()
}
logger.info(f"ワークフロー実行完了: {duration:.1f}秒")
return execution_result
except Exception as e:
logger.error(f"ワークフロー実行エラー: {e}")
return {
'status': 'failed',
'error': str(e),
'failed_at': datetime.now().isoformat()
}
# テスト関数
def test_task_workflows():
"""タスクワークフローのテスト"""
print("=== タスクワークフローテスト ===")
manager = TaskWorkflowManager()
# 1. チェーンワークフローのテスト
print("\n1. チェーンワークフローテスト:")
chain_workflow = manager.create_chain_workflow("テストデータソース")
chain_result = manager.execute_workflow(chain_workflow, timeout=30)
print(f"チェーン結果ステータス: {chain_result['status']}")
if chain_result['status'] == 'success':
final_result = chain_result['result']
print(f" 最終結果ソース: {final_result.get('source', 'N/A')}")
print(f" 保存ステータス: {final_result.get('save_status', 'N/A')}")
print(f" 実行時間: {chain_result['execution_time']:.1f}秒")
# 2. グループワークフローのテスト
print("\n2. グループワークフローテスト:")
# テスト用アイテム
test_items = [
{'id': 1, 'value': 100, 'name': 'アイテム1'},
{'id': 2, 'value': 200, 'name': 'アイテム2'},
{'id': 3, 'value': 300, 'name': 'アイテム3'},
{'id': 4, 'value': 400, 'name': 'アイテム4'},
{'id': 5, 'value': 500, 'name': 'アイテム5'},
]
group_workflow = manager.create_group_workflow(test_items)
group_result = manager.execute_workflow(group_workflow, timeout=30)
print(f"グループ結果ステータス: {group_result['status']}")
if group_result['status'] == 'success':
results = group_result['result']
print(f" 処理されたアイテム数: {len(results)}")
print(f" 実行時間: {group_result['execution_time']:.1f}秒")
# 最初の2つの結果を表示
for i, result in enumerate(results[:2]):
print(f" アイテム{i+1}: ID={result.get('id')}, "
f"処理値={result.get('processed_value', 0):.1f}")
# 3. コードワークフローのテスト
print("\n3. コードワークフローテスト:")
chord_workflow = manager.create_chord_workflow(test_items)
chord_result = manager.execute_workflow(chord_workflow, timeout=30)
print(f"コード結果ステータス: {chord_result['status']}")
if chord_result['status'] == 'success':
aggregation = chord_result['result']
print(f" 集計アイテム数: {aggregation.get('total_items', 0)}")
print(f" 総処理値: {aggregation.get('total_processed_value', 0):.1f}")
print(f" 平均値: {aggregation.get('average_value', 0):.1f}")
print(f" 総処理時間: {aggregation.get('total_process_time', 0):.1f}秒")
print(f" 実行時間: {chord_result['execution_time']:.1f}秒")
# 4. 複合ワークフローのテスト
print("\n4. 複合ワークフローテスト:")
test_sources = ['ソースA', 'ソースB', 'ソースC']
complex_workflow = manager.create_complex_workflow(test_sources)
complex_result = manager.execute_workflow(complex_workflow, timeout=60)
print(f"複合結果ステータス: {complex_result['status']}")
if complex_result['status'] == 'success':
final_aggregation = complex_result['result']
print(f" 処理ソース数: {final_aggregation.get('total_sources', 0)}")
print(f" 総アイテム数: {final_aggregation.get('total_items', 0)}")
print(f" 総値: {final_aggregation.get('total_value', 0):.1f}")
print(f" 実行時間: {complex_result['execution_time']:.1f}秒")
# 5. 個別タスクのテスト
print("\n5. 個別タスクテスト:")
# fetch_data_taskのテスト
fetch_result = fetch_data_task.delay("個別テストソース").get(timeout=10)
print(f" データ取得: {fetch_result['item_count']}件取得")
# process_data_taskのテスト
process_result = process_data_task.delay(fetch_result).get(timeout=10)
print(f" データ処理: 総処理値={process_result['total_processed_value']}")
# analyze_data_taskのテスト
analyze_result = analyze_data_task.delay(process_result).get(timeout=10)
print(f" データ分析: 平均値={analyze_result.get('average', 0):.1f}")
return {
'chain_test': chain_result,
'group_test': group_result,
'chord_test': chord_result,
'complex_test': complex_result
}
# ワークフローモニタリング
def monitor_workflow_progress(task_id, interval=1, timeout=300):
"""
ワークフローの進捗状況を監視
Args:
task_id: 親タスクのID
interval: チェック間隔（秒）
timeout: タイムアウト時間（秒）
"""
from celery.result import AsyncResult
import time
print(f"\nワークフロー進捗監視: タスクID={task_id}")
print("-" * 50)
start_time = time.time()
last_update = start_time
while time.time() - start_time < timeout:
task_result = AsyncResult(task_id, app=celery)
if task_result.ready():
if task_result.successful():
print("\n✅ ワークフロー完了!")
result = task_result.result
print(f"結果タイプ: {type(result).__name__}")
if isinstance(result, list):
print(f"結果件数: {len(result)}")
for i, item in enumerate(result[:3]):
print(f" 結果{i+1}: {str(item)[:100]}...")
elif isinstance(result, dict):
print(f"結果キー: {', '.join(result.keys())}")
else:
print(f"結果: {str(result)[:200]}")
else:
print(f"\n❌ ワークフロー失敗: {task_result.result}")
break
# ステータス表示（3秒ごとに更新）
if time.time() - last_update >= 3:
status = task_result.status
print(f"ステータス: {status} - 経過時間: {time.time() - start_time:.1f}秒")
last_update = time.time()
time.sleep(interval)
if not task_result.ready():
print(f"\n⏰ タイムアウト: ワークフローは{timeout}秒以内に完了しませんでした")
if __name__ == "__main__":
# テスト実行
test_results = test_task_workflows()
# ワークフロー監視のデモ
print("\n" + "="*60)
print("ワークフロー監視デモ")
print("="*60)
# 新しいワークフローを実行して監視
manager = TaskWorkflowManager()
demo_workflow = manager.create_chain_workflow("デモソース")
demo_task = demo_workflow.apply_async()
monitor_workflow_progress(demo_task.id, interval=2, timeout=30)問題7：優先度付きタスクキュー
# app/tasks/priority_tasks.py
from app.celery_worker import celery
import time
from datetime import datetime
import logging
from celery.exceptions import Reject
import redis
logger = logging.getLogger(__name__)
class PriorityTaskQueue:
"""優先度付きタスクキュー管理クラス"""
# 優先度定義
PRIORITIES = {
'URGENT': 0, # 緊急 - 即時処理が必要
'HIGH': 1, # 高 - 優先的に処理
'NORMAL': 2, # 通常 - 標準的な優先度
'LOW': 3, # 低 - 余裕があれば処理
'BACKGROUND': 4 # バックグラウンド - システムがアイドル時に処理
}
# 優先度に対応するキューの名前
QUEUE_NAMES = {
'URGENT': 'urgent',
'HIGH': 'high',
'NORMAL': 'normal',
'LOW': 'low',
'BACKGROUND': 'background'
}
@staticmethod
def get_queue_for_priority(priority):
"""
優先度に対応するキュー名を取得
Args:
priority: 優先度文字列または数値
Returns:
str: キュー名
"""
# 文字列の優先度を数値に変換
if isinstance(priority, str):
priority = priority.upper()
if priority in PriorityTaskQueue.PRIORITIES:
priority_num = PriorityTaskQueue.PRIORITIES[priority]
else:
# 不明な優先度はNORMALとして扱う
priority_num = PriorityTaskQueue.PRIORITIES['NORMAL']
else:
priority_num = int(priority)
# 優先度に基づいてキューを選択
if priority_num <= PriorityTaskQueue.PRIORITIES['URGENT']:
return PriorityTaskQueue.QUEUE_NAMES['URGENT']
elif priority_num <= PriorityTaskQueue.PRIORITIES['HIGH']:
return PriorityTaskQueue.QUEUE_NAMES['HIGH']
elif priority_num <= PriorityTaskQueue.PRIORITIES['NORMAL']:
return PriorityTaskQueue.QUEUE_NAMES['NORMAL']
elif priority_num <= PriorityTaskQueue.PRIORITIES['LOW']:
return PriorityTaskQueue.QUEUE_NAMES['LOW']
else:
return PriorityTaskQueue.QUEUE_NAMES['BACKGROUND']
@staticmethod
def get_priority_display_name(priority):
"""優先度の表示名を取得"""
priority_map = {
0: '緊急',
1: '高',
2: '通常',
3: '低',
4: 'バックグラウンド'
}
return priority_map.get(priority, f'不明({priority})')
# 優先度付きタスクのデコレータ
def priority_task(priority='NORMAL', queue=None):
"""
優先度付きタスクデコレータ
Args:
priority: 優先度（'URGENT', 'HIGH', 'NORMAL', 'LOW', 'BACKGROUND' または数値）
queue: 明示的にキューを指定する場合（Noneなら自動決定）
"""
def decorator(func):
# キューの決定
if queue is None:
task_queue = PriorityTaskQueue.get_queue_for_priority(priority)
else:
task_queue = queue
# タスクの作成
@celery.task(
bind=True,
queue=task_queue,
name=f'priority_{func.__name__}'
)
def priority_task_wrapper(self, *args, **kwargs):
"""優先度付きタスクラッパー"""
task_id = self.request.id
task_priority = priority
# 優先度情報をログ
priority_name = PriorityTaskQueue.get_priority_display_name(
PriorityTaskQueue.PRIORITIES.get(task_priority.upper(), 2)
)
logger.info(f"優先度付きタスク開始 [{priority_name}]: {task_id}")
# タスク開始時間
start_time = datetime.now()
try:
# タスク実行
result = func(self, *args, **kwargs)
# タスク完了時間
end_time = datetime.now()
duration = (end_time - start_time).total_seconds()
# 結果に優先度情報を追加
if isinstance(result, dict):
result['task_priority'] = {
'level': task_priority,
'display_name': priority_name,
'queue': task_queue
}
result['execution_time'] = duration
result['start_time'] = start_time.isoformat()
result['end_time'] = end_time.isoformat()
logger.info(f"優先度付きタスク完了 [{priority_name}]: {task_id} - "
f"{duration:.2f}秒")
return result
except Exception as e:
logger.error(f"優先度付きタスクエラー [{priority_name}]: {task_id} - {e}")
raise
return priority_task_wrapper
return decorator
# 優先度付きタスクの例
@priority_task(priority='URGENT')
def urgent_processing_task(self, data):
"""緊急処理タスク"""
logger.info(f"緊急処理実行: {data}")
# 緊急処理のシミュレーション
time.sleep(1) # 短い処理時間
return {
'status': 'urgent_processed',
'data': data,
'processed_at': datetime.now().isoformat(),
'message': '緊急処理が完了しました'
}
@priority_task(priority='HIGH')
def high_priority_task(self, user_id, action):
"""高優先度タスク"""
logger.info(f"高優先度処理: ユーザー {user_id}, アクション {action}")
# 高優先度処理のシミュレーション
time.sleep(2)
return {
'user_id': user_id,
'action': action,
'processed': True,
'priority': 'high',
'timestamp': datetime.now().isoformat()
}
@priority_task(priority='NORMAL')
def normal_processing_task(self, items):
"""通常優先度タスク"""
logger.info(f"通常処理: {len(items)}件のアイテム")
# 通常処理のシミュレーション
time.sleep(3)
# 処理結果
processed_items = []
for i, item in enumerate(items):
processed_items.append({
'original': item,
'processed': item * 2,
'index': i
})
return {
'total_items': len(items),
'processed_items': processed_items,
'average': sum(items) / len(items) if items else 0,
'priority': 'normal'
}
@priority_task(priority='LOW')
def low_priority_task(self, data):
"""低優先度タスク"""
logger.info(f"低優先度処理: {data}")
# 低優先度処理のシミュレーション（長い処理時間）
time.sleep(5)
return {
'data': data,
'processed': True,
'priority': 'low',
'processing_time': 'long',
'timestamp': datetime.now().isoformat()
}
@priority_task(priority='BACKGROUND')
def background_task(self, task_name):
"""バックグラウンドタスク"""
logger.info(f"バックグラウンド処理: {task_name}")
# バックグラウンド処理のシミュレーション（非常に長い処理時間）
time.sleep(10)
return {
'task': task_name,
'status': 'completed',
'priority': 'background',
'message': 'バックグラウンド処理が完了しました',
'timestamp': datetime.now().isoformat()
}
class PriorityQueueManager:
"""優先度付きキューマネージャー"""
def __init__(self, redis_url=None):
self.redis_client = self._get_redis_client(redis_url)
def _get_redis_client(self, redis_url):
"""Redisクライアントを取得"""
if redis_url is None:
redis_url = celery.conf.broker_url
import redis
from urllib.parse import urlparse
parsed = urlparse(redis_url)
return redis.Redis(
host=parsed.hostname or 'localhost',
port=parsed.port or 6379,
password=parsed.password,
decode_responses=True
)
def get_queue_stats(self):
"""
各キューの統計情報を取得
Returns:
dict: キューの統計情報
"""
stats = {}
for queue_name in PriorityTaskQueue.QUEUE_NAMES.values():
# キューの長さを取得
queue_length = self.redis_client.llen(f'celery:{queue_name}')
# ワーカー数（簡易的な取得）
workers_key = f'celery:active:worker:{queue_name}'
worker_count = len(self.redis_client.smembers(workers_key)) if self.redis_client.exists(workers_key) else 0
stats[queue_name] = {
'length': queue_length,
'worker_count': worker_count,
'priority': self._get_queue_priority(queue_name)
}
return stats
def _get_queue_priority(self, queue_name):
"""キューの優先度数値を取得"""
for priority, name in PriorityTaskQueue.QUEUE_NAMES.items():
if name == queue_name:
return PriorityTaskQueue.PRIORITIES[priority]
return PriorityTaskQueue.PRIORITIES['NORMAL']
def prioritize_queue(self, source_queue, target_queue, limit=10):
"""
タスクを優先キューに移動
Args:
source_queue: 元のキュー
target_queue: 移動先キュー（より高い優先度）
limit: 移動する最大タスク数
Returns:
dict: 移動結果
"""
moved_count = 0
try:
for _ in range(limit):
# タスクをポップ（ブロッキングなし）
task_data = self.redis_client.lpop(f'celery:{source_queue}')
if not task_data:
break
# ターゲットキューにプッシュ
self.redis_client.rpush(f'celery:{target_queue}', task_data)
moved_count += 1
result = {
'status': 'success',
'moved_count': moved_count,
'source_queue': source_queue,
'target_queue': target_queue,
'message': f'{moved_count}件のタスクを移動しました'
}
logger.info(f"キュー優先化: {source_queue} -> {target_queue}, {moved_count}件移動")
except Exception as e:
result = {
'status': 'error',
'error': str(e),
'moved_count': moved_count
}
logger.error(f"キュー優先化エラー: {e}")
return result
def rebalance_queues(self):
"""
キューのリバランスを実行
優先度の高いキューが空で、優先度の低いキューにタスクがある場合に移動する
"""
rebalance_log = []
# 優先度の高い順にチェック
queues_by_priority = sorted(
PriorityTaskQueue.QUEUE_NAMES.items(),
key=lambda x: PriorityTaskQueue.PRIORITIES[x[0]]
)
for i, (priority, queue_name) in enumerate(queues_by_priority):
# 現在のキューの長さを取得
current_length = self.redis_client.llen(f'celery:{queue_name}')
# キューが空で、より優先度の低いキューにタスクがある場合
if current_length == 0 and i < len(queues_by_priority) - 1:
# 次の優先度の低いキューを探す
for j in range(i + 1, len(queues_by_priority)):
lower_priority, lower_queue = queues_by_priority[j]
lower_length = self.redis_client.llen(f'celery:{lower_queue}')
if lower_length > 0:
# タスクを移動
move_result = self.prioritize_queue(
lower_queue, queue_name, limit=min(5, lower_length)
)
if move_result['status'] == 'success':
rebalance_log.append({
'from': lower_queue,
'to': queue_name,
'count': move_result['moved_count']
})
break
return {
'status': 'completed',
'rebalance_actions': rebalance_log,
'timestamp': datetime.now().isoformat()
}
def monitor_queue_health(self):
"""
キューの健全性を監視
Returns:
dict: 健全性チェック結果
"""
health_checks = []
warnings = []
stats = self.get_queue_stats()
for queue_name, queue_stats in stats.items():
length = queue_stats['length']
worker_count = queue_stats['worker_count']
priority = queue_stats['priority']
# チェック1: ワーカーがいないキューにタスクがある
if length > 0 and worker_count == 0:
warnings.append(f"キュー '{queue_name}' にワーカーがいませんが、{length}件のタスクがあります")
# チェック2: 緊急キューに大量のタスク
if queue_name == 'urgent' and length > 100:
warnings.append(f"緊急キューに{length}件のタスクがあります。処理が遅延する可能性があります")
# チェック3: キューが詰まっている
if length > 1000:
warnings.append(f"キュー '{queue_name}' に{length}件のタスクがあります。容量超過の可能性があります")
health_checks.append({
'queue': queue_name,
'length': length,
'workers': worker_count,
'priority': priority,
'health': 'warning' if warnings else 'healthy'
})
return {
'timestamp': datetime.now().isoformat(),
'health_checks': health_checks,
'warnings': warnings,
'overall_health': 'warning' if warnings else 'healthy'
}
@celery.task
def priority_queue_maintenance():
"""優先度付きキューのメンテナンスタスク"""
manager = PriorityQueueManager()
maintenance_log = []
# 1. キューのリバランス
rebalance_result = manager.rebalance_queues()
maintenance_log.append({
'action': 'rebalance',
'result': rebalance_result
})
# 2. 健全性チェック
health_result = manager.monitor_queue_health()
maintenance_log.append({
'action': 'health_check',
'result': health_result
})
# 3. 古いタスクのクリーンアップ（簡易版）
try:
# 全てのキューの合計長を取得
total_tasks = sum(
manager.redis_client.llen(f'celery:{queue}')
for queue in PriorityTaskQueue.QUEUE_NAMES.values()
)
maintenance_log.append({
'action': 'cleanup_check',
'total_tasks': total_tasks,
'message': 'タスククリーンアップチェック完了'
})
except Exception as e:
maintenance_log.append({
'action': 'cleanup_check',
'error': str(e)
})
result = {
'status': 'completed',
'maintenance_actions': maintenance_log,
'timestamp': datetime.now().isoformat()
}
logger.info(f"優先度キューメンテナンス完了: {len(maintenance_log)}アクション実行")
return result
# テスト関数
def test_priority_tasks():
"""優先度付きタスクのテスト"""
print("=== 優先度付きタスクシステムテスト ===")
# 優先度付きタスクの実行
print("\n1. 各優先度のタスクを実行:")
tasks = []
# 緊急タスク
urgent_task = urgent_processing_task.delay({"type": "emergency", "code": "ALERT-001"})
tasks.append(('緊急', urgent_task))
# 高優先度タスク
high_task = high_priority_task.delay(user_id=123, action="password_reset")
tasks.append(('高', high_task))
# 通常タスク
normal_task = normal_processing_task.delay([1, 2, 3, 4, 5])
tasks.append(('通常', normal_task))
# 低優先度タスク
low_task = low_priority_task.delay("batch_report_2024")
tasks.append(('低', low_task))
# バックグラウンドタスク
bg_task = background_task.delay("data_archiving")
tasks.append(('バックグラウンド', bg_task))
print(f" 実行したタスク数: {len(tasks)}")
# キューマネージャーのテスト
print("\n2. キューマネージャーテスト:")
manager = PriorityQueueManager()
# キュースタッツの取得
stats = manager.get_queue_stats()
print(f" キュースタッツ:")
for queue, queue_stats in stats.items():
print(f" {queue}: {queue_stats['length']}件のタスク, "
f"{queue_stats['worker_count']}ワーカー, "
f"優先度: {queue_stats['priority']}")
# 健全性チェック
health = manager.monitor_queue_health()
print(f" 健全性チェック: {health['overall_health']}")
if health['warnings']:
print(f" 警告:")
for warning in health['warnings']:
print(f" - {warning}")
# 優先度デコレータのテスト
print("\n3. 優先度デコレータテスト:")
# 動的に優先度付きタスクを作成
@priority_task(priority='HIGH')
def dynamic_high_priority_task(self, message):
return {'message': message, 'priority': 'high'}
@priority_task(priority=PriorityTaskQueue.PRIORITIES['LOW'])
def dynamic_low_priority_task(self, data):
return {'data': data, 'priority': 'low'}
# 動的タスクを実行
dyn_high_task = dynamic_high_priority_task.delay("動的高優先度タスク")
dyn_low_task = dynamic_low_priority_task.delay({"test": "data"})
print(f" 動的タスク作成:")
print(f" - {dynamic_high_priority_task.name}: 優先度 HIGH")
print(f" - {dynamic_low_priority_task.name}: 優先度 LOW")
# メンテナンスタスクのテスト
print("\n4. メンテナンスタスクテスト:")
maintenance_task = priority_queue_maintenance.delay()
try:
maintenance_result = maintenance_task.get(timeout=30)
print(f" メンテナンス結果: {maintenance_result['status']}")
print(f" 実行アクション数: {len(maintenance_result['maintenance_actions'])}")
except Exception as e:
print(f" メンテナンスタスクエラー: {e}")
# タスク結果の収集
print("\n5. タスク結果の収集:")
for priority_name, task in tasks:
try:
result = task.get(timeout=10)
print(f" {priority_name}優先度タスク: {result.get('status', 'completed')}")
except Exception as e:
print(f" {priority_name}優先度タスクエラー: {e}")
return {
'tasks_executed': len(tasks),
'queue_stats': stats,
'health_check': health
}
# 優先度キュー監視ダッシュボード（簡易版）
def display_queue_dashboard():
"""キューダッシュボードを表示"""
import time
from rich.console import Console
from rich.table import Table
from rich.live import Live
from rich.layout import Layout
try:
from rich import print as rprint
except ImportError:
print("Richライブラリがインストールされていません。pip install rich でインストールしてください。")
return
console = Console()
manager = PriorityQueueManager()
def generate_table():
"""キュースタッツテーブルを生成"""
stats = manager.get_queue_stats()
table = Table(title="優先度付きキュー監視ダッシュボード", show_lines=True)
table.add_column("キュー名", style="cyan", no_wrap=True)
table.add_column("優先度", style="magenta")
table.add_column("待機タスク数", justify="right", style="green")
table.add_column("ワーカー数", justify="right", style="yellow")
table.add_column("状態", style="red")
total_tasks = 0
total_workers = 0
for queue_name, queue_stats in stats.items():
length = queue_stats['length']
workers = queue_stats['worker_count']
priority_num = queue_stats['priority']
total_tasks += length
total_workers += workers
# 優先度名を取得
priority_name = PriorityTaskQueue.get_priority_display_name(priority_num)
# 状態の決定
if length == 0:
status = "✅ 空"
elif length < 10:
status = "🟢 正常"
elif length < 50:
status = "🟡 注意"
elif length < 100:
status = "🟠 警告"
else:
status = "🔴 危険"
# ワーカーがいない場合は追加の警告
if workers == 0 and length > 0:
status += " ⚠️ ワーカーなし"
table.add_row(
queue_name,
priority_name,
str(length),
str(workers),
status
)
# 合計行を追加
table.add_row(
"[bold]合計[/bold]",
"",
f"[bold]{total_tasks}[/bold]",
f"[bold]{total_workers}[/bold]",
f"[bold]{len(stats)}キュー[/bold]"
)
return table
# リアルタイム更新
print("\nキューダッシュボードを起動しています... (Ctrl+Cで終了)")
try:
with Live(generate_table(), refresh_per_second=1, screen=True) as live:
while True:
time.sleep(2)
live.update(generate_table())
except KeyboardInterrupt:
console.print("\n[yellow]ダッシュボードを終了します[/yellow]")
if __name__ == "__main__":
# テスト実行
test_results = test_priority_tasks()
# ダッシュボード表示（オプション）
# display_queue_dashboard() # Richライブラリが必要問題8：定期的なメンテナンスタスク
# app/tasks/maintenance_tasks.py
from app.celery_worker import celery
from celery.schedules import crontab
import time
from datetime import datetime, timedelta
import logging
import os
import shutil
from pathlib import Path
logger = logging.getLogger(__name__)
class DatabaseMaintenance:
"""データベースメンテナンスクラス"""
def __init__(self, db_session=None):
self.db_session = db_session
def cleanup_old_logs(self, days_old=30):
"""
古いログレコードを削除
Args:
days_old: 削除対象の日数
Returns:
dict: クリーニング結果
"""
from datetime import datetime, timedelta
try:
cutoff_date = datetime.now() - timedelta(days=days_old)
# 実際の実装ではデータベースモデルを使用
# 例: old_logs = LogEntry.query.filter(LogEntry.created_at < cutoff_date).delete()
# self.db_session.commit()
# ここではモック実装
deleted_count = 42 # モック値
result = {
'operation': 'cleanup_old_logs',
'cutoff_date': cutoff_date.isoformat(),
'deleted_count': deleted_count,
'status': 'success'
}
logger.info(f"古いログをクリーニング: {deleted_count}件削除")
return result
except Exception as e:
logger.error(f"ログクリーニングエラー: {e}")
return {
'operation': 'cleanup_old_logs',
'error': str(e),
'status': 'failed'
}
def cleanup_expired_sessions(self, days_old=7):
"""
期限切れのセッションを削除
Args:
days_old: 削除対象の日数
Returns:
dict: クリーニング結果
"""
try:
cutoff_date = datetime.now() - timedelta(days=days_old)
# 実際の実装ではデータベースモデルを使用
# 例: expired_sessions = Session.query.filter(Session.expires_at < datetime.now()).delete()
# self.db_session.commit()
# ここではモック実装
deleted_count = 15 # モック値
result = {
'operation': 'cleanup_expired_sessions',
'cutoff_date': cutoff_date.isoformat(),
'deleted_count': deleted_count,
'status': 'success'
}
logger.info(f"期限切れセッションをクリーニング: {deleted_count}件削除")
return result
except Exception as e:
logger.error(f"セッションクリーニングエラー: {e}")
return {
'operation': 'cleanup_expired_sessions',
'error': str(e),
'status': 'failed'
}
def cleanup_old_password_resets(self, hours_old=24):
"""
古いパスワードリセットトークンを削除
Args:
hours_old: 削除対象の時間
Returns:
dict: クリーニング結果
"""
try:
cutoff_time = datetime.now() - timedelta(hours=hours_old)
# 実際の実装ではデータベースモデルを使用
# old_resets = PasswordReset.query.filter(PasswordReset.created_at < cutoff_time).delete()
# モック実装
deleted_count = 8 # モック値
result = {
'operation': 'cleanup_old_password_resets',
'cutoff_time': cutoff_time.isoformat(),
'deleted_count': deleted_count,
'status': 'success'
}
logger.info(f"古いパスワードリセットをクリーニング: {deleted_count}件削除")
return result
except Exception as e:
logger.error(f"パスワードリセットクリーニングエラー: {e}")
return {
'operation': 'cleanup_old_password_resets',
'error': str(e),
'status': 'failed'
}
def optimize_database_tables(self):
"""データベーステーブルの最適化"""
try:
# 実際の実装ではデータベース固有の最適化コマンドを実行
# 例: for table in ['users', 'posts', 'comments']:
# db.session.execute(f'OPTIMIZE TABLE {table}')
# モック実装
tables_optimized = ['users', 'posts', 'comments', 'sessions']
result = {
'operation': 'optimize_database_tables',
'tables_optimized': tables_optimized,
'status': 'success',
'message': f'{len(tables_optimized)}テーブルを最適化しました'
}
logger.info(f"データベーステーブルを最適化: {len(tables_optimized)}テーブル")
return result
except Exception as e:
logger.error(f"データベース最適化エラー: {e}")
return {
'operation': 'optimize_database_tables',
'error': str(e),
'status': 'failed'
}
class FileSystemMaintenance:
"""ファイルシステムメンテナンスクラス"""
def __init__(self, base_dirs=None):
self.base_dirs = base_dirs or [
'./temp',
'./uploads',
'./logs',
'./cache'
]
def cleanup_temp_files(self, days_old=1):
"""
古い一時ファイルを削除
Args:
days_old: 削除対象の日数
Returns:
dict: クリーニング結果
"""
total_deleted = 0
total_size_freed = 0
errors = []
cutoff_time = time.time() - (days_old * 24 * 3600)
for base_dir in self.base_dirs:
if not os.path.exists(base_dir):
continue
try:
for root, dirs, files in os.walk(base_dir):
for file in files:
file_path = os.path.join(root, file)
try:
# ファイルの最終変更時間をチェック
mtime = os.path.getmtime(file_path)
if mtime < cutoff_time:
# ファイルサイズを記録
file_size = os.path.getsize(file_path)
# ファイルを削除
os.unlink(file_path)
total_deleted += 1
total_size_freed += file_size
logger.debug(f"一時ファイル削除: {file_path} ({file_size} bytes)")
except Exception as e:
errors.append(f"{file_path}: {str(e)}")
except Exception as e:
errors.append(f"{base_dir}: {str(e)}")
result = {
'operation': 'cleanup_temp_files',
'total_deleted': total_deleted,
'total_size_freed_bytes': total_size_freed,
'total_size_freed_mb': total_size_freed / (1024 * 1024),
'errors': errors[:10], # 最初の10エラーのみ
'status': 'success' if total_deleted > 0 or not errors else 'partial'
}
logger.info(f"一時ファイルをクリーニング: {total_deleted}件削除, "
f"{total_size_freed / (1024 * 1024):.1f}MB 解放")
return result
def rotate_log_files(self, keep_count=10, max_size_mb=100):
"""
ログファイルのローテーション
Args:
keep_count: 保持するログファイル数
max_size_mb: 最大ファイルサイズ（MB）
Returns:
dict: ローテーション結果
"""
max_size_bytes = max_size_mb * 1024 * 1024
rotated_count = 0
log_dirs = ['./logs', '/var/log/app']
for log_dir in log_dirs:
if not os.path.exists(log_dir):
continue
try:
# ログファイルを検索
log_files = []
for file in os.listdir(log_dir):
if file.endswith('.log'):
file_path = os.path.join(log_dir, file)
if os.path.isfile(file_path):
log_files.append(file_path)
# サイズベースのローテーション
for log_file in log_files:
try:
file_size = os.path.getsize(log_file)
if file_size > max_size_bytes:
# ログファイルをローテート
base_name = os.path.basename(log_file)
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
rotated_file = os.path.join(
log_dir,
f"{base_name}.{timestamp}"
)
shutil.move(log_file, rotated_file)
# 新しい空のログファイルを作成
open(log_file, 'w').close()
rotated_count += 1
logger.info(f"ログファイルをローテート: {log_file} -> {rotated_file}")
except Exception as e:
logger.error(f"ログローテーションエラー {log_file}: {e}")
# 古いログファイルの削除（数量ベース）
log_files_with_mtime = []
for file in os.listdir(log_dir):
if file.endswith('.log') and '.' in file and file.count('.') > 1:
file_path = os.path.join(log_dir, file)
mtime = os.path.getmtime(file_path)
log_files_with_mtime.append((mtime, file_path))
# 古い順にソート
log_files_with_mtime.sort()
# 保持数を超えた古いファイルを削除
if len(log_files_with_mtime) > keep_count:
for i in range(len(log_files_with_mtime) - keep_count):
old_file = log_files_with_mtime[i][1]
try:
os.unlink(old_file)
logger.info(f"古いログファイルを削除: {old_file}")
except Exception as e:
logger.error(f"古いログファイル削除エラー {old_file}: {e}")
except Exception as e:
logger.error(f"ログディレクトリ処理エラー {log_dir}: {e}")
result = {
'operation': 'rotate_log_files',
'rotated_count': rotated_count,
'keep_count': keep_count,
'max_size_mb': max_size_mb,
'status': 'success'
}
return result
def cleanup_old_backups(self, backup_dir='./backups', days_old=30):
"""
古いバックアップファイルを削除
Args:
backup_dir: バックアップディレクトリ
days_old: 削除対象の日数
Returns:
dict: クリーニング結果
"""
if not os.path.exists(backup_dir):
return {
'operation': 'cleanup_old_backups',
'message': 'バックアップディレクトリが存在しません',
'status': 'skipped'
}
cutoff_time = time.time() - (days_old * 24 * 3600)
deleted_count = 0
errors = []
try:
for file in os.listdir(backup_dir):
if file.endswith('.bak') or file.endswith('.backup') or file.endswith('.sql'):
file_path = os.path.join(backup_dir, file)
try:
mtime = os.path.getmtime(file_path)
if mtime < cutoff_time:
file_size = os.path.getsize(file_path)
os.unlink(file_path)
deleted_count += 1
logger.info(f"古いバックアップを削除: {file_path} "
f"({file_size / (1024*1024):.1f}MB)")
except Exception as e:
errors.append(f"{file_path}: {str(e)}")
except Exception as e:
errors.append(f"バックアップディレクトリスキャンエラー: {str(e)}")
result = {
'operation': 'cleanup_old_backups',
'backup_dir': backup_dir,
'days_old': days_old,
'deleted_count': deleted_count,
'errors': errors[:5],
'status': 'success' if deleted_count > 0 or not errors else 'partial'
}
logger.info(f"古いバックアップをクリーニング: {deleted_count}件削除")
return result
class SystemMaintenance:
"""システムメンテナンスクラス"""
def __init__(self):
self.db_maintenance = DatabaseMaintenance()
self.fs_maintenance = FileSystemMaintenance()
def run_daily_maintenance(self):
"""日次メンテナンスを実行"""
start_time = datetime.now()
operations = []
logger.info("日次メンテナンスを開始")
try:
# 1. データベースメンテナンス
logger.info("データベースメンテナンスを実行中...")
# 古いログの削除（30日以上前）
log_cleanup = self.db_maintenance.cleanup_old_logs(days_old=30)
operations.append(log_cleanup)
# 期限切れセッションの削除（7日以上前）
session_cleanup = self.db_maintenance.cleanup_expired_sessions(days_old=7)
operations.append(session_cleanup)
# 古いパスワードリセットの削除（24時間以上前）
reset_cleanup = self.db_maintenance.cleanup_old_password_resets(hours_old=24)
operations.append(reset_cleanup)
# データベーステーブルの最適化
db_optimize = self.db_maintenance.optimize_database_tables()
operations.append(db_optimize)
# 2. ファイルシステムメンテナンス
logger.info("ファイルシステムメンテナンスを実行中...")
# 一時ファイルの削除（1日以上前）
temp_cleanup = self.fs_maintenance.cleanup_temp_files(days_old=1)
operations.append(temp_cleanup)
# ログファイルのローテーション
log_rotation = self.fs_maintenance.rotate_log_files(keep_count=10, max_size_mb=100)
operations.append(log_rotation)
# 古いバックアップの削除（30日以上前）
backup_cleanup = self.fs_maintenance.cleanup_old_backups(days_old=30)
operations.append(backup_cleanup)
# 3. システムチェック
logger.info("システムチェックを実行中...")
# ディスク使用量チェック
disk_check = self.check_disk_usage()
operations.append(disk_check)
# メモリ使用量チェック
memory_check = self.check_memory_usage()
operations.append(memory_check)
# 成功した操作数をカウント
successful_ops = sum(1 for op in operations if op.get('status') == 'success')
total_ops = len(operations)
end_time = datetime.now()
duration = (end_time - start_time).total_seconds()
result = {
'maintenance_type': 'daily',
'start_time': start_time.isoformat(),
'end_time': end_time.isoformat(),
'duration_seconds': duration,
'total_operations': total_ops,
'successful_operations': successful_ops,
'success_rate': (successful_ops / total_ops * 100) if total_ops > 0 else 0,
'operations': operations,
'status': 'completed'
}
logger.info(f"日次メンテナンス完了: {successful_ops}/{total_ops}操作成功, "
f"{duration:.1f}秒")
return result
except Exception as e:
logger.error(f"日次メンテナンスエラー: {e}")
end_time = datetime.now()
return {
'maintenance_type': 'daily',
'start_time': start_time.isoformat(),
'end_time': end_time.isoformat(),
'error': str(e),
'operations_completed': len(operations),
'operations': operations,
'status': 'failed'
}
def check_disk_usage(self, threshold_percent=80):
"""
ディスク使用量をチェック
Args:
threshold_percent: 警告閾値（%）
Returns:
dict: ディスク使用量チェック結果
"""
try:
import shutil
total, used, free = shutil.disk_usage('/')
usage_percent = (used / total) * 100
result = {
'operation': 'check_disk_usage',
'total_gb': total / (1024**3),
'used_gb': used / (1024**3),
'free_gb': free / (1024**3),
'usage_percent': usage_percent,
'threshold_percent': threshold_percent,
'status': 'warning' if usage_percent > threshold_percent else 'ok'
}
if usage_percent > threshold_percent:
logger.warning(f"ディスク使用量が高い: {usage_percent:.1f}%")
return result
except Exception as e:
logger.error(f"ディスク使用量チェックエラー: {e}")
return {
'operation': 'check_disk_usage',
'error': str(e),
'status': 'failed'
}
def check_memory_usage(self, threshold_percent=80):
"""
メモリ使用量をチェック
Args:
threshold_percent: 警告閾値（%）
Returns:
dict: メモリ使用量チェック結果
"""
try:
import psutil
memory = psutil.virtual_memory()
result = {
'operation': 'check_memory_usage',
'total_gb': memory.total / (1024**3),
'available_gb': memory.available / (1024**3),
'used_gb': memory.used / (1024**3),
'usage_percent': memory.percent,
'threshold_percent': threshold_percent,
'status': 'warning' if memory.percent > threshold_percent else 'ok'
}
if memory.percent > threshold_percent:
logger.warning(f"メモリ使用量が高い: {memory.percent:.1f}%")
return result
except Exception as e:
logger.error(f"メモリ使用量チェックエラー: {e}")
return {
'operation': 'check_memory_usage',
'error': str(e),
'status': 'failed'
}
def run_weekly_maintenance(self):
"""週次メンテナンスを実行"""
start_time = datetime.now()
logger.info("週次メンテナンスを開始")
try:
# 日次メンテナンスを実行
daily_result = self.run_daily_maintenance()
# 追加の週次タスク
weekly_operations = []
# 古いユーザーアクティビティログの削除（90日以上前）
# 実際の実装ではデータベース操作
weekly_operations.append({
'operation': 'cleanup_old_user_activity',
'days_old': 90,
'status': 'success',
'message': '古いユーザーアクティビティを削除しました'
})
# 統計情報の更新
weekly_operations.append({
'operation': 'update_weekly_statistics',
'status': 'success',
'message': '週次統計情報を更新しました'
})
end_time = datetime.now()
duration = (end_time - start_time).total_seconds()
result = {
'maintenance_type': 'weekly',
'start_time': start_time.isoformat(),
'end_time': end_time.isoformat(),
'duration_seconds': duration,
'daily_maintenance': daily_result,
'weekly_operations': weekly_operations,
'status': 'completed'
}
logger.info(f"週次メンテナンス完了: {duration:.1f}秒")
return result
except Exception as e:
logger.error(f"週次メンテナンスエラー: {e}")
end_time = datetime.now()
return {
'maintenance_type': 'weekly',
'start_time': start_time.isoformat(),
'end_time': end_time.isoformat(),
'error': str(e),
'status': 'failed'
}
# Celeryタスク
@celery.task
def daily_maintenance_task():
"""日次メンテナンスタスク"""
logger.info("日次メンテナンスタスクを実行")
maintenance = SystemMaintenance()
result = maintenance.run_daily_maintenance()
return result
@celery.task
def weekly_maintenance_task():
"""週次メンテナンスタスク"""
logger.info("週次メンテナンスタスクを実行")
maintenance = SystemMaintenance()
result = maintenance.run_weekly_maintenance()
return result
@celery.task
def monthly_maintenance_task():
"""月次メンテナンスタスク"""
logger.info("月次メンテナンスタスクを実行")
start_time = datetime.now()
try:
maintenance = SystemMaintenance()
# 月次レポートの生成
monthly_report = {
'report_type': 'monthly',
'generated_at': datetime.now().isoformat(),
'system_checks': [
maintenance.check_disk_usage(),
maintenance.check_memory_usage()
],
'message': '月次メンテナンスレポートを生成しました'
}
end_time = datetime.now()
duration = (end_time - start_time).total_seconds()
result = {
'maintenance_type': 'monthly',
'start_time': start_time.isoformat(),
'end_time': end_time.isoformat(),
'duration_seconds': duration,
'report': monthly_report,
'status': 'completed'
}
logger.info(f"月次メンテナンス完了: {duration:.1f}秒")
return result
except Exception as e:
logger.error(f"月次メンテナンスエラー: {e}")
end_time = datetime.now()
return {
'maintenance_type': 'monthly',
'start_time': start_time.isoformat(),
'end_time': end_time.isoformat(),
'error': str(e),
'status': 'failed'
}
# Celery Beatスケジュール設定
# 注: これは設定ファイルに追加する必要があります
CELERY_BEAT_SCHEDULE = {
# 毎日午前3時に日次メンテナンスを実行
'daily-maintenance-3am': {
'task': 'app.tasks.maintenance_tasks.daily_maintenance_task',
'schedule': crontab(hour=3, minute=0),
'args': (),
'options': {'queue': 'maintenance'}
},
# 毎週月曜日午前4時に週次メンテナンスを実行
'weekly-maintenance-monday-4am': {
'task': 'app.tasks.maintenance_tasks.weekly_maintenance_task',
'schedule': crontab(day_of_week=1, hour=4, minute=0), # 1 = 月曜日
'args': (),
'options': {'queue': 'maintenance'}
},
# 毎月1日午前5時に月次メンテナンスを実行
'monthly-maintenance-1st-5am': {
'task': 'app.tasks.maintenance_tasks.monthly_maintenance_task',
'schedule': crontab(day_of_month=1, hour=5, minute=0),
'args': (),
'options': {'queue': 'maintenance'}
},
# 毎時0分にシステムチェックを実行
'hourly-system-check': {
'task': 'app.tasks.maintenance_tasks.system_health_check_task',
'schedule': crontab(minute=0),
'args': (),
'options': {'queue': 'monitoring'}
},
}
@celery.task
def system_health_check_task():
"""システムヘルスチェックタスク（毎時実行）"""
logger.info("システムヘルスチェックを実行")
start_time = datetime.now()
try:
maintenance = SystemMaintenance()
health_checks = [
maintenance.check_disk_usage(threshold_percent=85),
maintenance.check_memory_usage(threshold_percent=85)
]
# 警告があるかチェック
warnings = [check for check in health_checks if check.get('status') == 'warning']
result = {
'check_type': 'hourly_health_check',
'timestamp': datetime.now().isoformat(),
'health_checks': health_checks,
'warning_count': len(warnings),
'warnings': warnings,
'overall_status': 'warning' if warnings else 'healthy'
}
if warnings:
logger.warning(f"システムヘルスチェックで警告: {len(warnings)}件")
return result
except Exception as e:
logger.error(f"システムヘルスチェックエラー: {e}")
return {
'check_type': 'hourly_health_check',
'timestamp': datetime.now().isoformat(),
'error': str(e),
'status': 'failed'
}
# メンテナンスタスク管理
class MaintenanceTaskManager:
"""メンテナンスタスク管理クラス"""
@staticmethod
def run_maintenance_now(maintenance_type='daily'):
"""
即座にメンテナンスを実行
Args:
maintenance_type: メンテナンスタイプ ('daily', 'weekly', 'monthly')
Returns:
dict: 実行結果
"""
task_map = {
'daily': daily_maintenance_task,
'weekly': weekly_maintenance_task,
'monthly': monthly_maintenance_task
}
if maintenance_type not in task_map:
return {
'status': 'error',
'error': f"未知のメンテナンスタイプ: {maintenance_type}",
'timestamp': datetime.now().isoformat()
}
logger.info(f"即時メンテナンスを実行: {maintenance_type}")
try:
task = task_map[maintenance_type].delay()
result = task.get(timeout=300) # 5分タイムアウト
return {
'status': 'success',
'maintenance_type': maintenance_type,
'task_id': task.id,
'result': result,
'timestamp': datetime.now().isoformat()
}
except Exception as e:
logger.error(f"即時メンテナンス実行エラー: {e}")
return {
'status': 'error',
'maintenance_type': maintenance_type,
'error': str(e),
'timestamp': datetime.now().isoformat()
}
@staticmethod
def get_maintenance_schedule():
"""
メンテナンススケジュールを取得
Returns:
dict: スケジュール情報
"""
schedule = {}
# Celery Beatのスケジュールからメンテナンス関連を抽出
for task_name, task_config in CELERY_BEAT_SCHEDULE.items():
if 'maintenance' in task_name or 'check' in task_name:
schedule[task_name] = {
'task': task_config['task'],
'schedule': str(task_config['schedule']),
'queue': task_config.get('options', {}).get('queue', 'default')
}
return {
'schedule': schedule,
'total_tasks': len(schedule),
'timestamp': datetime.now().isoformat()
}
# テスト関数
def test_maintenance_tasks():
"""メンテナンスタスクのテスト"""
print("=== メンテナンスタスクシステムテスト ===")
# システムメンテナンスのテスト
print("\n1. システムメンテナンステスト:")
maintenance = SystemMaintenance()
# ディスク使用量チェック
disk_check = maintenance.check_disk_usage()
print(f" ディスク使用量: {disk_check.get('usage_percent', 0):.1f}%")
print(f" 状態: {disk_check.get('status', 'unknown')}")
# メモリ使用量チェック
memory_check = maintenance.check_memory_usage()
print(f" メモリ使用量: {memory_check.get('usage_percent', 0):.1f}%")
print(f" 状態: {memory_check.get('status', 'unknown')}")
# ファイルシステムメンテナンスのテスト
print("\n2. ファイルシステムメンテナンステスト:")
fs_maintenance = FileSystemMaintenance(['./test_temp'])
# テスト用一時ディレクトリの作成
test_dir = './test_temp'
os.makedirs(test_dir, exist_ok=True)
# テストファイルの作成
test_file = os.path.join(test_dir, 'test_file.txt')
with open(test_file, 'w') as f:
f.write('テストファイル')
# 一時ファイルクリーニング（実際には削除されないように日数を調整）
temp_cleanup = fs_maintenance.cleanup_temp_files(days_old=0) # 0日以上前を削除
print(f" 一時ファイル削除: {temp_cleanup.get('total_deleted', 0)}件")
# データベースメンテナンスのテスト（モック）
print("\n3. データベースメンテナンステスト（モック）:")
db_maintenance = DatabaseMaintenance()
log_cleanup = db_maintenance.cleanup_old_logs(days_old=30)
print(f" 古いログ削除: {log_cleanup.get('deleted_count', 0)}件")
session_cleanup = db_maintenance.cleanup_expired_sessions(days_old=7)
print(f" 期限切れセッション削除: {session_cleanup.get('deleted_count', 0)}件")
# メンテナンスタスクのテスト
print("\n4. メンテナンスタスクテスト:")
manager = MaintenanceTaskManager()
# スケジュール取得
schedule = manager.get_maintenance_schedule()
print(f" 登録されているメンテナンスタスク: {schedule.get('total_tasks', 0)}件")
for task_name, task_config in schedule.get('schedule', {}).items():
print(f" - {task_name}: {task_config['schedule']}")
# 即時メンテナンス実行（軽量なテスト）
print("\n5. 即時メンテナンス実行テスト:")
# システムヘルスチェックのみ実行（軽量）
health_task = system_health_check_task.delay()
try:
health_result = health_task.get(timeout=10)
print(f" ヘルスチェック結果: {health_result.get('overall_status', 'unknown')}")
print(f" 警告数: {health_result.get('warning_count', 0)}")
except Exception as e:
print(f" ヘルスチェックエラー: {e}")
# クリーンアップ
if os.path.exists(test_dir):
shutil.rmtree(test_dir)
print(f"\nテストディレクトリを削除: {test_dir}")
return {
'disk_check': disk_check,
'memory_check': memory_check,
'temp_cleanup': temp_cleanup,
'maintenance_schedule': schedule
}
# メンテナンスダッシュボード（簡易版）
def display_maintenance_dashboard():
"""メンテナンスダッシュボードを表示"""
import time
from datetime import datetime
print("\n" + "="*60)
print("メンテナンスダッシュボード")
print("="*60)
manager = MaintenanceTaskManager()
maintenance = SystemMaintenance()
while True:
# 画面をクリア（プラットフォームに依存）
import os
os.system('cls' if os.name == 'nt' else 'clear')
current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
print(f"\n現在時刻: {current_time}")
print("-"*60)
# システム状態を表示
print("\n[システム状態]")
disk_check = maintenance.check_disk_usage()
memory_check = maintenance.check_memory_usage()
print(f" 💾 ディスク使用量: {disk_check.get('usage_percent', 0):.1f}% "
f"({disk_check.get('status', 'unknown')})")
print(f" 🧠 メモリ使用量: {memory_check.get('usage_percent', 0):.1f}% "
f"({memory_check.get('status', 'unknown')})")
# メンテナンススケジュールを表示
print("\n[メンテナンススケジュール]")
schedule = manager.get_maintenance_schedule()
for task_name, task_config in schedule.get('schedule', {}).items():
print(f" ⏰ {task_name}:")
print(f" タスク: {task_config['task']}")
print(f" スケジュール: {task_config['schedule']}")
print(f" キュー: {task_config['queue']}")
# メニューを表示
print("\n" + "-"*60)
print("[操作メニュー]")
print(" 1. 日次メンテナンスを今すぐ実行")
print(" 2. システムヘルスチェックを実行")
print(" 3. スケジュールを更新")
print(" 4. 終了")
print("-"*60)
choice = input("選択してください (1-4): ").strip()
if choice == '1':
print("\n日次メンテナンスを実行中...")
result = manager.run_maintenance_now('daily')
print(f"結果: {result.get('status', 'unknown')}")
input("\nEnterキーを押して続行...")
elif choice == '2':
print("\nシステムヘルスチェックを実行中...")
health_task = system_health_check_task.delay()
try:
result = health_task.get(timeout=10)
print(f"結果: {result.get('overall_status', 'unknown')}")
print(f"警告: {result.get('warning_count', 0)}件")
except Exception as e:
print(f"エラー: {e}")
input("\nEnterキーを押して続行...")
elif choice == '3':
print("\nスケジュールを更新しました")
time.sleep(1)
elif choice == '4':
print("\nダッシュボードを終了します")
break
else:
print("\n無効な選択です")
time.sleep(1)
if __name__ == "__main__":
# テスト実行
test_results = test_maintenance_tasks()
# ダッシュボード表示（オプション）
# display_maintenance_dashboard()問題9：タスク結果の永続化
<pre><code class="language-python"># app/tasks/task_persistence.py
from app.celery_worker import celery
from celery.result import AsyncResult
from datetime import datetime, timedelta
import json
import logging
from pathlib import Path
import sqlite3
import pickle
import hashlib
logger = logging.getLogger(__name__)
class TaskResultDatabase:
"""タスク結果データベース"""
def __init__(self, db_path='./task_results.db'):
self.db_path = db_path
self.init_database()
def init_database(self):
"""データベースを初期化"""
try:
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
# タスク結果テーブル
cursor.execute('''
CREATE TABLE IF NOT EXISTS task_results (
id INTEGER PRIMARY KEY AUTOINCREMENT,
task_id TEXT UNIQUE NOT NULL,
task_name TEXT,
status TEXT NOT NULL,
result_data BLOB,
error_message TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
started_at TIMESTAMP,
completed_at TIMESTAMP,
execution_time_seconds REAL,
worker_id TEXT,
queue_name TEXT,
retry_count INTEGER DEFAULT 0,
priority INTEGER DEFAULT 2,
metadata TEXT
)
''')
# インデックスの作成
cursor.execute('CREATE INDEX IF NOT EXISTS idx_task_id ON task_results(task_id)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_status ON task_results(status)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_created_at ON task_results(created_at)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_completed_at ON task_results(completed_at)')
# タスク依存関係テーブル
cursor.execute('''
CREATE TABLE IF NOT EXISTS task_dependencies (
id INTEGER PRIMARY KEY AUTOINCREMENT,
parent_task_id TEXT NOT NULL,
child_task_id TEXT NOT NULL,
dependency_type TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (parent_task_id) REFERENCES task_results(task_id),
FOREIGN KEY (child_task_id) REFERENCES task_results(task_id)
)
''')
# タスクメトリクステーブル
cursor.execute('''
CREATE TABLE IF NOT EXISTS task_metrics (
id INTEGER PRIMARY KEY AUTOINCREMENT,
task_id TEXT NOT NULL,
metric_name TEXT NOT NULL,
metric_value REAL,
metric_text TEXT,
recorded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (task_id) REFERENCES task_results(task_id)
)
''')
conn.commit()
conn.close()
logger.info(f"タスク結果データベースを初期化: {self.db_path}")
except Exception as e:
logger.error(f"データベース初期化エラー: {e}")
raise
def save_task_result(self, task_id, task_name, status, result=None,
error=None, started_at=None, completed_at=None,
worker_id=None, queue_name=None, retry_count=0,
priority=2, metadata=None):
"""
タスク結果を保存
Args:
task_id: タスクID
task_name: タスク名
status: ステータス
result: タスク結果（シリアライズ可能なオブジェクト）
error: エラーメッセージ
started_at: 開始時刻
completed_at: 完了時刻
worker_id: ワーカーID
queue_name: キュー名
retry_count: リトライ回数
priority: 優先度
metadata: 追加メタデータ
Returns:
bool: 保存成功かどうか
"""
try:
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
# 結果データのシリアライズ
result_blob = None
if result is not None:
try:
result_blob = pickle.dumps(result)
except Exception as e:
logger.warning(f"結果シリアライズエラー: {e}")
result_blob = pickle.dumps({'serialization_error': str(e)})
# メタデータのシリアライズ
metadata_json = None
if metadata is not None:
try:
metadata_json = json.dumps(metadata, ensure_ascii=False)
except:
metadata_json = json.dumps({'error': 'metadata_serialization_failed'})
# 実行時間の計算
execution_time = None
if started_at and completed_at:
if isinstance(started_at, str):
started_at = datetime.fromisoformat(started_at.replace('Z', '+00:00'))
if isinstance(completed_at, str):
completed_at = datetime.fromisoformat(completed_at.replace('Z', '+00:00'))
execution_time = (completed_at - started_at).total_seconds()
# タスク結果を保存または更新
cursor.execute('''
INSERT OR REPLACE INTO task_results
(task_id, task_name, status, result_data, error_message,
started_at, completed_at, execution_time_seconds,
worker_id, queue_name, retry_count, priority, metadata)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
''', (
task_id, task_name, status, result_blob, error,
started_at, completed_at, execution_time,
worker_id, queue_name, retry_count, priority, metadata_json
))
conn.commit()
conn.close()
logger.debug(f"タスク結果を保存: {task_id} - {status}")
return True
except Exception as e:
logger.error(f"タスク結果保存エラー {task_id}: {e}")
return False
def get_task_result(self, task_id):
"""
タスク結果を取得
Args:
task_id: タスクID
Returns:
dict: タスク結果情報
"""
try:
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute('''
SELECT * FROM task_results WHERE task_id = ?
''', (task_id,))
row = cursor.fetchone()
conn.close()
if not row:
return None
# 行を辞書に変換
result = dict(row)
# 結果データのデシリアライズ
if result['result_data']:
try:
result['result'] = pickle.loads(result['result_data'])
except Exception as e:
result['result'] = {'deserialization_error': str(e)}
# バイナリデータは削除（メモリ節約）
del result['result_data']
# メタデータのデシリアライズ
if result['metadata']:
try:
result['metadata'] = json.loads(result['metadata'])
except:
result['metadata'] = {'error': 'metadata_deserialization_failed'}
return result
except Exception as e:
logger.error(f"タスク結果取得エラー {task_id}: {e}")
return None
def search_tasks(self, filters=None, limit=100, offset=0):
"""
タスクを検索
Args:
filters: 検索フィルタ（辞書）
limit: 取得件数
offset: オフセット
Returns:
list: タスク結果のリスト
"""
try:
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
# クエリの構築
query = 'SELECT * FROM task_results'
params = []
if filters:
conditions = []
for key, value in filters.items():
if key == 'status':
conditions.append('status = ?')
params.append(value)
elif key == 'task_name':
conditions.append('task_name LIKE ?')
params.append(f'%{value}%')
elif key == 'queue_name':
conditions.append('queue_name = ?')
params.append(value)
elif key == 'created_after':
conditions.append('created_at >= ?')
params.append(value)
elif key == 'created_before':
conditions.append('created_at <= ?')
params.append(value)
elif key == 'priority_min':
conditions.append('priority <= ?')
params.append(value)
elif key == 'priority_max':
conditions.append('priority >= ?')
params.append(value)
if conditions:
query += ' WHERE ' + ' AND '.join(conditions)
# ソートと制限
query += ' ORDER BY created_at DESC LIMIT ? OFFSET ?'
params.extend([limit, offset])
cursor.execute(query, params)
rows = cursor.fetchall()
conn.close()
# 結果を処理
tasks = []
for row in rows:
task = dict(row)
# 結果データのデシリアライズ（簡易版）
if task.get('result_data'):
try:
task['result'] = pickle.loads(task['result_data'])
except:
task['result'] = None
del task['result_data']
tasks.append(task)
return tasks
except Exception as e:
logger.error(f"タスク検索エラー: {e}")
return []
def get_task_statistics(self, days=7):
"""
タスク統計を取得
Args:
days: 統計を取得する日数
Returns:
dict: 統計情報
"""
try:
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cutoff_date = (datetime.now() - timedelta(days=days)).isoformat()
statistics = {}
# 総タスク数
cursor.execute('''
SELECT COUNT(*) FROM task_results
WHERE created_at >= ?
''', (cutoff_date,))
statistics['total_tasks'] = cursor.fetchone()[0]
# ステータス別のタスク数
cursor.execute('''
SELECT status, COUNT(*)
FROM task_results
WHERE created_at >= ?
GROUP BY status
''', (cutoff_date,))
statistics['by_status'] = dict(cursor.fetchall())
# キュータスク数
cursor.execute('''
SELECT queue_name, COUNT(*)
FROM task_results
WHERE created_at >= ? AND queue_name IS NOT NULL
GROUP BY queue_name
''', (cutoff_date,))
statistics['by_queue'] = dict(cursor.fetchall())
# 平均実行時間
cursor.execute('''
SELECT
AVG(execution_time_seconds) as avg_time,
MIN(execution_time_seconds) as min_time,
MAX(execution_time_seconds) as max_time
FROM task_results
WHERE created_at >= ? AND execution_time_seconds IS NOT NULL
''', (cutoff_date,))
avg_result = cursor.fetchone()
statistics['execution_time'] = {
'avg': avg_result[0] or 0,
'min': avg_result[1] or 0,
'max': avg_result[2] or 0
}
# 日別タスク数
cursor.execute('''
SELECT DATE(created_at), COUNT(*)
FROM task_results
WHERE created_at >= ?
GROUP BY DATE(created_at)
ORDER BY DATE(created_at) DESC
LIMIT 30
''', (cutoff_date,))
statistics['daily_counts'] = dict(cursor.fetchall())
conn.close()
return statistics
except Exception as e:
logger.error(f"タスク統計取得エラー: {e}")
return {}
def cleanup_old_results(self, days_old=30):
"""
古いタスク結果を削除
Args:
days_old: 削除対象の日数
Returns:
dict: クリーンアップ結果
"""
try:
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cutoff_date = (datetime.now() - timedelta(days=days_old)).isoformat()
# 削除前のカウント
cursor.execute('SELECT COUNT(*) FROM task_results WHERE created_at < ?', (cutoff_date,))
old_count = cursor.fetchone()[0]
# 古い結果を削除
cursor.execute('DELETE FROM task_results WHERE created_at < ?', (cutoff_date,))
# 依存関係テーブルもクリーンアップ
cursor.execute('''
DELETE FROM task_dependencies
WHERE parent_task_id NOT IN (SELECT task_id FROM task_results)
OR child_task_id NOT IN (SELECT task_id FROM task_results)
''')
# メトリクステーブルもクリーンアップ
cursor.execute('''
DELETE FROM task_metrics
WHERE task_id NOT IN (SELECT task_id FROM task_results)
''')
deleted_count = cursor.rowcount
conn.commit()
conn.close()
result = {
'operation': 'cleanup_old_results',
'cutoff_date': cutoff_date,
'old_records_count': old_count,
'deleted_records': deleted_count,
'status': 'success'
}
logger.info(f"古いタスク結果を削除: {deleted_count}件")
return result
except Exception as e:
logger.error(f"古いタスク結果削除エラー: {e}")
return {
'operation': 'cleanup_old_results',
'error': str(e),
'status': 'failed'
}
# タスク結果永続化デコレータ
def persist_task_result(db_path=None):
"""
タスク結果を自動的に永続化するデコレータ
Args:
db_path: データベースパス
"""
def decorator(func):
# データベースの初期化
db = TaskResultDatabase(db_path) if db_path else TaskResultDatabase()
@celery.task(bind=True, name=f'persisted_{func.__name__}')
def persisted_task_wrapper(self, *args, **kwargs):
"""永続化タスクラッパー"""
task_id = self.request.id
task_name = self.name
logger.info(f"永続化タスク開始: {task_id} - {task_name}")
# タスク開始情報を保存
task_info = {
'task_id': task_id,
'task_name': task_name,
'status': 'STARTED',
'started_at': datetime.now().isoformat(),
'worker_id': self.request.hostname,
'queue_name': self.request.delivery_info.get('routing_key') if hasattr(self.request, 'delivery_info') else None,
'args': str(args)[:500], # 最初の500文字のみ
'kwargs': str(kwargs)[:500],
'retry_count': self.request.retries if hasattr(self.request, 'retries') else 0,
'priority': getattr(self, 'priority', 2)
}
db.save_task_result(**task_info)
try:
# タスクを実行
result = func(self, *args, **kwargs)
# 成功情報を保存
task_info.update({
'status': 'SUCCESS',
'completed_at': datetime.now().isoformat(),
'result': result
})
db.save_task_result(**task_info)
logger.info(f"永続化タスク成功: {task_id}")
return result
except Exception as e:
# 失敗情報を保存
task_info.update({
'status': 'FAILURE',
'completed_at': datetime.now().isoformat(),
'error': str(e)
})
db.save_task_result(**task_info)
logger.error(f"永続化タスク失敗: {task_id} - {e}")
raise
return persisted_task_wrapper
return decorator
# 永続化タスクの例
@persist_task_result()
def example_persisted_task(self, data):
"""永続化タスクの例"""
logger.info(f"永続化タスク実行: {data}")
# タスク処理のシミュレーション
import time
time.sleep(2)
result = {
'input_data': data,
'processed': True,
'timestamp': datetime.now().isoformat(),
'checksum': hashlib.md5(str(data).encode()).hexdigest()[:8]
}
return result
@persist_task_result('./special_tasks.db')
def special_persisted_task(self, user_id, action):
"""特別なデータベースに保存する永続化タスク"""
logger.info(f"特別永続化タスク: ユーザー {user_id}, アクション {action}")
import time
time.sleep(1)
return {
'user_id': user_id,
'action': action,
'processed_at': datetime.now().isoformat(),
'status': 'completed'
}
class TaskResultMonitor:
"""タスク結果モニター"""
def __init__(self, db_path='./task_results.db'):
self.db = TaskResultDatabase(db_path)
def get_recent_tasks(self, limit=20):
"""最近のタスクを取得"""
return self.db.search_tasks(limit=limit)
def get_task_timeline(self, hours=24):
"""タスクタイムラインを取得"""
try:
conn = sqlite3.connect(self.db.db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cutoff_time = (datetime.now() - timedelta(hours=hours)).isoformat()
cursor.execute('''
SELECT
task_id,
task_name,
status,
created_at,
started_at,
completed_at,
execution_time_seconds,
queue_name
FROM task_results
WHERE created_at >= ?
ORDER BY created_at DESC
''', (cutoff_time,))
rows = cursor.fetchall()
conn.close()
timeline = []
for row in rows:
timeline.append(dict(row))
return timeline
except Exception as e:
logger.error(f"タスクタイムライン取得エラー: {e}")
return []
def generate_task_report(self, days=7, output_format='json'):
"""
タスクレポートを生成
Args:
days: レポート期間（日）
output_format: 出力形式 ('json', 'html', 'text')
Returns:
str: レポート
"""
statistics = self.db.get_task_statistics(days=days)
# 最近のタスクを取得
recent_tasks = self.get_recent_tasks(limit=10)
# 成功率の計算
total = statistics.get('total_tasks', 0)
success = statistics.get('by_status', {}).get('SUCCESS', 0)
failure = statistics.get('by_status', {}).get('FAILURE', 0)
success_rate = (success / total * 100) if total > 0 else 0
# レポートデータの構築
report_data = {
'report_period_days': days,
'generated_at': datetime.now().isoformat(),
'summary': {
'total_tasks': total,
'successful_tasks': success,
'failed_tasks': failure,
'success_rate': success_rate,
'avg_execution_time': statistics.get('execution_time', {}).get('avg', 0)
},
'statistics': statistics,
'recent_tasks': recent_tasks[:5] # 最初の5件のみ
}
if output_format == 'json':
return json.dumps(report_data, indent=2, ensure_ascii=False)
elif output_format == 'text':
lines = []
lines.append(f"タスクレポート ({days}日間)")
lines.append("="*50)
lines.append(f"生成日時: {report_data['generated_at']}")
lines.append(f"総タスク数: {total}")
lines.append(f"成功タスク: {success}")
lines.append(f"失敗タスク: {failure}")
lines.append(f"成功率: {success_rate:.1f}%")
lines.append(f"平均実行時間: {report_data['summary']['avg_execution_time']:.2f}秒")
lines.append("")
lines.append("最近のタスク:")
for task in report_data['recent_tasks']:
lines.append(f" - {task.get('task_name')}: {task.get('status')} "
f"({task.get('created_at')})")
return "\n".join(lines)
elif output_format == 'html':
html = f'''
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<title>タスクレポート</title>
<style>
body {{ font-family: Arial, sans-serif; margin: 20px; }}
.report-header {{ background-color: #f0f0f0; padding: 20px; border-radius: 5px; }}
.stats-grid {{ display: grid; grid-template-columns: repeat(auto-fit, minmax(200px, 1fr)); gap: 20px; margin: 20px 0; }}
.stat-card {{ background: white; border: 1px solid #ddd; border-radius: 5px; padding: 15px; }}
.stat-value {{ font-size: 24px; font-weight: bold; }}
.stat-label {{ color: #666; }}
.success {{ color: #28a745; }}
.failure {{ color: #dc3545; }}
table {{ width: 100%; border-collapse: collapse; margin: 20px 0; }}
th, td {{ border: 1px solid #ddd; padding: 8px; text-align: left; }}
th {{ background-color: #f8f9fa; }}
</style>
</head>
<body>
<div class="report-header">
<h1>タスクレポート</h1>
<p>期間: {days}日間 | 生成日時: {report_data['generated_at']}</p>
</div>
<div class="stats-grid">
<div class="stat-card">
<div class="stat-value">{total}</div>
<div class="stat-label">総タスク数</div>
</div>
<div class="stat-card">
<div class="stat-value success">{success}</div>
<div class="stat-label">成功タスク</div>
</div>
<div class="stat-card">
<div class="stat-value failure">{failure}</div>
<div class="stat-label">失敗タスク</div>
</div>
<div class="stat-card">
<div class="stat-value">{success_rate:.1f}%</div>
<div class="stat-label">成功率</div>
</div>
</div>
<h2>最近のタスク</h2>
<table>
<thead>
<tr>
<th>タスクID</th>
<th>タスク名</th>
<th>ステータス</th>
<th>作成日時</th>
<th>実行時間</th>
</tr>
</thead>
<tbody>
'''
for task in report_data['recent_tasks']:
status_class = 'success' if task.get('status') == 'SUCCESS' else 'failure'
exec_time = task.get('execution_time_seconds', 0)
html += f'''
<tr>
<td><code>{task.get('task_id', '')[:8]}...</code></td>
<td>{task.get('task_name', '')}</td>
<td class="{status_class}">{task.get('status', '')}</td>
<td>{task.get('created_at', '')}</td>
<td>{exec_time:.2f}秒</td>
</tr>
'''
html += '''
</tbody>
</table>
</body>
</html>
'''
return html
else:
raise ValueError(f"未知の出力形式: {output_format}")
# Celeryシグナルハンドラー
from celery.signals import task_prerun, task_postrun, task_failure, task_retry
@task_prerun.connect
def task_prerun_handler(sender=None, task_id=None, task=None, args=None, kwargs=None, **kwds):
"""タスク実行前のハンドラー"""
try:
db = TaskResultDatabase()
task_info = {
'task_id': task_id,
'task_name': task.name if hasattr(task, 'name') else str(task),
'status': 'STARTED',
'started_at': datetime.now().isoformat(),
'args': str(args)[:500],
'kwargs': str(kwargs)[:500]
}
db.save_task_result(**task_info)
logger.debug(f"タスク開始を記録: {task_id}")
except Exception as e:
logger.error(f"タスク開始記録エラー: {e}")
@task_postrun.connect
def task_postrun_handler(sender=None, task_id=None, task=None, args=None, kwargs=None,
retval=None, state=None, **kwds):
"""タスク実行後のハンドラー"""
try:
db = TaskResultDatabase()
task_info = {
'task_id': task_id,
'status': state,
'completed_at': datetime.now().isoformat(),
'result': retval if state == 'SUCCESS' else None,
'error_message': str(retval) if state == 'FAILURE' else None
}
db.save_task_result(**task_info)
logger.debug(f"タスク完了を記録: {task_id} - {state}")
except Exception as e:
logger.error(f"タスク完了記録エラー: {e}")
@task_failure.connect
def task_failure_handler(sender=None, task_id=None, exception=None,
traceback=None, einfo=None, **kwds):
"""タスク失敗ハンドラー"""
try:
db = TaskResultDatabase()
task_info = {
'task_id': task_id,
'status': 'FAILURE',
'error_message': str(exception),
'completed_at': datetime.now().isoformat()
}
db.save_task_result(**task_info)
logger.debug(f"タスク失敗を記録: {task_id}")
except Exception as e:
logger.error(f"タスク失敗記録エラー: {e}")
@task_retry.connect
def task_retry_handler(sender=None, request=None, reason=None, einfo=None, **kwds):
"""タスクリトライハンドラー"""
try:
if request and hasattr(request, 'id'):
db = TaskResultDatabase()
# 現在のリトライカウントを取得して更新
existing = db.get_task_result(request.id)
retry_count = existing.get('retry_count', 0) + 1 if existing else 1
task_info = {
'task_id': request.id,
'retry_count': retry_count,
'status': 'RETRY',
'error_message': str(reason) if reason else None
}
db.save_task_result(**task_info)
logger.debug(f"タスクリトライを記録: {request.id} - リトライ {retry_count}")
except Exception as e:
logger.error(f"タスクリトライ記録エラー: {e}")
# テスト関数
def test_task_persistence():
"""タスク永続化システムのテスト"""
print("=== タスク永続化システムテスト ===")
# データベースの初期化
print("\n1. データベース初期化テスト:")
db = TaskResultDatabase('./test_task_results.db')
# テストタスク結果の保存
print("\n2. タスク結果保存テスト:")
test_tasks = [
{
'task_id': 'test-task-001',
'task_name': 'test_task_1',
'status': 'SUCCESS',
'result': {'data': 'test result 1', 'processed': True},
'started_at': (datetime.now() - timedelta(hours=2)).isoformat(),
'completed_at': (datetime.now() - timedelta(hours=1, minutes=55)).isoformat(),
'queue_name': 'test_queue',
'priority': 1
},
{
'task_id': 'test-task-002',
'task_name': 'test_task_2',
'status': 'FAILURE',
'error': 'テストエラーが発生しました',
'started_at': (datetime.now() - timedelta(hours=1)).isoformat(),
'completed_at': (datetime.now() - timedelta(minutes=55)).isoformat(),
'queue_name': 'test_queue',
'priority': 2
},
{
'task_id': 'test-task-003',
'task_name': 'test_task_3',
'status': 'STARTED',
'started_at': datetime.now().isoformat(),
'queue_name': 'urgent_queue',
'priority': 0
}
]
for task_data in test_tasks:
success = db.save_task_result(**task_data)
print(f" タスク {task_data['task_id']} 保存: {'成功' if success else '失敗'}")
# タスク結果の取得テスト
print("\n3. タスク結果取得テスト:")
for task_id in ['test-task-001', 'test-task-002', 'test-task-999']:
result = db.get_task_result(task_id)
if result:
print(f" タスク {task_id}: {result.get('status')}, "
f"結果タイプ: {type(result.get('result')).__name__}")
else:
print(f" タスク {task_id}: 見つかりません")
# タスク検索テスト
print("\n4. タスク検索テスト:")
# ステータスで検索
success_tasks = db.search_tasks(filters={'status': 'SUCCESS'})
print(f" 成功タスク数: {len(success_tasks)}")
# キューで検索
test_queue_tasks = db.search_tasks(filters={'queue_name': 'test_queue'})
print(f" テストキュータスク数: {len(test_queue_tasks)}")
# 最近のタスクを取得
recent_tasks = db.search_tasks(limit=5)
print(f" 最近のタスク: {len(recent_tasks)}件")
for task in recent_tasks:
print(f" - {task.get('task_id')}: {task.get('task_name')} - {task.get('status')}")
# タスク統計テスト
print("\n5. タスク統計テスト:")
stats = db.get_task_statistics(days=7)
print(f" 総タスク数: {stats.get('total_tasks', 0)}")
print(f" ステータス別: {stats.get('by_status', {})}")
print(f" キュー別: {stats.get('by_queue', {})}")
# 永続化デコレータのテスト
print("\n6. 永続化デコレータテスト:")
# テスト用の永続化タスクを実行
test_task = example_persisted_task.delay({"test": "data"})
try:
result = test_task.get(timeout=10)
print(f" 永続化タスク実行: 成功")
# 結果をデータベースから確認
db_result = db.get_task_result(test_task.id)
if db_result:
print(f" データベース保存確認: {db_result.get('status')}")
else:
print(f" データベース保存確認: 見つかりません")
except Exception as e:
print(f" 永続化タスク実行エラー: {e}")
# タスクレポート生成テスト
print("\n7. タスクレポート生成テスト:")
monitor = TaskResultMonitor('./test_task_results.db')
# テキストレポート
text_report = monitor.generate_task_report(days=1, output_format='text')
print(" テキストレポート (一部):")
for line in text_report.split('\n')[:10]:
print(f" {line}")
# JSONレポート
json_report = monitor.generate_task_report(days=1, output_format='json')
print(f" JSONレポートサイズ: {len(json_report)} 文字")
# クリーンアップテスト
print("\n8. クリーンアップテスト:")
cleanup_result = db.cleanup_old_results(days_old=0) # 0日以上前を削除
print(f" 削除されたレコード: {cleanup_result.get('deleted_records', 0)}件")
# クリーンアップ（テスト用ファイルを削除）
import os
if os.path.exists('./test_task_results.db'):
os.unlink('./test_task_results.db')
print(f"\nテストデータベースを削除: ./test_task_results.db")
if os.path.exists('./special_tasks.db'):
os.unlink('./special_tasks.db')
print(f"特別データベースを削除: ./special_tasks.db")
return {
'tasks_saved': len(test_tasks),
'tasks_retrieved': len(success_tasks) + len(test_queue_tasks),
'statistics': stats,
'cleanup_result': cleanup_result
}
# タスク結果分析ツール（CLI）
def task_analysis_cli():
"""タスク分析CLIツール"""
import sys
import argparse
parser = argparse.ArgumentParser(description='タスク結果分析ツール')
parser.add_argument('--db', default='./task_results.db', help='データベースパス')
parser.add_argument('--days', type=int, default=7, help='分析期間（日）')
parser.add_argument('--format', choices=['text', 'json', 'html'], default='text', help='出力形式')
parser.add_argument('--recent', type=int, default=0, help='最近のタスクを表示する件数')
parser.add_argument('--status', help='特定のステータスのタスクを表示')
parser.add_argument('--cleanup', type=int, metavar='DAYS', help='古いタスクを削除（日数を指定）')
args = parser.parse_args()
if not os.path.exists(args.db):
print(f"エラー: データベースファイルが見つかりません: {args.db}")
sys.exit(1)
monitor = TaskResultMonitor(args.db)
if args.recent > 0:
# 最近のタスクを表示
tasks = monitor.get_recent_tasks(limit=args.recent)
print(f"\n最近の{args.recent}件のタスク:")
print("-" * 80)
for task in tasks:
print(f"ID: {task.get('task_id')}")
print(f" 名前: {task.get('task_name')}")
print(f" ステータス: {task.get('status')}")
print(f" 作成日時: {task.get('created_at')}")
print(f" 実行時間: {task.get('execution_time_seconds', 0):.2f}秒")
print()
elif args.status:
# 特定のステータスのタスクを表示
tasks = monitor.db.search_tasks(filters={'status': args.status}, limit=50)
print(f"\nステータス '{args.status}' のタスク ({len(tasks)}件):")
print("-" * 80)
for task in tasks[:10]: # 最初の10件のみ表示
print(f"ID: {task.get('task_id')}")
print(f" 名前: {task.get('task_name')}")
print(f" 作成日時: {task.get('created_at')}")
print(f" キュー: {task.get('queue_name', 'N/A')}")
print()
if len(tasks) > 10:
print(f"... 他 {len(tasks) - 10}件")
elif args.cleanup is not None:
# 古いタスクを削除
result = monitor.db.cleanup_old_results(days_old=args.cleanup)
print(f"\n古いタスクを削除しました:")
print(f" 削除対象: {args.cleanup}日以上前")
print(f" 削除件数: {result.get('deleted_records', 0)}件")
print(f" ステータス: {result.get('status')}")
else:
# レポートを生成
report = monitor.generate_task_report(days=args.days, output_format=args.format)
print(report)
if __name__ == "__main__":
# テスト実行
test_results = test_task_persistence()
# CLIツールをテスト（コメントアウト）
# task_analysis_cli()
print("\n" + "="*60)
print("タスク永続化システムテスト完了")
print("="*60)</code></pre>上級問題
問題10：分散タスク処理システム
# app/tasks/distributed_tasks.py
from app.celery_worker import celery
import time
from datetime import datetime, timedelta
import logging
import random
import threading
from collections import defaultdict
import json
import redis
from celery import current_app
from celery.result import AsyncResult, GroupResult
logger = logging.getLogger(__name__)
class DistributedTaskManager:
"""分散タスク管理クラス"""
def __init__(self, redis_url=None):
self.redis_client = self._get_redis_client(redis_url)
self.node_status = {}
self.load_balancer = LoadBalancer()
def _get_redis_client(self, redis_url):
"""Redisクライアントを取得"""
if redis_url is None:
redis_url = celery.conf.broker_url
from urllib.parse import urlparse
parsed = urlparse(redis_url)
return redis.Redis(
host=parsed.hostname or 'localhost',
port=parsed.port or 6379,
password=parsed.password,
decode_responses=True
)
def register_worker_node(self, node_id, node_info):
"""
ワーカーノードを登録
Args:
node_id: ノードID
node_info: ノード情報
Returns:
bool: 登録成功かどうか
"""
try:
# ノード情報をRedisに保存
node_key = f"distributed:worker:{node_id}"
self.redis_client.hset(node_key, mapping=node_info)
# ノードをアクティブセットに追加
self.redis_client.sadd("distributed:active_workers", node_id)
# ノードの最終更新時間を設定
self.redis_client.hset(node_key, "last_heartbeat", datetime.now().isoformat())
logger.info(f"ワーカーノードを登録: {node_id}")
return True
except Exception as e:
logger.error(f"ワーカーノード登録エラー: {e}")
return False
def update_node_status(self, node_id, status_update):
"""
ノードのステータスを更新
Args:
node_id: ノードID
status_update: ステータス更新情報
Returns:
bool: 更新成功かどうか
"""
try:
node_key = f"distributed:worker:{node_id}"
# 現在のステータスを取得
current_status = self.redis_client.hgetall(node_key)
# ステータスを更新
status_update["last_heartbeat"] = datetime.now().isoformat()
self.redis_client.hset(node_key, mapping=status_update)
# 負荷情報をロードバランサーに通知
if "current_load" in status_update:
self.load_balancer.update_node_load(
node_id,
float(status_update["current_load"])
)
logger.debug(f"ノードステータスを更新: {node_id}")
return True
except Exception as e:
logger.error(f"ノードステータス更新エラー: {e}")
return False
def get_available_nodes(self, min_capacity=0.1):
"""
利用可能なノードを取得
Args:
min_capacity: 最小容量閾値
Returns:
list: 利用可能なノードのリスト
"""
try:
# アクティブなワーカーを取得
active_workers = self.redis_client.smembers("distributed:active_workers")
available_nodes = []
for worker_id in active_workers:
node_key = f"distributed:worker:{worker_id}"
node_info = self.redis_client.hgetall(node_key)
# ノードが利用可能かチェック
if self._is_node_available(node_info, min_capacity):
available_nodes.append({
'node_id': worker_id,
'info': node_info
})
# 負荷でソート
available_nodes.sort(
key=lambda x: float(x['info'].get('current_load', 1.0))
)
return available_nodes
except Exception as e:
logger.error(f"利用可能ノード取得エラー: {e}")
return []
def _is_node_available(self, node_info, min_capacity):
"""ノードが利用可能かチェック"""
try:
# 最終ハートビートをチェック
last_heartbeat_str = node_info.get('last_heartbeat')
if not last_heartbeat_str:
return False
last_heartbeat = datetime.fromisoformat(last_heartbeat_str)
heartbeat_timeout = 60 # 60秒
if (datetime.now() - last_heartbeat).total_seconds() > heartbeat_timeout:
return False
# 容量をチェック
current_load = float(node_info.get('current_load', 1.0))
if current_load >= 1.0 - min_capacity:
return False
# ステータスをチェック
status = node_info.get('status', 'unknown')
if status != 'active':
return False
return True
except Exception as e:
logger.warning(f"ノード可用性チェックエラー: {e}")
return False
def distribute_task(self, task_name, task_args, task_kwargs=None,
strategy='load_balanced', max_nodes=None):
"""
タスクを分散して実行
Args:
task_name: タスク名
task_args: タスク引数
task_kwargs: タスクキーワード引数
strategy: 分散戦略 ('load_balanced', 'round_robin', 'random')
max_nodes: 最大ノード数
Returns:
dict: 分散実行結果
"""
try:
# 利用可能なノードを取得
available_nodes = self.get_available_nodes()
if not available_nodes:
raise Exception("利用可能なノードがありません")
# 最大ノード数の制限
if max_nodes and len(available_nodes) > max_nodes:
available_nodes = available_nodes[:max_nodes]
# 分散戦略に基づいてノードを選択
if strategy == 'load_balanced':
selected_nodes = self.load_balancer.select_nodes(
available_nodes, len(task_args) if isinstance(task_args, list) else 1
)
elif strategy == 'round_robin':
selected_nodes = self._round_robin_selection(
available_nodes, len(task_args) if isinstance(task_args, list) else 1
)
elif strategy == 'random':
selected_nodes = random.sample(
available_nodes,
min(len(available_nodes), len(task_args) if isinstance(task_args, list) else 1)
)
else:
selected_nodes = available_nodes[:1] # デフォルトは最初のノード
# タスクを分散実行
tasks = []
task_results = []
if isinstance(task_args, list):
# 複数のタスク引数がある場合、各ノードに分散
for i, (node, task_arg) in enumerate(zip(selected_nodes, task_args)):
# ノード固有のタスクキューを作成
queue_name = f"node_{node['node_id']}"
# タスクを実行
task = celery.send_task(
task_name,
args=[task_arg],
kwargs=task_kwargs or {},
queue=queue_name
)
tasks.append({
'task_id': task.id,
'node_id': node['node_id'],
'queue': queue_name,
'argument_index': i
})
else:
# 単一のタスク引数の場合、1つのノードで実行
node = selected_nodes[0]
queue_name = f"node_{node['node_id']}"
task = celery.send_task(
task_name,
args=[task_args],
kwargs=task_kwargs or {},
queue=queue_name
)
tasks.append({
'task_id': task.id,
'node_id': node['node_id'],
'queue': queue_name
})
# タスクの完了を待機（非同期の場合は別の方法で処理）
for task_info in tasks:
try:
task_result = AsyncResult(task_info['task_id'], app=celery)
result = task_result.get(timeout=300) # 5分タイムアウト
task_results.append({
**task_info,
'status': 'success',
'result': result
})
except Exception as e:
task_results.append({
**task_info,
'status': 'failed',
'error': str(e)
})
# 分散実行結果を返す
return {
'distribution_strategy': strategy,
'total_nodes_available': len(available_nodes),
'nodes_used': len(selected_nodes),
'tasks_dispatched': len(tasks),
'tasks_completed': len([r for r in task_results if r['status'] == 'success']),
'tasks_failed': len([r for r in task_results if r['status'] == 'failed']),
'task_results': task_results,
'distributed_at': datetime.now().isoformat()
}
except Exception as e:
logger.error(f"タスク分散エラー: {e}")
return {
'status': 'failed',
'error': str(e),
'distributed_at': datetime.now().isoformat()
}
def _round_robin_selection(self, nodes, count):
"""ラウンドロビン方式でノードを選択"""
if not nodes:
return []
# ラウンドロビンインデックスを取得/更新
index_key = "distributed:round_robin_index"
current_index = int(self.redis_client.get(index_key) or 0)
selected_nodes = []
for i in range(count):
selected_index = (current_index + i) % len(nodes)
selected_nodes.append(nodes[selected_index])
# インデックスを更新
new_index = (current_index + count) % len(nodes)
self.redis_client.set(index_key, new_index)
return selected_nodes
def monitor_system_health(self):
"""
システムの健全性を監視
Returns:
dict: 健全性チェック結果
"""
try:
# アクティブなワーカーを取得
active_workers = self.redis_client.smembers("distributed:active_workers")
node_health = []
total_load = 0
active_count = 0
for worker_id in active_workers:
node_key = f"distributed:worker:{worker_id}"
node_info = self.redis_client.hgetall(node_key)
# ノードの健全性をチェック
health_status = self._check_node_health(node_info)
node_health.append({
'node_id': worker_id,
'health': health_status,
'info': node_info
})
# 統計情報を収集
if health_status['status'] == 'healthy':
active_count += 1
total_load += float(node_info.get('current_load', 0))
# 平均負荷を計算
avg_load = total_load / active_count if active_count > 0 else 0
result = {
'timestamp': datetime.now().isoformat(),
'total_workers_registered': len(active_workers),
'active_workers': active_count,
'average_load': avg_load,
'node_health': node_health,
'overall_status': 'healthy' if active_count > 0 else 'unhealthy'
}
if avg_load > 0.8:
result['overall_status'] = 'warning'
result['warning'] = 'システム負荷が高いです'
return result
except Exception as e:
logger.error(f"システム健全性監視エラー: {e}")
return {
'timestamp': datetime.now().isoformat(),
'error': str(e),
'overall_status': 'unhealthy'
}
def _check_node_health(self, node_info):
"""ノードの健全性をチェック"""
try:
# 最終ハートビートをチェック
last_heartbeat_str = node_info.get('last_heartbeat')
if not last_heartbeat_str:
return {'status': 'unhealthy', 'reason': 'ハートビートがありません'}
last_heartbeat = datetime.fromisoformat(last_heartbeat_str)
heartbeat_timeout = 60 # 60秒
if (datetime.now() - last_heartbeat).total_seconds() > heartbeat_timeout:
return {'status': 'unhealthy', 'reason': 'ハートビートがタイムアウトしました'}
# 負荷をチェック
current_load = float(node_info.get('current_load', 0))
if current_load > 0.9:
return {'status': 'warning', 'reason': '負荷が高いです', 'load': current_load}
# メモリ使用量をチェック
memory_usage = node_info.get('memory_usage')
if memory_usage and float(memory_usage) > 90:
return {'status': 'warning', 'reason': 'メモリ使用量が高いです'}
return {'status': 'healthy', 'load': current_load}
except Exception as e:
return {'status': 'unhealthy', 'reason': f'健全性チェックエラー: {str(e)}'}
class LoadBalancer:
"""ロードバランサークラス"""
def __init__(self):
self.node_loads = defaultdict(float)
self.node_capacities = defaultdict(float)
self.selection_history = []
def update_node_load(self, node_id, load):
"""
ノードの負荷を更新
Args:
node_id: ノードID
load: 負荷値 (0.0〜1.0)
"""
self.node_loads[node_id] = load
def update_node_capacity(self, node_id, capacity):
"""
ノードの容量を更新
Args:
node_id: ノードID
capacity: 容量値
"""
self.node_capacities[node_id] = capacity
def select_nodes(self, available_nodes, count):
"""
負荷に基づいてノードを選択
Args:
available_nodes: 利用可能なノードのリスト
count: 選択するノード数
Returns:
list: 選択されたノード
"""
if not available_nodes:
return []
# ノードの可用性を計算
node_scores = []
for node in available_nodes:
node_id = node['node_id']
node_info = node['info']
# 現在の負荷を取得
current_load = float(node_info.get('current_load', 0.5))
# ノードの容量を考慮
capacity = float(node_info.get('capacity', 1.0))
# 可用性スコアを計算（負荷が低いほど高スコア）
availability_score = capacity * (1.0 - current_load)
# 過去の選択履歴を考慮（最近選択されていないほど高スコア）
recent_selections = len([
h for h in self.selection_history[-10:]
if h.get('node_id') == node_id
])
history_penalty = recent_selections * 0.1
final_score = max(0.01, availability_score - history_penalty)
node_scores.append((final_score, node))
# スコアでソート（降順）
node_scores.sort(key=lambda x: x[0], reverse=True)
# 上位のノードを選択
selected_nodes = [node for _, node in node_scores[:count]]
# 選択履歴を記録
for node in selected_nodes:
self.selection_history.append({
'node_id': node['node_id'],
'timestamp': datetime.now().isoformat(),
'selection_count': count
})
# 履歴が長すぎる場合は古いものを削除
if len(self.selection_history) > 100:
self.selection_history = self.selection_history[-100:]
return selected_nodes
def get_load_distribution(self):
"""
負荷分布を取得
Returns:
dict: 負荷分布情報
"""
total_load = sum(self.node_loads.values())
node_count = len(self.node_loads)
avg_load = total_load / node_count if node_count > 0 else 0
# 負荷の標準偏差を計算
if node_count > 1:
variance = sum((load - avg_load) ** 2 for load in self.node_loads.values()) / node_count
std_dev = variance ** 0.5
else:
std_dev = 0
return {
'total_nodes': node_count,
'average_load': avg_load,
'std_deviation': std_dev,
'node_loads': dict(self.node_loads),
'load_imbalance': std_dev / avg_load if avg_load > 0 else 0
}
# 分散タスクの例
@celery.task(bind=True)
def distributed_data_processing_task(self, data_chunk, processing_type='default'):
"""
分散データ処理タスク
Args:
data_chunk: 処理するデータチャンク
processing_type: 処理タイプ
Returns:
dict: 処理結果
"""
task_id = self.request.id
worker_node = self.request.hostname
logger.info(f"分散データ処理開始 [{worker_node}]: {task_id}")
try:
# 処理タイプに基づいた処理
if processing_type == 'analysis':
result = self._analyze_data(data_chunk)
elif processing_type == 'transformation':
result = self._transform_data(data_chunk)
elif processing_type == 'validation':
result = self._validate_data(data_chunk)
else:
result = self._process_data_default(data_chunk)
# ノードの負荷情報を更新（実際の実装では別の方法で）
manager = DistributedTaskManager()
manager.update_node_status(worker_node, {
'last_task_completed': datetime.now().isoformat(),
'task_result': 'success'
})
return {
'task_id': task_id,
'worker_node': worker_node,
'processing_type': processing_type,
'result': result,
'processed_at': datetime.now().isoformat()
}
except Exception as e:
logger.error(f"分散データ処理エラー [{worker_node}]: {task_id} - {e}")
# エラー情報を更新
manager = DistributedTaskManager()
manager.update_node_status(worker_node, {
'last_task_failed': datetime.now().isoformat(),
'task_error': str(e)
})
raise
def _analyze_data(self, data):
"""データ分析"""
import time
time.sleep(random.uniform(0.5, 2.0))
# 簡易的な分析
if isinstance(data, list):
return {
'count': len(data),
'sum': sum(data) if all(isinstance(x, (int, float)) for x in data) else None,
'average': sum(data)/len(data) if len(data) > 0 and all(isinstance(x, (int, float)) for x in data) else None
}
else:
return {'input': str(data)[:100], 'analysis': 'completed'}
def _transform_data(self, data):
"""データ変換"""
import time
time.sleep(random.uniform(0.3, 1.5))
# 簡易的な変換
if isinstance(data, dict):
return {k.upper(): str(v).upper() for k, v in data.items()}
elif isinstance(data, list):
return [x * 2 if isinstance(x, (int, float)) else str(x) * 2 for x in data]
else:
return str(data) * 2
def _validate_data(self, data):
"""データ検証"""
import time
time.sleep(random.uniform(0.2, 1.0))
# 簡易的な検証
validation_result = {
'is_valid': True,
'errors': [],
'warnings': []
}
if data is None:
validation_result['is_valid'] = False
validation_result['errors'].append('データがNoneです')
elif isinstance(data, str) and len(data) > 1000:
validation_result['warnings'].append('データが長すぎます')
return validation_result
def _process_data_default(self, data):
"""デフォルトのデータ処理"""
import time
time.sleep(random.uniform(0.1, 0.5))
return {
'processed': True,
'input_type': type(data).__name__,
'input_length': len(str(data)),
'checksum': hash(str(data)) % 10000
}
# ワーカーノードエミュレーター
class WorkerNodeEmulator:
"""ワーカーノードエミュレーター（テスト用）"""
def __init__(self, node_id, capacity=1.0):
self.node_id = node_id
self.capacity = capacity
self.current_load = 0.0
self.status = 'active'
self.manager = DistributedTaskManager()
# ノードを登録
self.register_node()
# ハートビートスレッドを開始
self.heartbeat_thread = threading.Thread(target=self._heartbeat_loop, daemon=True)
self.heartbeat_thread.start()
def register_node(self):
"""ノードを登録"""
node_info = {
'node_id': self.node_id,
'capacity': str(self.capacity),
'current_load': str(self.current_load),
'status': self.status,
'registered_at': datetime.now().isoformat(),
'type': 'emulated',
'version': '1.0'
}
return self.manager.register_worker_node(self.node_id, node_info)
def _heartbeat_loop(self):
"""ハートビートループ"""
while True:
try:
# ランダムな負荷を生成
self.current_load = random.uniform(0.0, 0.8)
# ステータスを更新
status_update = {
'current_load': str(self.current_load),
'status': self.status,
'memory_usage': str(random.uniform(30.0, 80.0)),
'cpu_usage': str(random.uniform(10.0, 70.0)),
'tasks_processed': str(random.randint(100, 1000))
}
self.manager.update_node_status(self.node_id, status_update)
# 30秒ごとにハートビート
time.sleep(30)
except Exception as e:
logger.error(f"ハートビートエラー [{self.node_id}]: {e}")
time.sleep(10)
def simulate_failure(self):
"""ノード障害をシミュレート"""
self.status = 'failed'
logger.warning(f"ノード障害をシミュレート: {self.node_id}")
def simulate_recovery(self):
"""ノード復旧をシミュレート"""
self.status = 'active'
logger.info(f"ノード復旧をシミュレート: {self.node_id}")
# 分散タスク実行のテスト
def test_distributed_system():
"""分散タスクシステムのテスト"""
print("=== 分散タスクシステムテスト ===")
# 分散タスクマネージャーの作成
manager = DistributedTaskManager()
# ワーカーノードエミュレーターの作成
print("\n1. ワーカーノードエミュレーターの作成:")
nodes = []
for i in range(5):
node_id = f"worker-node-{i:03d}"
node = WorkerNodeEmulator(node_id, capacity=1.0 + i * 0.2)
nodes.append(node)
print(f" ノード作成: {node_id}")
time.sleep(2) # ノード登録を待機
# 利用可能なノードを確認
print("\n2. 利用可能なノードの確認:")
available_nodes = manager.get_available_nodes()
print(f" 利用可能なノード数: {len(available_nodes)}")
for node in available_nodes[:3]: # 最初の3ノードのみ表示
print(f" - {node['node_id']}: 負荷={node['info'].get('current_load', 'N/A')}")
# 分散タスク実行のテスト
print("\n3. 分散タスク実行テスト:")
# テストデータの準備
data_chunks = [
[1, 2, 3, 4, 5],
[10, 20, 30, 40, 50],
[100, 200, 300, 400, 500],
{"name": "test", "value": 123},
"test string"
]
# 負荷分散戦略でタスクを分散実行
distribution_result = manager.distribute_task(
task_name='app.tasks.distributed_tasks.distributed_data_processing_task',
task_args=data_chunks,
task_kwargs={'processing_type': 'analysis'},
strategy='load_balanced',
max_nodes=3
)
print(f" 分散戦略: {distribution_result.get('distribution_strategy')}")
print(f" 利用可能ノード: {distribution_result.get('total_nodes_available')}")
print(f" 使用ノード: {distribution_result.get('nodes_used')}")
print(f" 実行タスク: {distribution_result.get('tasks_dispatched')}")
print(f" 成功タスク: {distribution_result.get('tasks_completed')}")
print(f" 失敗タスク: {distribution_result.get('tasks_failed')}")
# システム健全性チェック
print("\n4. システム健全性チェック:")
health_result = manager.monitor_system_health()
print(f" 登録ワーカー: {health_result.get('total_workers_registered')}")
print(f" アクティブワーカー: {health_result.get('active_workers')}")
print(f" 平均負荷: {health_result.get('average_load', 0):.2f}")
print(f" 全体ステータス: {health_result.get('overall_status')}")
# ロードバランサーのテスト
print("\n5. ロードバランサーテスト:")
load_balancer = LoadBalancer()
# ノード負荷を更新
for node in available_nodes:
load = random.uniform(0.1, 0.9)
load_balancer.update_node_load(node['node_id'], load)
# 負荷分布を取得
load_dist = load_balancer.get_load_distribution()
print(f" 総ノード数: {load_dist['total_nodes']}")
print(f" 平均負荷: {load_dist['average_load']:.2f}")
print(f" 負荷不均一性: {load_dist['load_imbalance']:.2f}")
# ノード選択テスト
selected_nodes = load_balancer.select_nodes(available_nodes, 2)
print(f" 選択されたノード: {[n['node_id'] for n in selected_nodes]}")
# 障害シミュレーション
print("\n6. 障害シミュレーションテスト:")
if nodes:
nodes[0].simulate_failure()
time.sleep(5) # 状態更新を待機
# 障害後の健全性チェック
health_after_failure = manager.monitor_system_health()
print(f" 障害後のアクティブワーカー: {health_after_failure.get('active_workers')}")
# 復旧シミュレーション
nodes[0].simulate_recovery()
time.sleep(5)
health_after_recovery = manager.monitor_system_health()
print(f" 復旧後のアクティブワーカー: {health_after_recovery.get('active_workers')}")
# クリーンアップ（エミュレータの停止）
print("\n7. クリーンアップ:")
print(" エミュレータを停止します...")
return {
'distribution_test': distribution_result,
'health_check': health_result,
'load_balancer_test': load_dist,
'nodes_created': len(nodes)
}
# 分散タスクモニタリングダッシュボード
def display_distributed_dashboard():
"""分散タスクダッシュボードを表示"""
import time
from datetime import datetime
print("\n" + "="*60)
print("分散タスクシステムダッシュボード")
print("="*60)
manager = DistributedTaskManager()
load_balancer = LoadBalancer()
try:
while True:
# 画面をクリア（プラットフォームに依存）
import os
os.system('cls' if os.name == 'nt' else 'clear')
current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
print(f"\n現在時刻: {current_time}")
print("-"*60)
# システム概要を表示
print("\n[システム概要]")
health = manager.monitor_system_health()
print(f" 🔧 総ワーカー数: {health.get('total_workers_registered', 0)}")
print(f" ✅ アクティブワーカー: {health.get('active_workers', 0)}")
print(f" 📊 平均負荷: {health.get('average_load', 0):.2f}")
print(f" 🩺 全体ステータス: {health.get('overall_status', 'unknown')}")
if 'warning' in health:
print(f" ⚠️ 警告: {health['warning']}")
# ノード詳細を表示
print("\n[ノード詳細]")
available_nodes = manager.get_available_nodes()
print(f" 利用可能ノード: {len(available_nodes)}")
for i, node in enumerate(available_nodes[:5]): # 最初の5ノードのみ表示
node_info = node['info']
load = float(node_info.get('current_load', 0))
status = node_info.get('status', 'unknown')
# 負荷バーを表示
bar_length = 20
filled = int(load * bar_length)
bar = '█' * filled + '░' * (bar_length - filled)
print(f" {i+1}. {node['node_id']}:")
print(f" 負荷: [{bar}] {load:.1%}")
print(f" ステータス: {status}")
print(f" 最終更新: {node_info.get('last_heartbeat', 'N/A')}")
if len(available_nodes) > 5:
print(f" ... 他 {len(available_nodes) - 5}ノード")
# 負荷分布を表示
print("\n[負荷分布]")
load_dist = load_balancer.get_load_distribution()
print(f" 負荷不均一性: {load_dist.get('load_imbalance', 0):.3f}")
if load_dist['node_loads']:
sorted_loads = sorted(load_dist['node_loads'].items(), key=lambda x: x[1])
print(f" 最低負荷: {sorted_loads[0][0]} = {sorted_loads[0][1]:.2f}")
print(f" 最高負荷: {sorted_loads[-1][0]} = {sorted_loads[-1][1]:.2f}")
# メニューを表示
print("\n" + "-"*60)
print("[操作メニュー]")
print(" 1. システム健全性を更新")
print(" 2. テストタスクを実行")
print(" 3. 負荷分散テスト")
print(" 4. 終了")
print("-"*60)
choice = input("選択してください (1-4): ").strip()
if choice == '1':
print("\nシステム健全性をチェック中...")
health = manager.monitor_system_health()
print(f"結果: {health.get('overall_status')}")
input("\nEnterキーを押して続行...")
elif choice == '2':
print("\nテストタスクを実行中...")
test_data = [[1, 2, 3], [4, 5, 6], [7, 8, 9]]
result = manager.distribute_task(
task_name='app.tasks.distributed_tasks.distributed_data_processing_task',
task_args=test_data,
strategy='load_balanced'
)
print(f"結果: {result.get('tasks_completed')}/{result.get('tasks_dispatched')} 成功")
input("\nEnterキーを押して続行...")
elif choice == '3':
print("\n負荷分散テストを実行中...")
available_nodes = manager.get_available_nodes()
if available_nodes:
selected = load_balancer.select_nodes(available_nodes, 2)
print(f"選択されたノード: {[n['node_id'] for n in selected]}")
else:
print("利用可能なノードがありません")
input("\nEnterキーを押して続行...")
elif choice == '4':
print("\nダッシュボードを終了します")
break
else:
print("\n無効な選択です")
time.sleep(1)
except KeyboardInterrupt:
print("\n\nダッシュボードを終了します")
if __name__ == "__main__":
# テスト実行
test_results = test_distributed_system()
# ダッシュボード表示（オプション）
# display_distributed_dashboard()問題11：リアルタイムタスクモニタリングダッシュボード
<pre><code class="language-python"># app/tasks/realtime_monitor.py
from app.celery_worker import celery
from celery.result import AsyncResult
import time
from datetime import datetime, timedelta
import logging
import json
import asyncio
import websockets
from threading import Thread
import psutil
import redis
from collections import deque
from pathlib import Path
logger = logging.getLogger(__name__)
class RealTimeTaskMonitor:
"""リアルタイムタスクモニター"""
def __init__(self, redis_url=None, history_size=1000):
self.redis_client = self._get_redis_client(redis_url)
self.history_size = history_size
self.active_tasks = {}
self.task_history = deque(maxlen=history_size)
self.system_metrics = deque(maxlen=100)
# WebSocket接続を管理
self.websocket_clients = set()
# 監視スレッド
self.monitoring_thread = None
self.is_monitoring = False
def _get_redis_client(self, redis_url):
"""Redisクライアントを取得"""
if redis_url is None:
redis_url = celery.conf.broker_url
from urllib.parse import urlparse
parsed = urlparse(redis_url)
return redis.Redis(
host=parsed.hostname or 'localhost',
port=parsed.port or 6379,
password=parsed.password,
decode_responses=False # バイナリデータも扱うためFalse
)
def start_monitoring(self, interval=1.0):
"""
監視を開始
Args:
interval: 監視間隔（秒）
"""
if self.is_monitoring:
logger.warning("監視は既に実行中です")
return
self.is_monitoring = True
self.monitoring_thread = Thread(
target=self._monitoring_loop,
args=(interval,),
daemon=True
)
self.monitoring_thread.start()
logger.info(f"リアルタイム監視を開始: 間隔={interval}秒")
def stop_monitoring(self):
"""監視を停止"""
self.is_monitoring = False
if self.monitoring_thread:
self.monitoring_thread.join(timeout=5)
logger.info("リアルタイム監視を停止")
def _monitoring_loop(self, interval):
"""監視ループ"""
while self.is_monitoring:
try:
# システムメトリクスを収集
system_metrics = self._collect_system_metrics()
self.system_metrics.append(system_metrics)
# アクティブなタスクを収集
active_tasks = self._collect_active_tasks()
# 変更を検出
task_updates = self._detect_task_changes(active_tasks)
# 履歴を更新
self._update_task_history(task_updates)
# WebSocketクライアントに更新を通知
self._notify_clients({
'type': 'update',
'timestamp': datetime.now().isoformat(),
'system_metrics': system_metrics,
'task_updates': task_updates,
'active_task_count': len(active_tasks)
})
# 指定間隔で待機
time.sleep(interval)
except Exception as e:
logger.error(f"監視ループエラー: {e}")
time.sleep(interval)
def _collect_system_metrics(self):
"""システムメトリクスを収集"""
try:
# CPU使用率
cpu_percent = psutil.cpu_percent(interval=0.1)
# メモリ使用率
memory = psutil.virtual_memory()
# ディスク使用率
disk = psutil.disk_usage('/')
# ネットワークI/O
net_io = psutil.net_io_counters()
# Redis接続情報
redis_info = {}
try:
redis_info = {
'connected_clients': self.redis_client.info().get('connected_clients', 0),
'used_memory': self.redis_client.info().get('used_memory', 0),
'total_commands_processed': self.redis_client.info().get('total_commands_processed', 0)
}
except:
pass
metrics = {
'timestamp': datetime.now().isoformat(),
'cpu': {
'percent': cpu_percent,
'cores': psutil.cpu_count(),
'load_avg': psutil.getloadavg() if hasattr(psutil, 'getloadavg') else [0, 0, 0]
},
'memory': {
'total': memory.total,
'available': memory.available,
'percent': memory.percent,
'used': memory.used
},
'disk': {
'total': disk.total,
'used': disk.used,
'free': disk.free,
'percent': disk.percent
},
'network': {
'bytes_sent': net_io.bytes_sent,
'bytes_recv': net_io.bytes_recv,
'packets_sent': net_io.packets_sent,
'packets_recv': net_io.packets_recv
},
'redis': redis_info
}
return metrics
except Exception as e:
logger.error(f"システムメトリクス収集エラー: {e}")
return {
'timestamp': datetime.now().isoformat(),
'error': str(e)
}
def _collect_active_tasks(self):
"""アクティブなタスクを収集"""
active_tasks = {}
try:
# Celeryのアクティブタスクを取得
# 注: 実際の実装ではCeleryの監視APIを使用
# ここではRedisから簡易的に取得
# アクティブなワーカーを取得
workers = self.redis_client.smembers('celery:workers')
for worker in workers:
try:
# ワーカーのアクティブタスクを取得
worker_key = f'celery:active:{worker}'
active_task = self.redis_client.get(worker_key)
if active_task:
# タスク情報をパース
task_info = json.loads(active_task)
task_id = task_info.get('id')
if task_id:
active_tasks[task_id] = {
'task_id': task_id,
'worker': worker,
'name': task_info.get('name', 'unknown'),
'args': task_info.get('args', []),
'kwargs': task_info.get('kwargs', {}),
'received': task_info.get('time_start', datetime.now().timestamp()),
'state': 'PROGRESS',
'info': task_info.get('info', {})
}
except Exception as e:
logger.debug(f"ワーカー {worker} のタスク取得エラー: {e}")
# 保留中のタスク（キュー内）を取得
queues = ['default', 'high', 'low', 'urgent']
for queue in queues:
queue_length = self.redis_client.llen(f'celery:{queue}')
if queue_length > 0:
# キューの最初のタスクを取得（サンプル）
task_data = self.redis_client.lrange(f'celery:{queue}', 0, 0)
if task_data:
try:
task_info = json.loads(task_data[0])
task_id = task_info.get('headers', {}).get('id')
if task_id and task_id not in active_tasks:
active_tasks[task_id] = {
'task_id': task_id,
'queue': queue,
'name': task_info.get('headers', {}).get('task', 'unknown'),
'state': 'PENDING',
'queued_at': datetime.now().isoformat()
}
except Exception as e:
logger.debug(f"キュー {queue} のタスク解析エラー: {e}")
except Exception as e:
logger.error(f"アクティブタスク収集エラー: {e}")
return active_tasks
def _detect_task_changes(self, new_active_tasks):
"""タスクの変更を検出"""
updates = []
# 新しいタスクまたは更新されたタスク
for task_id, task_info in new_active_tasks.items():
if task_id not in self.active_tasks:
# 新しいタスク
updates.append({
'type': 'task_started',
'task_id': task_id,
'task_info': task_info,
'timestamp': datetime.now().isoformat()
})
else:
# 既存のタスクの更新をチェック
old_info = self.active_tasks[task_id]
# 状態の変化をチェック
if task_info.get('state') != old_info.get('state'):
updates.append({
'type': 'task_state_changed',
'task_id': task_id,
'old_state': old_info.get('state'),
'new_state': task_info.get('state'),
'task_info': task_info,
'timestamp': datetime.now().isoformat()
})
# 進捗状況の更新をチェック
old_progress = old_info.get('info', {}).get('current', 0)
new_progress = task_info.get('info', {}).get('current', 0)
if new_progress != old_progress:
updates.append({
'type': 'task_progress',
'task_id': task_id,
'progress': new_progress,
'task_info': task_info,
'timestamp': datetime.now().isoformat()
})
# 完了または削除されたタスク
for task_id in list(self.active_tasks.keys()):
if task_id not in new_active_tasks:
# タスクが完了または削除
old_info = self.active_tasks[task_id]
# タスク結果を確認
try:
task_result = AsyncResult(task_id, app=celery)
if task_result.ready():
if task_result.successful():
updates.append({
'type': 'task_completed',
'task_id': task_id,
'result': task_result.result,
'task_info': old_info,
'timestamp': datetime.now().isoformat()
})
else:
updates.append({
'type': 'task_failed',
'task_id': task_id,
'error': str(task_result.result),
'task_info': old_info,
'timestamp': datetime.now().isoformat()
})
else:
# 単にアクティブリストから消えた場合
updates.append({
'type': 'task_disappeared',
'task_id': task_id,
'task_info': old_info,
'timestamp': datetime.now().isoformat()
})
except Exception as e:
updates.append({
'type': 'task_unknown',
'task_id': task_id,
'error': str(e),
'task_info': old_info,
'timestamp': datetime.now().isoformat()
})
# アクティブタスクリストを更新
self.active_tasks = new_active_tasks
return updates
def _update_task_history(self, updates):
"""タスク履歴を更新"""
for update in updates:
self.task_history.append(update)
def _notify_clients(self, data):
"""WebSocketクライアントに通知"""
if not self.websocket_clients:
return
try:
message = json.dumps(data, default=str)
# 各クライアントに非同期で送信
for client in list(self.websocket_clients):
try:
asyncio.run_coroutine_threadsafe(
client.send(message),
asyncio.get_event_loop()
)
except Exception as e:
logger.debug(f"クライアント通知エラー: {e}")
# 接続が閉じている場合は削除
self.websocket_clients.discard(client)
except Exception as e:
logger.error(f"クライアント通知エラー: {e}")
def register_websocket_client(self, websocket):
"""WebSocketクライアントを登録"""
self.websocket_clients.add(websocket)
logger.info(f"WebSocketクライアントを登録: 総数={len(self.websocket_clients)}")
def unregister_websocket_client(self, websocket):
"""WebSocketクライアントを登録解除"""
self.websocket_clients.discard(websocket)
logger.info(f"WebSocketクライアントを登録解除: 残数={len(self.websocket_clients)}")
def get_dashboard_data(self):
"""
ダッシュボードデータを取得
Returns:
dict: ダッシュボードデータ
"""
# 最新のシステムメトリクス
latest_metrics = list(self.system_metrics)[-1] if self.system_metrics else {}
# アクティブタスクの統計
task_stats = {
'total_active': len(self.active_tasks),
'by_state': {},
'by_queue': {},
'by_worker': {}
}
for task_id, task_info in self.active_tasks.items():
state = task_info.get('state', 'UNKNOWN')
task_stats['by_state'][state] = task_stats['by_state'].get(state, 0) + 1
queue = task_info.get('queue', 'unknown')
task_stats['by_queue'][queue] = task_stats['by_queue'].get(queue, 0) + 1
worker = task_info.get('worker', 'unknown')
task_stats['by_worker'][worker] = task_stats['by_worker'].get(worker, 0) + 1
# 最近のタスク更新
recent_updates = list(self.task_history)[-20:] # 最新20件
# キューの統計
queue_stats = {}
try:
queues = ['default', 'high', 'low', 'urgent']
for queue in queues:
length = self.redis_client.llen(f'celery:{queue}')
queue_stats[queue] = length
except:
pass
data = {
'timestamp': datetime.now().isoformat(),
'system_metrics': latest_metrics,
'task_stats': task_stats,
'recent_updates': recent_updates,
'queue_stats': queue_stats,
'active_tasks': list(self.active_tasks.values())[:50], # 最新50件
'history_size': len(self.task_history),
'websocket_clients': len(self.websocket_clients)
}
return data
def get_task_details(self, task_id):
"""
タスクの詳細情報を取得
Args:
task_id: タスクID
Returns:
dict: タスク詳細情報
"""
try:
# タスク結果を取得
task_result = AsyncResult(task_id, app=celery)
details = {
'task_id': task_id,
'status': task_result.status,
'ready': task_result.ready(),
'successful': task_result.successful(),
'failed': task_result.failed(),
'state': task_result.state
}
if task_result.ready():
if task_result.successful():
details['result'] = task_result.result
else:
details['error'] = str(task_result.result)
# 進捗情報
if task_result.status == 'PROGRESS':
details['progress'] = task_result.info
# タスクメタデータ（Redisから）
try:
task_key = f'celery-task-meta-{task_id}'
task_data = self.redis_client.get(task_key)
if task_data:
task_meta = json.loads(task_data)
details['metadata'] = task_meta
except:
pass
# 履歴から関連情報を取得
related_updates = []
for update in self.task_history:
if update.get('task_id') == task_id:
related_updates.append(update)
details['history'] = related_updates[-10:] # 最新10件
return details
except Exception as e:
return {
'task_id': task_id,
'error': str(e),
'timestamp': datetime.now().isoformat()
}
# WebSocketサーバー
class WebSocketServer:
"""WebSocketサーバー"""
def __init__(self, host='localhost', port=8765):
self.host = host
self.port = port
self.monitor = RealTimeTaskMonitor()
self.server = None
async def handler(self, websocket, path):
"""WebSocketハンドラー"""
# クライアントを登録
self.monitor.register_websocket_client(websocket)
try:
# 初期データを送信
initial_data = self.monitor.get_dashboard_data()
await websocket.send(json.dumps({
'type': 'initial',
'data': initial_data
}, default=str))
# メッセージの受信を待機
async for message in websocket:
try:
data = json.loads(message)
await self._handle_client_message(websocket, data)
except json.JSONDecodeError:
await websocket.send(json.dumps({
'type': 'error',
'message': 'Invalid JSON'
}))
except Exception as e:
logger.error(f"メッセージ処理エラー: {e}")
except websockets.exceptions.ConnectionClosed:
logger.debug("WebSocket接続が閉じられました")
finally:
# クライアントを登録解除
self.monitor.unregister_websocket_client(websocket)
async def _handle_client_message(self, websocket, data):
"""クライアントメッセージを処理"""
message_type = data.get('type')
if message_type == 'get_dashboard':
# ダッシュボードデータを送信
dashboard_data = self.monitor.get_dashboard_data()
await websocket.send(json.dumps({
'type': 'dashboard_data',
'data': dashboard_data
}, default=str))
elif message_type == 'get_task_details':
# タスク詳細を送信
task_id = data.get('task_id')
if task_id:
task_details = self.monitor.get_task_details(task_id)
await websocket.send(json.dumps({
'type': 'task_details',
'data': task_details
}, default=str))
elif message_type == 'subscribe_task':
# 特定のタスクの更新を購読
task_id = data.get('task_id')
# 実装は省略（必要に応じて実装）
await websocket.send(json.dumps({
'type': 'subscribed',
'task_id': task_id
}))
elif message_type == 'control_task':
# タスクを制御（キャンセルなど）
task_id = data.get('task_id')
action = data.get('action')
if action == 'cancel':
try:
celery.control.revoke(task_id, terminate=True)
await websocket.send(json.dumps({
'type': 'task_cancelled',
'task_id': task_id
}))
except Exception as e:
await websocket.send(json.dumps({
'type': 'error',
'message': f'Failed to cancel task: {str(e)}'
}))
else:
await websocket.send(json.dumps({
'type': 'error',
'message': f'Unknown message type: {message_type}'
}))
def start(self):
"""サーバーを開始"""
# 監視を開始
self.monitor.start_monitoring(interval=1.0)
# WebSocketサーバーを開始
start_server = websockets.serve(self.handler, self.host, self.port)
# イベントループを取得
loop = asyncio.get_event_loop()
self.server = loop.run_until_complete(start_server)
logger.info(f"WebSocketサーバーを開始: ws://{self.host}:{self.port}")
# サーバーを実行
loop.run_forever()
def stop(self):
"""サーバーを停止"""
if self.server:
self.server.close()
self.monitor.stop_monitoring()
logger.info("WebSocketサーバーを停止")
# HTMLダッシュボードテンプレート
HTML_DASHBOARD_TEMPLATE = """
...（中略：HTMLテンプレートは元の内容のままです）...
"""
# テスト関数
def test_realtime_monitor():
"""リアルタイムモニターのテスト"""
...
</code></pre>問題12：フォールトトレラントなメール送信システム
```python
app/tasks/fault_tolerant_email.py
from app.celery_worker import celery
from flask_mail import Message
from flask import current_app
import logging
from datetime import datetime, timedelta
import time
import random
from typing import List, Dict, Optional, Tuple
import sqlite3
import json
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import ssl
logger = logging.getLogger(name)
class FaultTolerantMailSystem:
"""フォールトトレラントなメール送信システム"""
def __init__(self, db_path='./email_queue.db'):
self.db_path = db_path
self.smtp_servers = []
self.init_database()
self.load_smtp_config()
def init_database(self):
"""データベースを初期化"""
try:
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
# メールキュー
cursor.execute('''
CREATE TABLE IF NOT EXISTS email_queue (
id INTEGER PRIMARY KEY AUTOINCREMENT,
to_address TEXT NOT NULL,
subject TEXT NOT NULL,
html_body TEXT,
text_body TEXT,
cc TEXT,
bcc TEXT,
reply_to TEXT,
attachments TEXT, -- JSON形式
status TEXT NOT NULL DEFAULT 'pending',
priority INTEGER DEFAULT 2, -- 1:高, 2:中, 3:低
retry_count INTEGER DEFAULT 0,
max_retries INTEGER DEFAULT 3,
last_attempt TIMESTAMP,
next_retry TIMESTAMP,
error_message TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
sent_at TIMESTAMP,
smtp_server_used TEXT
)
''')
# インデックスの作成
cursor.execute('CREATE INDEX IF NOT EXISTS idx_status ON email_queue(status)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_priority ON email_queue(priority)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_next_retry ON email_queue(next_retry)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_created_at ON email_queue(created_at)')
# SMTPサーバー設定
cursor.execute('''
CREATE TABLE IF NOT EXISTS smtp_servers (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
host TEXT NOT NULL,
port INTEGER NOT NULL,
username TEXT,
password TEXT,
use_tls BOOLEAN DEFAULT 1,
use_ssl BOOLEAN DEFAULT 0,
max_daily_limit INTEGER DEFAULT 1000,
today_sent INTEGER DEFAULT 0,
last_reset_date DATE DEFAULT CURRENT_DATE,
is_active BOOLEAN DEFAULT 1,
failure_count INTEGER DEFAULT 0,
last_failure TIMESTAMP,
priority INTEGER DEFAULT 2,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# 送信ログ
cursor.execute('''
CREATE TABLE IF NOT EXISTS email_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
email_id INTEGER NOT NULL,
smtp_server_id INTEGER,
status TEXT NOT NULL,
error_message TEXT,
response_time_ms INTEGER,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (email_id) REFERENCES email_queue(id),
FOREIGN KEY (smtp_server_id) REFERENCES smtp_servers(id)
)
''')
conn.commit()
conn.close()
logger.info(f"メールシステムデータベースを初期化: {self.db_path}")
except Exception as e:
logger.error(f"データベース初期化エラー: {e}")
raise
def load_smtp_config(self):
"""SMTPサーバー設定を読み込み"""
try:
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute('SELECT * FROM smtp_servers WHERE is_active = 1 ORDER BY priority, failure_count')
rows = cursor.fetchall()
self.smtp_servers = []
for row in rows:
server = dict(row)
# 日次リセットをチェック
if server['last_reset_date'] != datetime.now().date().isoformat():
self._reset_daily_limit(server['id'])
server['today_sent'] = 0
self.smtp_servers.append(server)
conn.close()
logger.info(f"SMTPサーバーを読み込み: {len(self.smtp_servers)}台")
except Exception as e:
logger.error(f"SMTP設定読み込みエラー: {e}")
# デフォルト設定同じカテゴリーの記事
FlaskテンプレートエンジンJinja2の基礎
はじめに 前回までにFlaskの環境構築とルーティングについて学びました。今回は、Webアプリケーションの見た目を担当する「テンプレートエンジンJinja2」に […]
Flaskルーティングとビュー関数
はじめに 前回の記事でFlask開発環境の構築を完了したら、次はWebアプリケーションの核心である「ルーティング」と「ビュー関数」について学びましょう。ルーティ […]
Flask開発環境の構築 – Windows/Mac対応
はじめに FlaskはPythonでWebアプリケーションを開発するための軽量で柔軟なフレームワークです。今回は、WindowsとMacの両方でFlask開発環 […]