rq のジョブに独自のメタ情報を付与する方法

• 概要
• 環境
• サンプルコード
• 最後に
• 参考サイト

概要

ジョブ登録時に meta 引数を指定することができます
辞書で好きな値を指定できます
pickle するので pickle 可能なオブジェクトにしましょう

環境

• Python 3.11.3
• flask-rq2 18.3
• rq-scheduler 0.13.1

サンプルコード

from flask import Flask, request
from flask_rq2 import RQ

app = Flask(__name__)
app.config["RQ_REDIS_URL"] = "redis://localhost:6379/0"

rq = RQ(app)


def hello():
    raise Exception()


@app.route("/")
def get_jobs():
    # 検索する id をクエリストリングで渡せる
    id = request.args.get("id")
    scheduler = rq.get_scheduler(queue="bar")
    jobs_ = scheduler.get_jobs()
    # id が設定されていない場合は全件取得、そうでない場合は meta 情報を使って id 検索
    if id is None:
        jobs = jobs_
    else:
        jobs = [
            j for j in jobs_ if j.meta.get("id") is not None and j.meta.get("id") == id
        ]
    return [
        {
            "id": job.id,
            "meta": job.meta.get("id"),
        }
        for job in jobs
    ]


@app.route("/register")
def register_job():
    # id を meta 情報として設定する
    id = request.args.get("id")
    scheduler = rq.get_scheduler(queue="bar")
    job = scheduler.cron(
        "*/1 * * * *",
        func=hello,
        args=[],
        meta={"id": id},  # meta 属性に辞書を設定可能
    )
    return job.get_id()

最後に

ジョブを特定する際に使えると思います
ただ meta 情報で直接検索する機能がないので全件取得してから検索するしかないので redis のデータが膨大になる場合にはパフォーマンスなどの注意が必要です

参考サイト

• https://github.com/rq/rq-scheduler