I tested following situation with partial: from functools import wra

I tested the last comment from Issue <a href="https://github.com/samuelcolvin/arq/issu

cron job with parameter about arq HOT 2 CLOSED

fmatten commented on June 2, 2024

cron job with parameter

from arq.

Comments (2)

joshwilson-dbx commented on June 2, 2024 3

A few things here:

You're dropping the ctx on the floor. It should be something like this instead:

def partial(f, *args, **kwargs):

    @wraps(f)
    async def partial_function(ctx):
        return await f(ctx, *args, **kwargs)

    return partial_function

You're also currently passing in an empty {} which is populating your ctx parameter, hence the empty context. You don't need that anymore.

Finally, it looks like arq uses the job name as a key for its internal lookup tables. By default, it uses your job function's name. So since you're using the same function twice, the first gets stomped.

Try using the name= parameter to cron(). To keep with the default convention, you'll probably want to keep the cron: prefix:

class WorkerSettings:
    cron_jobs = [
        cron(
            partial(my_test_cron_job, store_id="826b2e76-be55-42d0-9022-7f7c26b430cd"),
            hour=at_every_x_hour(1),
            minute=at_every_y_minutes(1),
            run_at_startup=True,
            job_id="id1",
            name="cron:my_test_cron_job_id1",
        ),
        cron(
            partial(my_test_cron_job, store_id="33beef9c-dfd3-42b7-a78b-706f29f8fcgt"),
            hour=at_every_x_hour(1),
            minute=at_every_y_minutes(1),
            run_at_startup=True,
            job_id="id2",
            name="cron:my_test_cron_job_id2",
        ),
    ]