Skip to content

Python API

Application

Potatoq 

Potatoq(
    main: str | None = None,
    broker: str | None = None,
    backend: str | None = None,
    *,
    include: Iterable[str] | None = None,
    config_source: Any = None,
    set_as_current: bool = True,
    task_cls: type[Task] | str | None = None,
    namespace: str | None = None,
    autofinalize: bool = True,
    **kwargs: Any,
)

The task queue application.

Potatoq("proj") with no other arguments works: the broker comes from $POTATOQ_BROKER_URL (or $CELERY_BROKER_URL), Django's database when running under Django, or a local SQLite file for development.

broker property 

broker: Broker

backend property 

backend: Broker | None

Where results are stored (the broker itself unless result_backend is set).

task 

task(*args: Any, **opts: Any) -> Any

Decorator: @app.task or @app.task(bind=True, ...).

send_task 

send_task(
    name: str,
    args: Any = None,
    kwargs: dict[str, Any] | None = None,
    countdown: float | None = None,
    eta: Any = None,
    task_id: str | None = None,
    **options: Any,
) -> AsyncResult

Enqueue a task by name, without needing its code.

config_from_object 

config_from_object(
    obj: Any,
    silent: bool = False,
    force: bool = False,
    namespace: str | None = None,
) -> None

autodiscover_tasks 

autodiscover_tasks(
    packages: Any = None,
    related_name: str = "tasks",
    force: bool = False,
) -> None

Import <package>.tasks for each package (Django apps when packages is None).

add_periodic_task 

add_periodic_task(
    schedule: Any,
    sig: Any,
    args: Any = (),
    kwargs: Any = (),
    name: str | None = None,
    **opts: Any,
) -> str

Celery's sender.add_periodic_task(10.0, my_task.s(), name="...") idiom.

AsyncResult 

AsyncResult(task_id: str, **kwargs: Any) -> AsyncResult

publish 

publish(
    messages: list[Message],
    connection: Any = None,
    on_commit: bool | None = None,
    using: Any = None,
) -> None

Send messages to the broker.

Inside a database transaction (Django atomic(), an SQLAlchemy session) the messages are sent when it commits and dropped if it rolls back, so a worker never sees a task before the data it needs. When the broker is that database, the rows are written inside the transaction itself, which is the same behaviour without the gap between COMMIT and publishing.

on_commit 

on_commit(fn: Callable[[], Any], using: Any = None) -> None

Run fn after the current transaction commits (or now, if there is none).

shared_task 

shared_task(*args: Any, **options: Any) -> Any

Define a task that isn't bound to a specific app (for reusable apps / Django).

Tasks

Task

Base class for tasks. Usually created with @app.task.

request property 

request: Context

delay 

delay(*args: Any, **kwargs: Any) -> AsyncResult

apply_async 

apply_async(
    args: Any = None,
    kwargs: dict[str, Any] | None = None,
    task_id: str | None = None,
    producer: Any = None,
    link: Any = None,
    link_error: Any = None,
    shadow: str | None = None,
    **options: Any,
) -> AsyncResult

delay_on_commit 

delay_on_commit(*args: Any, **kwargs: Any) -> None

Enqueue when the current database transaction commits (Celery 5.4 API).

With Potatoq this is what delay() already does by default inside a transaction (task_enqueue_on_commit); this method forces it.

adelay async 

adelay(*args: Any, **kwargs: Any) -> AsyncResult

delay for async code: the broker round trip runs in a thread.

apply 

apply(
    args: Any = None,
    kwargs: dict[str, Any] | None = None,
    task_id: str | None = None,
    link: Any = None,
    link_error: Any = None,
    **options: Any,
) -> EagerResult

Run the task in this process and return an EagerResult.

s 

s(*args: Any, **kwargs: Any) -> Signature

si 

si(*args: Any, **kwargs: Any) -> Signature

signature 

signature(
    args: Any = None, *starargs: Any, **starkwargs: Any
) -> Signature

retry 

retry(
    args: Any = None,
    kwargs: dict[str, Any] | None = None,
    exc: BaseException | None = None,
    throw: bool = True,
    eta: datetime | None = None,
    countdown: float | None = None,
    max_retries: int | None = None,
    **options: Any,
) -> Retry

Retry the current task. Raises Retry (or exc once retries are exhausted).

replace 

replace(sig: Any) -> None

Replace this task with sig, which inherits its id, callbacks and chord.

The current task ends and the replacement is enqueued atomically with its acknowledgement.

update_state 

update_state(
    task_id: str | None = None,
    state: str | None = None,
    meta: Any = None,
    **kwargs: Any,
) -> None

before_start 

before_start(task_id: str, args: Any, kwargs: Any) -> None

on_success 

on_success(
    retval: Any, task_id: str, args: Any, kwargs: Any
) -> None

on_failure 

on_failure(
    exc: BaseException,
    task_id: str,
    args: Any,
    kwargs: Any,
    einfo: Any,
) -> None

on_retry 

on_retry(
    exc: BaseException,
    task_id: str,
    args: Any,
    kwargs: Any,
    einfo: Any,
) -> None

after_return 

after_return(
    status: str,
    retval: Any,
    task_id: str,
    args: Any,
    kwargs: Any,
    einfo: Any,
) -> None

Results

AsyncResult 

AsyncResult(
    id: str,
    backend: Any = None,
    task_name: str | None = None,
    app: Potatoq | None = None,
    parent: Any = None,
    ignored: bool = False,
)

The result of a task that was sent to the broker.

state property 

state: str

result class-attribute instance-attribute 

result = info

traceback property 

traceback: str | None

get 

get(
    timeout: float | None = None,
    propagate: bool = True,
    interval: float = 0.5,
    no_ack: bool = True,
    follow_parents: bool = True,
    callback: Any = None,
    on_message: Any = None,
    on_interval: Any = None,
    disable_sync_subtasks: bool = True,
    **kwargs: Any,
) -> Any

aget async 

aget(
    timeout: float | None = None,
    propagate: bool = True,
    **kwargs: Any,
) -> Any

get for async code (waits in a thread, never blocks the event loop).

ready 

ready() -> bool

successful 

successful() -> bool

failed 

failed() -> bool

forget 

forget() -> None

revoke 

revoke(
    connection: Any = None,
    terminate: bool = False,
    signal: Any = None,
    wait: bool = False,
    timeout: float | None = None,
) -> None

GroupResult 

GroupResult(
    id: str | None = None,
    results: list[AsyncResult] | None = None,
    parent: Any = None,
    app: Potatoq | None = None,
    **kwargs: Any,
)

Bases: ResultSet

save 

save(backend: Any = None) -> GroupResult

restore classmethod 

restore(
    id: str, backend: Any = None, app: Potatoq | None = None
) -> GroupResult | None

Workflows

Signature 

Signature(
    task: Any = None,
    args: Any = None,
    kwargs: dict[str, Any] | None = None,
    options: dict[str, Any] | None = None,
    type: Any = None,
    subtask_type: str | None = None,
    immutable: bool = False,
    app: Potatoq | None = None,
    **ex: Any,
)

Bases: dict

A serializable description of a task call (task name, args, kwargs, options).

delay 

delay(
    *partial_args: Any, **partial_kwargs: Any
) -> AsyncResult

apply_async 

apply_async(
    args: Any = None,
    kwargs: dict[str, Any] | None = None,
    route_name: Any = None,
    **options: Any,
) -> AsyncResult

clone 

clone(
    args: Any = None,
    kwargs: dict[str, Any] | None = None,
    **opts: Any,
) -> Signature

set 

set(
    immutable: bool | None = None, **options: Any
) -> Signature

link(callback: Any) -> Any

link_error(errback: Any) -> Any

freeze 

freeze(
    _id: str | None = None,
    group_id: str | None = None,
    chord: dict[str, Any] | None = None,
    group_index: int | None = None,
    **kwargs: Any,
) -> AsyncResult

Fix the id (_id, or a new one) so the result is known before sending. group_id/group_index/chord make it a group member or chord part; a chain or chord passes them on to the task whose result is its result.

chain module-attribute 

chain = _chain

group 

group(
    *tasks: Any, app: Potatoq | None = None, **options: Any
)

Bases: Signature

Run tasks in parallel. group(a.s(1), a.s(2))() returns a GroupResult.

chord module-attribute 

chord = _chord

Schedules

crontab 

crontab(
    minute: Any = "*",
    hour: Any = "*",
    day_of_week: Any = "*",
    day_of_month: Any = "*",
    month_of_year: Any = "*",
    tz: str | tzinfo | None = None,
    **kwargs: Any,
)

Bases: BaseSchedule

A cron-like schedule. Fields accept ints, iterables or cron syntax strings.

day_of_week uses 0 (or 7) = Sunday and accepts names ("mon-fri"). As in standard cron, when both day_of_month and day_of_week are restricted a day matches if either matches.

from_string classmethod 

from_string(
    expr: str, tz: str | tzinfo | None = None
) -> crontab

next_after 

next_after(
    after: datetime, tz: tzinfo | None = None
) -> datetime

schedule 

schedule(
    run_every: float | timedelta,
    relative: bool = False,
    **kwargs: Any,
)

Bases: BaseSchedule

Run every run_every (seconds or timedelta), aligned to the epoch.

Testing

drain 

drain(
    app: Potatoq,
    queues: list[str] | None = None,
    *,
    include_scheduled: bool = True,
    max_tasks: int = 10000,
    raise_on_failure: bool = False,
) -> list[DrainedTask]

Run queued tasks in this process until the queues are empty.

include_scheduled also runs tasks with an ETA/countdown and retries (time is skipped ahead, they are not waited for), on every broker but RabbitMQ, whose delays live in the broker (it warns there).

DrainedTask dataclass 

DrainedTask(
    id: str,
    name: str,
    args: list[Any],
    kwargs: dict[str, Any],
    state: str,
    result: Any,
    exception: BaseException | None,
)

Exceptions

Exceptions, named after their celery.exceptions counterparts.

PotatoqError

Bases: Exception

Base class for all Potatoq errors.

ImproperlyConfigured

Bases: PotatoqError

Potatoq is configured incorrectly.

OperationalError

Bases: PotatoqError

The broker couldn't be reached (or refused the operation) while sending tasks.

Raised by .delay()/apply_async() whatever the broker, like kombu's OperationalError in Celery. The broker's own exception is __cause__.

EnqueueAfterCommitError 

EnqueueAfterCommitError(
    message: str, task_ids: list[str] | None = None
)

Bases: OperationalError

Tasks deferred to the end of a transaction couldn't be sent after it committed.

The transaction's data is saved, but these tasks are lost (each is logged at ERROR on the potatoq logger with its id). Only with Redis and RabbitMQ: database brokers write tasks inside the transaction.

NotRegistered

Bases: PotatoqError, KeyError

The task is not registered with the app.

TaskError

Bases: PotatoqError

Base class for errors raised from inside tasks.

Retry 

Retry(
    message: str | None = None,
    exc: BaseException | None = None,
    when: float | datetime | None = None,
    **kwargs: Any,
)

Bases: TaskError

Raised by Task.retry() to signal that the task should be retried.

MaxRetriesExceededError 

MaxRetriesExceededError(
    *args: Any,
    task_args: Any = None,
    task_kwargs: Any = None,
    **kwargs: Any,
)

Bases: TaskError

The task has been retried the maximum number of times.

Ignore

Bases: TaskError

Raise from a task to stop processing without recording a state.

Replace 

Replace(sig: Any)

Bases: Ignore

Raised by Task.replace(); carries the replacement signature.

Reject 

Reject(reason: Any = None, requeue: bool = False)

Bases: TaskError

Raise from a task to reject the message (optionally requeueing it).

TimeoutError

Bases: PotatoqError, TimeoutError

Waiting for a result timed out.

TimeLimitExceeded

Bases: PotatoqError

The hard time limit of a task was exceeded and the worker process was killed.

SoftTimeLimitExceeded

Bases: PotatoqError

Raised inside a task when its soft time limit is exceeded.

Deliberately inherits from Exception (via PotatoqError) like Celery, so tasks can catch it to clean up.

WorkerLostError

Bases: PotatoqError

The worker process executing the task exited unexpectedly.

TaskRevokedError

Bases: PotatoqError

The task was revoked and will not run.

ResultBackendDisabled

Bases: ImproperlyConfigured

Results were requested but no result backend is available.

DeadLettered

Bases: PotatoqError

The task was delivered too many times without completing and was dead-lettered.

RemoteError 

RemoteError(
    exc_type: str,
    exc_message: str,
    exc_module: str | None = None,
)

Bases: PotatoqError

Stand-in for an exception raised by a task that could not be reconstructed locally.

ChordError

Bases: PotatoqError

A task in a chord's header failed, so the chord callback can't run.

WorkerTerminate

Bases: BaseException

Raised inside a running task when the worker is forced to stop (shutdown timeout). The task is requeued, not failed. BaseException so tasks don't catch it.