procrastinate

When multiple layers of middleware are used, they form a chain. The execution order from outermost to innermost is:

1. Worker middleware (if any)
2. Worker-wide task middleware (in the order provided in the list)
3. Per-task middleware (in the order provided in the list)
4. The task function itself

The first middleware in a list executes its 'before' code first and its 'after' code (the code following call_next()) last.

A queueing_lock limits the number of jobs in the todo status (the queue). It allows exactly one job to be in the todo state for a given lock string.

However, it does not prevent multiple jobs from being in the doing status (currently being processed by workers). To ensure only one job runs at a time while also limiting the queue size, you must combine queueing_lock with a standard execution lock.

The Procrastinate worker manages Django database connections at task boundaries (before and after every task) by calling close_old_connections() and reset_queries(). This ensures connections are cleaned up and CONN_MAX_AGE is respected.

Important: If a single task performs a long stretch of non-database work, the connection opened at the start of the task will remain open and occupy a connection slot. To prevent idle connections from being dropped by the database, manually call close_old_connections() from within your task after your database work is complete.

from django.db import close_old_connections

@app.task
def long_task():
    do_early_db_work()
    close_old_connections()  # release the connection before the long idle stretch
    do_hours_of_non_db_work()
    do_late_db_work()  # reconnects fresh

A queueing_lock ensures that a specific set of jobs can never appear more than once in the queue simultaneously. This is useful for tasks like cleanup where multiple pending jobs add little value if the queue is already backed up.

When a queueing_lock is active, attempting to defer another task with the same lock string will raise an AlreadyEnqueued exception. You can choose to catch and ignore this exception to implement a 'skip if already queued' pattern.

# Using a lock on a per-call basis
my_task.configure(queueing_lock="arbitrary_string").defer(a=1)
my_task.configure(queueing_lock="arbitrary_string").defer(a=2)

Procrastinate provides two types of locking mechanisms that can be attached to a task via Task.configure to control execution flow:

• Lock: Guarantees that jobs with the same lock are executed sequentially and in the order they were deferred. No two jobs sharing the same lock can run at the same time.
• Queueing Lock: Prevents multiple jobs with the same queueing lock from waiting in the queue simultaneously. This is useful for limiting the number of pending jobs of a certain type.

When using external connections with Procrastinate, be aware of the following:

• NOTIFY Latency: The database NOTIFY signal that wakes workers is only sent when you commit() your transaction. This means workers will not see a new job or an abort request until the transaction is finalized.
• Transaction Errors: Errors during job insertion (like lock violations) are wrapped in Procrastinate's exception hierarchy but can abort your entire transaction. If you want to isolate the job deferral from other operations, use database savepoints.
• Type Validation: Procrastinate does not validate the connection type at the library level; passing an incorrect type will result in a runtime error from your database driver.

Since periodic schedules are typically defined at task definition time, dynamic scheduling (where schedules change at runtime) requires a different approach:

1. Store desired schedules in a database.
2. Create a single 'manager' periodic task that runs at the highest required frequency (e.g., every minute).
3. In the manager task, read the configuration from your database.
4. Determine if any tasks need to be run for the current timestamp (always use the provided timestamp argument rather than time.time() to account for delays).
5. Defer the corresponding tasks manually from within the manager task.

In the context of Procrastinate, the Schema refers to the collection of database objects (tables, relations, indexes, procedures, etc.) required for the task queue to function.

• Applying the schema: The process of installing these objects into your PostgreSQL database.
• Migration: An evolution or modification of the schema (e.g., changing table structures or procedures).

Note: This is distinct from PostgreSQL 'schemas', which are namespaces within a database.

Procrastinate provides different connectors depending on whether you are running a worker or just deferring jobs, and whether your application is asynchronous or synchronous.

Requirements

• To run a worker: You must use an asynchronous connector.
• To defer jobs: You can use either asynchronous or synchronous connectors.

Available Connectors

Asynchronous Connectors (Required for Workers)

• PsycopgConnector: Based on psycopg v3.
• AiopgConnector: Based on aiopg.

Synchronous Connectors (Deferring jobs only)

• SyncPsycopgConnector: Based on psycopg v3.
• Psycopg2Connector: Based on psycopg2.
• SQLAlchemyPsycopg2Connector: Specialized for SQLAlchemy applications. Use this to share your existing SQLAlchemy connection pool. Note: This can only be used for deferring jobs, not for running workers.