Glinteco事例研究：ロック機構を用いたCeleryタスクにおけるレート制限の実装

class RateLimitTask(Task):
    LOCK_KEY_TEMPLATE = "celery:rate_limit_task-{0}"
    RATELIMIT = {
        "tasks": 4,
        "time": 60,  # 秒
    }

    def __call__(self, *args, **kwargs):
        rate_limit = self.RATELIMIT
        custom_rate_limit = getattr(self, "custom_rate_limit", None)
        if custom_rate_limit:
            rate_limit = custom_rate_limit

        base_task_id = self.generate_lock_id(*args, **kwargs)
        first_task_at_lock_id = f"{base_task_id}-first_task_at"
        count_tasks_lock_id = f"{base_task_id}-count_tasks"

        first_task_at = cache.get(first_task_at_lock_id)
        count_tasks = cache.get(count_tasks_lock_id, 0)

        now = datetime.now()

        if not first_task_at:
            first_task_at = now
            count_tasks = 0
            cache.set(first_task_at_lock_id, first_task_at, rate_limit["time"])
            cache.set(count_tasks_lock_id, count_tasks, rate_limit["time"])

        time_from_first_task = (now - first_task_at).total_seconds()
        valid_to_run = (
            time_from_first_task < rate_limit["time"]
            and count_tasks < rate_limit["tasks"]
        )
        if valid_to_run:
            cache.incr(count_tasks_lock_id)
            return self.run(*args, **kwargs)

        logger.warning(
            f"レート制限に達しました: {count_tasks}/{rate_limit['tasks']} "
            f"{rate_limit['time']}秒あたり、task_id: {base_task_id}"
        )
        raise Ignore()

    # ... (その他のメソッドはUnique Taskの実装と同一のままです)

from celery import Celery

# Celeryアプリケーションの初期化
app = Celery("example")
app.config_from_object("celeryconfig")

# GlintecoのRateLimitTaskを使用したサンプルタスクの定義
@app.task(base=RateLimitTask)
def process_data(data_id):
    """
    GlintecoのRateLimitTaskを使用する例
    """
    # data_idに基づいてデータ処理を実行します
    pass