chancy.job module

class chancy.job.ErrorT

Bases: TypedDict

attempt: int

The attempt number of the job when the error occurred.

traceback: str

The error message, typically the traceback of an exception.

class chancy.job.IsAJob(*args, **kwargs)

Bases: Protocol[P, R]

job: Job

class chancy.job.Job(*, func, queue='default', kwargs=<factory>, priority=0, max_attempts=1, scheduled_at=<factory>, limits=<factory>, unique_key=None, meta=<factory>)

Bases: object

A job is an immutable, stateless unit of work that can be pushed onto a Chancy queue and executed elsewhere.

Parameters:

• func (str)

• queue (str)

• kwargs (dict[str, Any] | None)

• priority (int)

• max_attempts (int)

• scheduled_at (datetime)

• limits (list[Limit])

• unique_key (str | None)

• meta (dict[str, Any])

classmethod from_func(func, **kwargs)

Create a job from a function, attempting to determine the function’s importable name automatically.

def hello_world():
    pass

job = Job.from_func(hello_world)

func: str

An importable name for the function that should be executed when this job is run. Ex: my_module.my_function

kwargs: dict[str, Any] | None

The keyword arguments to pass to the job function when it is executed.

limits: list[Limit]

A list of resource limits that should be applied to this job.

max_attempts: int = 1

The maximum number of times this job can be attempted before it is considered failed.

meta: dict[str, Any]

Arbitrary metadata associated with this job instance. Plugins can use this to store additional information during the execution of a job.

pack()

Pack the job into a dictionary that can be serialized and used to recreate the job later.

Return type:

dict

priority: int = 0

The priority of this job. Jobs with higher priority values will be executed before jobs with lower priority values.

queue: str = 'default'

The queue to which this job should be pushed.

scheduled_at: datetime

The time at which this job should be scheduled to run.

unique_key: str | None = None

An optional, globally unique identifier for this job. If provided, only 1 copy of a job with this key will be allowed to run or be scheduled at a time.

classmethod unpack(data)

Unpack a serialized job into a Job instance.

Parameters:

data (dict)

Return type:

Job

with_kwargs(**kwargs)

Return type:

Job

with_limits(limits)

Parameters:

limits (list[Limit])

Return type:

Job

with_max_attempts(max_attempts)

Parameters:

max_attempts (int)

Return type:

Job

with_meta(meta)

Parameters:

meta (dict[str, Any])

Return type:

Job

with_priority(priority)

Parameters:

priority (int)

Return type:

Job

with_queue(queue)

Parameters:

queue (str)

Return type:

Job

with_scheduled_at(scheduled_at)

Parameters:

scheduled_at (datetime)

Return type:

Job

with_unique_key(unique_key)

Parameters:

unique_key (str)

Return type:

Job

class chancy.job.Limit(type_, value)

Bases: object

A limit that can be applied to a job.

Parameters:

• type_ (Type)

• value (int)

class Type(*values)

Bases: Enum

MEMORY = 'memory'

TIME = 'time'

classmethod deserialize(data)

Parameters:

data (dict)

Return type:

Limit

serialize()

Return type:

dict

type_: Type

value: int

class chancy.job.QueuedJob(*, func, queue='default', kwargs=<factory>, priority=0, max_attempts=1, scheduled_at=<factory>, limits=<factory>, unique_key=None, meta=<factory>, id, claim_id=None, created_at, started_at=None, completed_at=None, attempts=0, state=State.PENDING, errors=<factory>, _time_limit_deadline=None)

Bases: Job

A job instance is a job that has been pushed onto a queue and now has stateful information associated with it, such as the number of attempts so far.

Parameters:

• func (str)

• queue (str)

• kwargs (dict[str, Any] | None)

• priority (int)

• max_attempts (int)

• scheduled_at (datetime)

• limits (list[Limit])

• unique_key (str | None)

• meta (dict[str, Any])

• id (UUID)

• claim_id (UUID | None)

• created_at (datetime)

• started_at (datetime | None)

• completed_at (datetime | None)

• attempts (int)

• state (State)

• errors (list[ErrorT])

• _time_limit_deadline (float | None)

class State(*values)

Bases: Enum

FAILED = 'failed'

PENDING = 'pending'

RETRYING = 'retrying'

RUNNING = 'running'

SUCCEEDED = 'succeeded'

attempts: int = 0

The number of times this job has been attempted.

checkpoint()

Raise TimeoutError if the cooperative time limit has expired.

Jobs running in a threaded or sub-interpreter executor should call this method periodically at points where execution can be safely interrupted.

Return type:

None

claim_id: UUID | None = None

Identifies this execution’s claim. Replaced whenever a worker claims the job; cleared by recovery, cancellation, or a manual retry.

completed_at: datetime | None = None

The time at which this job was completed, if it has been completed.

created_at: datetime

The time at which this job was created.

errors: list[ErrorT]

A list of errors that occurred during the execution of this job.

id: UUID

The unique identifier for this job instance.

started_at: datetime | None = None

The time at which this job was started, if it has been started.

state: State = 'pending'

The current state of this job instance.

property time_remaining: float | None

Return the number of seconds remaining before the job’s cooperative time limit expires.

Returns None when the executor has not configured a cooperative time limit for this execution.

classmethod unpack(data)

Unpack a serialized job into a Job instance.

Parameters:

data (dict)

Return type:

QueuedJob

class chancy.job.Reference(identifier)

Bases: object

References a Job in the queue.

This object can be used to retrieve the job instance later, or wait for it to complete. It is returned by the push(), push_many(), and push_many_ex() functions.

Waiting for a job to finish:

async with Chancy("postgresql://localhost/postgres") as chancy:
   ref = await chancy.push(Job.from_func(my_function))
   job = await chancy.wait_for_job(ref)
   print(job.state)  # "succeeded"

Retrieving a job instance:

async with Chancy("postgresql://localhost/postgres") as chancy:
    ref = await chancy.push(Job.from_func(my_function))
    job = await chancy.get_job(ref)
    print(job.state)  # "pending"

Parameters:

identifier (str | UUID)

identifier

chancy.job.job(**options)

A decorator that wraps a function and turns it into a job.

The wrapped function can still be called as normal, but will have an extra job attribute that contains the job instance.

>>> @job()
... def hello_world():
...    return "Hello, world!"
>>> hello_world.job
Job(func=<function my_job at 0x1033041e0>, kwargs={}, ...)
>>> hello_world()
'Hello, world!'

Your decorated function can be pushed like any other job, or you can use the job property to modify its properties before pushing it.

await chancy.push(hello_world)
await chancy.push(hello_world.job.with_queue("low_priority"))

Return type:

Callable[[Callable[[~P], R]], IsAJob[~P, R]]