Python API

pip install cronoq            # embedded scheduler (module: cronoq)
pip install "cronoq[client]"  # + cronoq.client: gRPC client and remote worker for cronoq-server

Wheels use the stable ABI (abi3, CPython 3.14+) for Linux, macOS and Windows, on x86_64 and arm64. The client extra needs grpcio, which has no prebuilt wheel for Windows on arm64, so there pip tries to build it from source. Type stubs ship with the package (py.typed). Durations are seconds (float); absolute times accept a datetime or Unix seconds. Invalid durations or configuration (negative, NaN, out of range) raise ValueError.

cronoq.Scheduler

Scheduler(pools: Sequence[Pool], resources: Sequence[Resource] | None = None,
          emit_transitions: bool = False, *, shutdown_grace: float = 5.0,
          event_capacity: int = 1024, store: SqliteStore | None = None,
          codec: Codec | None = None)

A scheduler is bound to the event loop that starts it. If that loop closes without a shutdown, the scheduler stops and further calls raise SchedulerError; calls before start() raise RuntimeError.

Task options

Accepted by submit, enqueue and Task(...).

Strings must not contain NUL characters.

Status events

sched.events() returns an EventStream, an async iterator of StatusEvent (id, pool, key, status, attempt, at in Unix seconds, message).

async with sched.events() as events:
    async for e in events:
        print(e.id, e.status, e.message)

• Each stream buffers up to event_capacity events. When the consumer falls behind, the oldest are dropped and counted in events.lagged. With on_lag="raise" the next iteration raises EventsLagged (.skipped); iterating again continues with the buffered events.
• The stream ends (StopAsyncIteration) when the scheduler shuts down or the stream is closed (close(), await aclose(), or leaving async with). closed tells whether it ended.

cronoq.SqliteStore

SqliteStore(path, *, lease_ttl=10.0, lease_renew=2.0, lock_poll=0.5, retention=3600.0,
            full_sync=True)

Durable state in a local SQLite file, created with owner-only permissions. Submissions, cancellations and schedule changes return only once they are stored. After a restart, queued and interrupted tasks run again (an attempt that was running counts as used), recurring schedules resume, and results of finished tasks stay available through handle().

With a store, payloads and results are encoded by the scheduler's codec:

• cronoq.JsonCodec (the default) stores JSON. Payloads it cannot encode (bytes, sets, arbitrary objects) are refused at submit with TypeError, before anything is stored. Decoding JSON never runs code, so a tampered file cannot execute anything in your process.
• Pass your own codec for other types, for example msgpack. A pickle codec works, but anyone who can write the file can then run code in your process.
• A result the codec cannot encode is not stored (a warning is logged on the cronoq logger); the task still succeeds.

cronoq.Pool

Pool(name, executor, *, min=1, max=10, initial=None, autoscale=None, lease_ttl=None,
     max_pending=None, overflow="wait", retry=None, unknown_dependency="reject",
     await_timeout=None, default_duration=1.0, reservable_fraction=0.5,
     tenant_weights=None, fair_share=False)

A sync executor cannot be interrupted: on timeout or cancellation the attempt is aborted, but the function keeps running until it returns. Check ctx.is_cancelled() in long-running sync executors.

Other types

cronoq.__version__ is the package version as a string.

Exceptions

All live in cronoq (and cronoq.errors), derive from CronoqError and can be pickled.

Any other exception from an executor is a retryable failure. See Errors and status codes.

cronoq.client

The gRPC client and remote worker for cronoq-server, speaking worker protocol version 3 (cronoq.client.PROTOCOL_VERSION). Both can be created outside a running loop; they connect on first use.

Client(target, *, token=None, tls=False, credentials=None, options=None)

target is host:port. tls=True uses system roots; pass grpc.ssl_channel_credentials(...) as credentials for a private CA or mTLS. Use it as async with Client(...) as c or call await c.close().

Result types (frozen dataclasses):

Failed calls raise ClientError: .code is the lower-case gRPC status name and .retryable is true for unavailable and resource_exhausted. A watch stream ends with ClientError unavailable when the server stops serving and data_loss when the consumer fell behind: reconnect and reconcile with get or wait.

task(payload=None, **options)

Builds a TaskSpec for submit_batch and add_recurring. Options: pool, kind, priority, key, deadline_in / deadline_at, start_in / start_at, tolerance, on_miss, duration_hint, timeout, retry (a RetryPolicy or a dict), after, deps ((ref, "on_success") tuples or dicts), resources, tags, labels (worker labels the task requires), tenant, content_type.

Payloads are encoded by type:

content_type= overrides the content type. Worker outputs are encoded the same way (without an override). Result.json() returns None for an empty output.

Worker(target, pool, handler, *, slots=1, token=None, name="", tls=False, credentials=None, reconnect=True, backoff=(0.1, 5.0), grace=10.0, options=None, labels=None, insecure_redirect_http=False)

handler(lease) is an async def receiving a Lease: id, task_id, attempt, payload, content_type, kind, key, tags, timeout, deadline (Unix seconds), pool, priority, lease_ttl, cancelled (an asyncio.Event), .json(), .text(). It returns the output, or raises:

The worker heartbeats each running lease every lease_ttl / 3. When the server aborts an attempt (timeout, cancellation, lease expiry) the handler's task is cancelled and lease.cancelled is set. A handler that swallows CancelledError keeps running and may overlap with the retry of the same task. When the server announces Shutdown (restart or leadership change) the worker reconnects; attempts in flight are retried by the server. When another shard serves the pool, the worker connects to that shard's endpoint, keeping its credentials (insecure_redirect_http=True allows plain http:// to other hosts; see Sharding).

cronoq.compat

A synchronous, thread-safe API for code written against the ElasticPriorityScheduler prototype: ElasticPriorityScheduler(initial_workers=1, max_workers=10, scale_up_threshold=0.7, scale_down_threshold=0.9, worker_cost_per_second=0.001) with start(), stop(), submit(Task) -> bool, get_metrics() -> QualityMetrics, get_cost(), completed_count, completed_tasks, pending_tasks, and the types Task, Priority, ScheduledTask, QualityMetrics, ScalingConfig. See Migrating.