API Reference

Protocols

ProgressTrackerProtocol

source

Protocol for tracking and broadcasting task progress.

Implementations must be safe for concurrent use — multiple coroutines may call update on the same task_id simultaneously, and multiple consumers may subscribe to the same task.

The subscribe method returns a live AsyncIterator that yields a new ProgressSnapshot every time update, complete, or fail is called for the given task. The iterator terminates automatically when the task reaches a terminal state (COMPLETE or FAILED).

Example

async def export_data(tracker: ProgressTrackerProtocol) -> None:
    records = await fetch_all()
    total = len(records)
    for i, record in enumerate(records, start=1):
        await process(record)
        await tracker.update("export-1", i, total, f"Row {i}/{total}")
    await tracker.complete("export-1", "Export finished")

update

async def update(
    task_id: str,
    current: int,
    total: int,
    message: str = ''
) -> None

source

Record incremental progress for a task.

Parameters

Parameter	Type	Description
`task_id`	str	Unique identifier for the task.
`current`	int	Units completed so far.
`total`	int	Total units to process (0 = unknown).
`message`	str	Optional human-readable status message.

complete

async def complete(
    task_id: str,
    result: str = ''
) -> None

source

Mark a task as successfully completed.

Closes all active subscriptions for task_id after broadcasting the terminal ProgressSnapshot.

Parameters

Parameter	Type	Description
`task_id`	str	Unique identifier for the task.
`result`	str	Optional human-readable completion message.

fail

async def fail(
    task_id: str,
    error: str
) -> None

source

Mark a task as failed.

Closes all active subscriptions for task_id after broadcasting the terminal ProgressSnapshot.

Parameters

Parameter	Type	Description
`task_id`	str	Unique identifier for the task.
`error`	str	Description of the failure.

get

async def get(task_id: str) -> ProgressSnapshot | None

source

Return the current progress state for a task.

Parameters

Parameter	Type	Description
`task_id`	str	Unique identifier for the task.

Returns

Type	Description
ProgressSnapshot | None	The most recent ProgressSnapshot, or ``None`` if the task is not known to this tracker.

subscribe

def subscribe(task_id: str) -> AsyncIterator[ProgressSnapshot]

source

Subscribe to live progress updates for a task.

Returns an async iterator that yields a new ProgressSnapshot each time the task’s state changes. The iterator stops automatically when the task reaches a terminal state (COMPLETE or FAILED).

If the task is already in a terminal state when subscribe is called, the iterator yields the final snapshot once and then stops.

Parameters

Parameter	Type	Description
`task_id`	str	Unique identifier for the task to observe.

Returns

Type	Description
AsyncIterator[ProgressSnapshot]	An AsyncIterator of ProgressSnapshot objects.

Example

async for snap in tracker.subscribe("job-42"):
    print(f"{snap.percent:.1f}% — {snap.message}")

---

TaskExecutorProtocol

source

Protocol for task executor implementations.

Executors process tasks from a queue by dispatching them to registered handlers.

Example

class TaskExecutorProtocol:
    async def execute(self, task: Task) -> Any:
        handler = self._handlers.get(task.name)
        if handler:
            return await handler(task.payload)
        raise UnknownTaskError(task.name)

execute

async def execute(task: Any) -> Any

source

Execute a task using the registered handler.

Parameters

Parameter	Type	Description
`task`	Any	Task to execute.

Returns

Type	Description
Any	Task execution result.

Raises

Exception	Description
UnknownTaskError	If no handler registered for task.

register_handler

def register_handler(
    task_name: str,
    handler: Any
) -> None

source

Register a handler for a task type.

Parameters

Parameter	Type	Description
`task_name`	str	Name of the task type.
`handler`	Any	Async callable to handle the task.

get_handler

def get_handler(task_name: str) -> Any | None

source

Get the handler for a task type.

Parameters

Parameter	Type	Description
`task_name`	str	Name of the task type.

Returns

Type	Description
Any | None	Handler if registered, None otherwise.

---

TaskQueueProtocol

source

Protocol for task queue implementations.

This defines the contract for background task queues (Redis-based, PostgreSQL-based, in-memory, etc.).

Example

class RedisTaskQueue:
    async def enqueue(self, task: Task) -> str:
        task_id = str(uuid4())
        await self._redis.rpush("tasks", task.to_json())
        return task_id

    async def dequeue(self) -> Task | None:
        data = await self._redis.lpop("tasks")
        return Task.from_json(data) if data else None

enqueue

async def enqueue(task: Any) -> Result[str, TaskQueueError]

source

Add a task to the queue.

Parameters

Parameter	Type	Description
`task`	Any	Task instance to enqueue.

Returns

Type	Description
Result[str, TaskQueueError]	Ok(task_id) on success; Err(TaskQueueError) if the queue reports a recoverable failure (e.g. full queue, invalid payload). Only infrastructure failures (lost connection) are raised as exceptions.

dequeue

async def dequeue() -> Any | None

source

Remove and return the next task.

Returns

Type	Description
Any | None	Next Task or None if queue is empty.

get_task_count

async def get_task_count() -> int

source

Get the number of tasks in the queue.

Returns

Type	Description
int	Number of pending tasks.

clear

async def clear() -> None

source

Clear all tasks from the queue.

get_job

async def get_job(job_id: str) -> Any | None

source

Get a job by ID.

Parameters

Parameter	Type	Description
`job_id`	str	Unique job identifier.

Returns

Type	Description
Any | None	Job data or None if not found.

delete_job

async def delete_job(job_id: str) -> bool

source

Delete a job from the queue.

Parameters

Parameter	Type	Description
`job_id`	str	Unique job identifier.

Returns

Type	Description
bool	True if the job was deleted.

count

async def count(
    queue_name: str,
    status: Any | None = None
) -> int

source

Count queued jobs.

Parameters

Parameter	Type	Description
`queue_name`	str	Queue to inspect.
`status`	Any | None	Optional status filter.

Returns

Type	Description
int	Number of matching jobs.

ack

async def ack(task_id: str) -> None

source

Acknowledge successful processing of a dequeued task.

Signals that the task has been processed successfully and can be permanently discarded from any in-flight tracking.

Parameters

Parameter	Type	Description
`task_id`	str	ID of the task to acknowledge.

nack

async def nack(
    task_id: str,
    requeue: bool = True
) -> None

source

Negative-acknowledge a dequeued task.

Signals that the task could not be processed. The task is optionally requeued for another processing attempt.

Parameters

Parameter	Type	Description
`task_id`	str	ID of the task to negative-acknowledge.
`requeue`	bool	If True, return the task to the queue for retry. If False, discard the task permanently.

close

async def close() -> None

source

Close the queue connection and cleanup resources.

---

Classes

BackgroundTaskManager

source

Tracks and shuts down background asyncio tasks.

Inject as a singleton from any DI provider. Call shutdown() during application stop to cancel all in-flight tasks.

Example

manager = BackgroundTaskManager()
task = manager.track(some_coroutine())
# … later …
await manager.shutdown(timeout=30.0)

manager = BackgroundTaskManager()
task = manager.track(some_coroutine())
# … later …
await manager.shutdown(timeout=30.0)

__init__

def __init__() -> None

source

track

def track(coro: Awaitable[T]) -> asyncio.Task[T]

source

Schedule coro as an asyncio task and track the handle.

The task is removed from the tracked set automatically via a done_callback so completed tasks don’t accumulate.

Parameters

Parameter	Type	Description
`coro`	Awaitable[T]	Any awaitable to wrap in a task.

Returns

Type	Description
asyncio.Task[T]	The created asyncio.Task.

track_named

def track_named(
    name: str,
    coro: Awaitable[T]
) -> asyncio.Task[T]

source

Like track but attach a human-readable name for logs.

Named tasks appear in shutdown() timeout warnings so operators can identify which work is still running at stop time.

Parameters

Parameter	Type	Description
`name`	str	Label surfaced in observability/log output.
`coro`	Awaitable[T]	Any awaitable to wrap in a task.

Returns

Type	Description
asyncio.Task[T]	The created asyncio.Task.

pending_count

property pending_count() -> int

source

Number of tasks that have not yet completed.

shutdown

async def shutdown(timeout: float = 30.0) -> None

source

Cancel all pending tracked tasks and await them.

After cancellation each task is given up to timeout seconds collectively to finish (honour CancelledError and finalise any finally blocks). Tasks that do not stop within the timeout are logged by name; no exception is raised.

Parameters

Parameter	Type	Description
`timeout`	float	Seconds to wait for all tasks to finish after cancellation. Default 30 seconds.

---

BranchStep

source

Branch based on a condition evaluated against the input.

Example

workflow = BranchStep( condition=lambda data: data[“amount”] > 1000, if_true=TaskStep(“manual_review”, manual_review), if_false=TaskStep(“auto_approve”, auto_approve), )

__init__

def __init__(
    condition: Callable[[Any], bool],
    if_true: TaskStep,
    if_false: TaskStep
) -> None

source

execute

async def execute(input_data: Any = None) -> Result[WorkflowResult, WorkflowError]

source

Evaluate the condition and execute either the true or false branch.

---

CacheBackendProgressStore

source

Distributed progress store backed by the platform ``CacheBackendProtocol``.

Stores serialised ProgressInfo objects in the configured cache backend so that progress is visible to all worker processes. TTL-based expiry is delegated to the backend.

Parameters

Parameter	Type	Description
`cache`		Platform cache backend (Redis, Memcached, …).
`ttl`		Seconds before a progress entry expires. Defaults to 3600 (1 hour).

__init__

def __init__(
    cache: CacheBackendProtocol,
    *,
    ttl: int = 3600
) -> None

source

save

async def save(info: ProgressInfo) -> None

source

Persist a ProgressInfo entry to the cache backend.

Parameters

Parameter	Type	Description
`info`	ProgressInfo	Progress state to save.

get

async def get(job_id: str) -> ProgressInfo | None

source

Return the ProgressInfo for the given job ID, or None if not found.

Parameters

Parameter	Type	Description
`job_id`	str	Unique identifier for the job.

delete

async def delete(job_id: str) -> None

source

Remove the progress entry for the given job ID.

Parameters

Parameter	Type	Description
`job_id`	str	Unique identifier for the job.

list_active

async def list_active() -> list[ProgressInfo]

source

Return all active (incomplete) progress entries.

Note This implementation scans the key prefix via get, which may be slow on large datasets. Consider using a Redis-specific implementation with SCAN for production at scale.

Returns

Type	Description
list[ProgressInfo]	List of incomplete ProgressInfo.

---

CacheBackendResultStore

source

Distributed result store backed by the platform ``CacheBackendProtocol``.

Stores serialised JobResult objects in the configured cache backend so that results are visible to all worker processes. TTL-based expiry is delegated to the backend.

Parameters

Parameter	Type	Description
`cache`		Platform cache backend (Redis, Memcached, …).
`ttl`		Seconds before a result entry expires. Defaults to 3600 (1 hour).
`poll_interval`		Seconds between polls in wait. Defaults to 0.5.

__init__

def __init__(
    cache: CacheBackendProtocol,
    *,
    ttl: int = 3600,
    poll_interval: float = 0.5
) -> None

source

store

async def store(
    job_id: str,
    result: JobResult
) -> None

source

Persist a job result in the cache backend with TTL.

get

async def get(job_id: str) -> JobResult | None

source

Return the stored result, or None if absent or expired.

delete

async def delete(job_id: str) -> bool

source

Remove the result from the cache and return True if it existed.

wait

async def wait(
    job_id: str,
    *,
    timeout: float = 30.0,
    poll_interval: float | None = None
) -> JobResult | None

source

Poll the cache until the result appears or the timeout elapses.

cleanup_expired

async def cleanup_expired() -> int

source

No-op: the cache backend handles TTL expiry automatically.

---

CronExpression

source

Wrapper for cron expression parsing.

Provides a unified interface for cron expression handling with optional croniter dependency.

Note This class requires the croniter optional dependency. See lexigram.tasks.scheduling.cron for installation instructions. Instantiation raises :exc:ImportError when croniter is absent.

Example

cron = CronExpression("0 9 * * 1-5")  # Weekdays at 9am
next_run = cron.get_next()

__init__

def __init__(expression: str)

source

Initialize cron expression

Parameters

Parameter	Type	Description
`expression`	str	Cron expression string

Raises

Exception	Description
ImportError	If croniter is not installed
ValueError	If expression is invalid

get_next

def get_next(base_time: float | None = None) -> float

source

Get next execution time

Parameters

Parameter	Type	Description
`base_time`	float | None	Base time (defaults to current time)

Returns

Type	Description
float	Unix timestamp of next execution

get_prev

def get_prev(base_time: float | None = None) -> float

source

Get previous execution time

Parameters

Parameter	Type	Description
`base_time`	float | None	Base time (defaults to current time)

Returns

Type	Description
float	Unix timestamp of previous execution

---

DeadLetterQueue

source

Manages failed jobs for inspection, retry, and cleanup.

__init__

def __init__(
    *,
    max_size: int = 10000,
    retention_hours: float = 168
) -> None

source

add

def add(
    job: JobProtocol,
    error: str,
    traceback: str | None = None,
    **metadata: Any
) -> None

source

Add a failed job to the DLQ.

Parameters

Parameter	Type	Description
`job`	JobProtocol	The failed job.
`error`	str	Error message.
`traceback`	str | None	Full traceback string. **metadata: Additional metadata.

get

def get(job_id: str) -> FailureRecord | None

source

Get a specific failure record.

list_failed

def list_failed(
    *,
    limit: int = 50,
    job_name: str | None = None
) -> list[FailureRecord]

source

List failed jobs, optionally filtered by job name.

Parameters

Parameter	Type	Description
`limit`	int	Maximum records to return.
`job_name`	str | None	Filter by job type name.

Returns

Type	Description
list[FailureRecord]	List of failure records (newest first).

retry

def retry(job_id: str) -> JobProtocol | None

source

Remove a job from DLQ and prepare it for retry.

Returns the job with status reset to PENDING, or None if not found.

retry_all

def retry_all() -> list[JobProtocol]

source

Retry all failed jobs. Returns list of jobs to re-enqueue.

remove

def remove(job_id: str) -> bool

source

Permanently remove a failure from the DLQ.

purge

def purge(
    *,
    older_than_hours: float | None = None
) -> int

source

Purge old failures.

Parameters

Parameter	Type	Description
`older_than_hours`	float | None	Remove records older than this. Defaults to retention_hours.

Returns

Type	Description
int	Number of records purged.

clear

def clear() -> int

source

Clear all DLQ entries. Returns count cleared.

size

property size() -> int

source

Return the number of currently tracked failure records.

get_stats

def get_stats() -> dict[str, Any]

source

Get DLQ statistics.

---

DistributedLockProtocol

source

In-memory lock for preventing duplicate task execution within a single process.

Note: This is NOT a distributed lock. For true distributed locking across multiple processes/servers, use a Redis-based or database-based lock implementation.

Example

lock_manager = LockManager()
async with lock_manager.acquire("user:123:sync", timeout=60):
    await sync_user_data(user_id=123)

__init__

def __init__(
    key: str,
    timeout: float,
    locks_dict: dict[str, tuple[float, asyncio.Lock]],
    locks_lock: asyncio.Lock
)

source

Initialize in-memory lock.

Parameters

Parameter	Type	Description
`key`	str	Unique lock key
`timeout`	float	Lock timeout in seconds (auto-release)
`locks_dict`	dict[str, tuple[float, asyncio.Lock]]	Shared locks dictionary (from LockManager)
`locks_lock`	asyncio.Lock	Shared lock for accessing locks_dict (from LockManager)

acquire

async def acquire() -> bool

source

Acquire the lock.

Returns

Type	Description
bool	True if lock was acquired, False if already held

try_acquire

async def try_acquire() -> bool

source

Try to acquire lock without waiting.

Returns

Type	Description
bool	True if acquired, False if already held

release

async def release() -> None

source

Release the lock.

---

ExecutionRecord

source

Record of a single task execution.

to_dict

def to_dict() -> dict[str, Any]

source

Serialize the execution record to a plain dictionary.

---

FailureRecord

source

Record of a failed job in the DLQ.

to_dict

def to_dict() -> dict[str, Any]

source

Serialize the failure record to a plain dictionary.

from_dict

def from_dict(
    cls,
    data: dict[str, Any]
) -> FailureRecord

source

Deserialize a failure record from a plain dictionary.

Parameters

Parameter	Type	Description
`data`	dict[str, Any]	Dictionary as produced by to_dict.

Returns

Type	Description
FailureRecord	Reconstructed FailureRecord.

---

GlobalRateLimiter

source

System-wide throughput cap applied across all queues.

Delegates to a single TokenBucket for consistent rate enforcement regardless of which queue a task originates from.

Example

limiter = GlobalRateLimiter(rate=100, per=1.0)  # 100 tasks / sec
await limiter.acquire()

limiter = GlobalRateLimiter(rate=100, per=1.0)  # 100 tasks / sec
await limiter.acquire()

__init__

def __init__(
    rate: int,
    per: float = 1.0,
    burst: int | None = None
) -> None

source

Initialise the global rate limiter.

Parameters

Parameter	Type	Description
`rate`	int	Maximum allowed tasks per time period.
`per`	float	Length of the time period in seconds.
`burst`	int | None	Maximum burst capacity (defaults to *rate*).

acquire

async def acquire() -> None

source

Block until a global token is available.

try_acquire

async def try_acquire() -> bool

source

Non-blocking global token acquisition.

Returns

Type	Description
bool	``True`` if a token was obtained; ``False`` if globally rate-limited.

---

HandlerRegistry

source

Registry for task handlers that also satisfies the TaskExecutorProtocol contract.

Maps task names to their handler functions. Handlers can be registered and retrieved by name. The execute method dispatches a task to its registered handler, making this class usable wherever a TaskExecutorProtocol is expected.

Extends Registry for unified introspection and lifecycle hooks.

Example

handler = registry.get("send_email")
```python

__init__

def __init__() -> None

source

Initialize handler registry.

register

def register(
    name: str,
    handler: Callable[Ellipsis, Awaitable[Any]] | None = None
) -> Callable[Ellipsis, Any]

source

Register a task handler.

Can be used as a decorator or called directly.

Parameters

Parameter	Type	Description
`name`	str	Task name.
`handler`	Callable[Ellipsis, Awaitable[Any]] | None	Handler function (if not using as decorator).

Returns

Type	Description
Callable[Ellipsis, Any]	Handler function or decorator.

Example

# As decorator
@registry.register("task_name")
async def my_handler(...):
    pass

# Direct call
registry.register("task_name", my_handler)

register_handler

def register_handler(
    task_name: str,
    handler: Any
) -> None

source

Register a handler for a task type (TaskExecutorProtocol method).

This method satisfies the register_handler requirement defined in lexigram.contracts.infra.tasks.protocols.TaskExecutorProtocol. It is a protocol-mandated entry point that delegates directly to register so that HandlerRegistry can be used wherever a TaskExecutorProtocol is expected without any additional adapter. Do not remove — the protocol contract requires this exact signature.

Parameters

Parameter	Type	Description
`task_name`	str	Name of the task type.
`handler`	Any	Async callable to handle the task.

get_handler

def get_handler(task_name: str) -> Any | None

source

Get the handler for a task type (TaskExecutorProtocol method).

This method satisfies the get_handler requirement defined in lexigram.contracts.infra.tasks.protocols.TaskExecutorProtocol. It is a protocol-mandated entry point that delegates directly to get so that HandlerRegistry can be used wherever a TaskExecutorProtocol is expected without any additional adapter. Do not remove — the protocol contract requires this exact signature.

Parameters

Parameter	Type	Description
`task_name`	str	Name of the task type.

Returns

Type	Description
Any | None	Handler if registered, ``None`` otherwise.

execute

async def execute(task: Any) -> Any

source

Execute a task using the registered handler (TaskExecutorProtocol protocol).

Resolves the handler for task.name, then invokes it with task.args and task.kwargs. Supports both sync and async handlers.

Parameters

Parameter	Type	Description
`task`	Any	Task object with ``name``, ``args``, and ``kwargs`` attributes.

Returns

Type	Description
Any	Result returned by the handler.

Raises

Exception	Description
TaskNotFoundError	If no handler is registered for the task name.

remove

def remove(name: str) -> bool

source

Remove handler from registry.

Parameters

Parameter	Type	Description
`name`	str	Task name.

Returns

Type	Description
bool	True if handler was removed.

list_handlers

def list_handlers() -> list[str]

source

List all registered handler names.

Returns

Type	Description
list[str]	List of task names.

to_dict

def to_dict() -> dict[str, Callable[Ellipsis, Awaitable[Any]]]

source

Get handlers as dictionary.

Returns

Type	Description
dict[str, Callable[Ellipsis, Awaitable[Any]]]	Dictionary of all handlers.

---

InMemoryProgressStore

source

In-memory progress store for development/testing.

__init__

def __init__() -> None

source

save

async def save(info: ProgressInfo) -> None

source

Persist a ProgressInfo entry in the in-memory store.

get

async def get(job_id: str) -> ProgressInfo | None

source

Return the ProgressInfo for the given job ID, or None if not found.

delete

async def delete(job_id: str) -> None

source

Remove the progress entry for the given job ID if it exists.

list_active

async def list_active() -> list[ProgressInfo]

source

Return all progress entries whose current count has not yet reached total.

---

InMemoryProgressTracker

source

Thread-safe, in-memory ProgressTrackerProtocol implementation.

Maintains the latest ProgressSnapshot for each tracked task and fans out every state change to all active subscribers via asyncio.Queue.

Subscriptions are kept alive until the task reaches a terminal state (COMPLETE or FAILED), at which point the iterator closes automatically. Subscribers that call subscribe after the task has already finished receive the terminal snapshot once and then stop.

All public methods are safe to call concurrently from multiple coroutines.

__init__

def __init__() -> None

source

update

async def update(
    task_id: str,
    current: int,
    total: int,
    message: str = ''
) -> None

source

Record incremental progress and broadcast to all subscribers.

Parameters

Parameter	Type	Description
`task_id`	str	Unique identifier for the task.
`current`	int	Units completed so far.
`total`	int	Total units to process (0 means unknown).
`message`	str	Optional human-readable status line.

complete

async def complete(
    task_id: str,
    result: str = ''
) -> None

source

Mark a task as successfully completed and close all subscriptions.

Parameters

Parameter	Type	Description
`task_id`	str	Unique identifier for the task.
`result`	str	Optional human-readable completion message.

fail

async def fail(
    task_id: str,
    error: str
) -> None

source

Mark a task as failed and close all subscriptions.

Parameters

Parameter	Type	Description
`task_id`	str	Unique identifier for the task.
`error`	str	Description of the failure.

get

async def get(task_id: str) -> ProgressSnapshot | None

source

Return the current progress state for a task.

Parameters

Parameter	Type	Description
`task_id`	str	Unique identifier for the task.

Returns

Type	Description
ProgressSnapshot | None	The most recent ProgressSnapshot, or ``None`` if the task has not been seen by this tracker.

subscribe

def subscribe(task_id: str) -> AsyncGenerator[ProgressSnapshot, None]

source

Subscribe to live progress updates for a task.

Returns an async generator that yields one ProgressSnapshot per state change. The generator stops automatically when the task reaches a terminal state. If the task is already finished, the terminal snapshot is yielded once and the generator stops immediately.

Parameters

Parameter	Type	Description
`task_id`	str	Unique identifier for the task to observe.

Returns

Type	Description
AsyncGenerator[ProgressSnapshot, None]	An async generator of ProgressSnapshot objects.

---

InMemoryResultStore

source

In-memory result store with TTL-based expiry.

Suitable for development and single-process deployments. For production, implement a Redis-backed store.

__init__

def __init__(
    *,
    ttl: int = 3600,
    max_size: int = 10000
) -> None

source

store

async def store(
    job_id: str,
    result: JobResult
) -> None

source

Persist a job result, evicting the oldest entry when at capacity.

get

async def get(job_id: str) -> JobResult | None

source

Return the result for the given job ID, or None if absent or expired.

delete

async def delete(job_id: str) -> bool

source

Remove the result for the given job ID and return True if it existed.

wait

async def wait(
    job_id: str,
    *,
    timeout: float = 30.0,
    poll_interval: float = 0.5
) -> JobResult | None

source

Wait for a result to appear, with timeout.

cleanup_expired

async def cleanup_expired() -> int

source

Remove all entries whose TTL has elapsed and return the count removed.

get_stats

def get_stats() -> dict[str, Any]

source

Return store statistics including size, capacity, and waiter count.

---

JobProtocol

source

Represents a background job with comprehensive status tracking

Jobs are executable work units that can be queued, scheduled, tracked, and retried. They support priority ordering, dependency management, and timeout control.

Attributes: id: Unique job identifier name: Human-readable job name/type args: Positional arguments for job handler kwargs: Keyword arguments for job handler priority: Execution priority (higher = more important) max_retries: Maximum retry attempts on failure timeout: Execution timeout in seconds depends_on: List of job IDs this job depends on status: Current job status created_at: Timestamp when job was created started_at: Timestamp when job started executing completed_at: Timestamp when job finished retry_count: Current number of retry attempts last_error: Most recent error message result: JobProtocol execution result scheduled_at: Timestamp for scheduled execution cron_expression: Cron expression for recurring jobs

is_pending

property is_pending() -> bool

source

Check if job is pending execution

is_running

property is_running() -> bool

source

Check if job is currently running

is_completed

property is_completed() -> bool

source

Check if job completed successfully

is_failed

property is_failed() -> bool

source

Check if job failed

can_retry

property can_retry() -> bool

source

Check if job can be retried (hasn’t exceeded max_retries)

duration_ms

property duration_ms() -> float | None

source

Get job execution duration in milliseconds

mark_running

def mark_running() -> None

source

Mark job as running and record start time

mark_completed

def mark_completed(result: JobResult) -> None

source

Mark job as completed with result

Parameters

Parameter	Type	Description
`result`	JobResult	JobProtocol execution result

mark_failed

def mark_failed(error: str) -> None

source

Mark job as failed with error message

Parameters

Parameter	Type	Description
`error`	str	Error message describing the failure

mark_retrying

def mark_retrying() -> None

source

Mark job as retrying and increment retry count

mark_cancelled

def mark_cancelled() -> None

source

Mark job as cancelled

to_dict

def to_dict() -> dict[str, Any]

source

Convert job to dictionary for serialization

Returns

Type	Description
dict[str, Any]	Dictionary representation of job

from_dict

def from_dict(
    cls,
    data: dict[str, Any]
) -> JobProtocol

source

Create job from dictionary

Parameters

Parameter	Type	Description
`data`	dict[str, Any]	Dictionary representation of job

Returns

Type	Description
JobProtocol	JobProtocol instance

---

JobResult

source

Result of a job execution

Captures the outcome of job execution including success status, return data, error information, and performance metrics.

ok

def ok(
    cls,
    data: Any = None,
    duration: float = 0.0
) -> JobResult

source

Create a successful result

Parameters

Parameter	Type	Description
`data`	Any	Return value from the job
`duration`	float	Execution duration in seconds

Returns

Type	Description
JobResult	JobResult with success=True

fail

def fail(
    cls,
    error: str,
    retry_count: int = 0,
    duration: float = 0.0
) -> JobResult

source

Create a failure result

Parameters

Parameter	Type	Description
`error`	str	Error message describing the failure
`retry_count`	int	Number of retry attempts made
`duration`	float	Execution duration in seconds

Returns

Type	Description
JobResult	JobResult with success=False

to_dict

def to_dict() -> dict[str, Any]

source

Convert result to dictionary for serialization

from_dict

def from_dict(
    cls,
    data: dict[str, Any]
) -> JobResult

source

Create result from dictionary

---

JobStatus

source

JobProtocol execution status.

---

JobTemplateProtocol

source

Template for creating scheduled jobs

JobTemplateProtocol defines a reusable job configuration that can be instantiated multiple times with different IDs.

Attributes: name: JobProtocol name/type args: Positional arguments kwargs: Keyword arguments priority: Execution priority max_retries: Maximum retry attempts timeout: Execution timeout depends_on: JobProtocol dependencies template

create_job

def create_job(job_id: str = '') -> JobProtocol

source

Create a JobProtocol instance from this template

Parameters

Parameter	Type	Description
`job_id`	str	JobProtocol ID (generated if empty)

Returns

Type	Description
JobProtocol	New JobProtocol instance

from_job

def from_job(
    cls,
    job: JobProtocol
) -> JobTemplateProtocol

source

Create template from existing job

Parameters

Parameter	Type	Description
`job`	JobProtocol	JobProtocol to convert to template

Returns

Type	Description
JobTemplateProtocol	JobTemplateProtocol instance

---

LoggingMiddleware

source

Logs task start, completion, and failures.

before_execute

async def before_execute(ctx: TaskExecutionContext) -> None

source

Log task start before execution.

after_execute

async def after_execute(ctx: TaskExecutionContext) -> None

source

Log task completion status and duration after execution.

on_error

async def on_error(
    ctx: TaskExecutionContext,
    error: Exception
) -> None

source

Log the error when a task raises an exception.

---

MemoryTaskQueue

source

In-memory task queue implementation

Uses Python’s heapq for priority-based ordering. Tasks are stored in memory and will be lost on process restart. Thread-safe using asyncio.Lock for concurrent access.

Characteristics:

• Storage: In-memory heap-based priority queue
• Concurrency: asyncio.Lock for thread safety
• Persistence: No persistence (volatile)
• Use Case: Development, testing, single-process applications
• Limitations: No persistence, single-process only

Example

queue = MemoryTaskQueue()
await queue.enqueue(task)
task = await queue.dequeue()

__init__

def __init__() -> None

source

Initialize memory task queue

set_hook_registry

def set_hook_registry(hooks: HookRegistryProtocol | None) -> None

source

Attach an optional hook registry after provider boot wiring.

enqueue

async def enqueue(task: JobProtocol) -> Result[str, TaskError]

source

Add a task to the queue, returning Ok(task_id) or Err(TaskError) on failure.

Tasks are stored in a min-heap with negative priority to achieve max-heap behavior (highest priority first).

Parameters

Parameter	Type	Description
`task`	JobProtocol	Task to enqueue

Returns

Type	Description
Result[str, TaskError]	Ok containing the enqueued task id, or Err containing a TaskError.

dequeue

async def dequeue() -> JobProtocol | None

source

Remove and return the next task from the queue

Returns the highest priority task. If multiple tasks have the same priority, returns the oldest (FIFO). Respects task delay - tasks with future available_at are skipped.

Returns

Type	Description
JobProtocol | None	Next task or None if queue is empty or no tasks are available

ack

async def ack(task_id: str) -> None

source

Acknowledge successful processing of a task

Removes the task from the in-flight tracking set.

Parameters

Parameter	Type	Description
`task_id`	str	ID of the task to acknowledge

nack

async def nack(
    task_id: str,
    requeue: bool = True
) -> None

source

Negative-acknowledge a task, optionally requeuing it

Removes the task from in-flight tracking. If requeue is True the task is pushed back onto the priority heap for retry.

Parameters

Parameter	Type	Description
`task_id`	str	ID of the task to negative-acknowledge
`requeue`	bool	If True, return the task to the queue for retry

get_task_count

async def get_task_count() -> int

source

Get the number of tasks in the queue

Returns

Type	Description
int	Number of pending tasks

clear

async def clear() -> None

source

Clear all tasks from the queue

close

async def close() -> None

source

Close the queue connection

For memory queue, there are no external resources to clean up.

---

MetricsMiddleware

source

Collects per-task-type execution metrics.

__init__

def __init__() -> None

source

after_execute

async def after_execute(ctx: TaskExecutionContext) -> None

source

Record execution metrics for the completed task.

on_error

async def on_error(
    ctx: TaskExecutionContext,
    error: Exception
) -> None

source

Increment failure counter for the task type that raised an error.

get_stats

def get_stats() -> dict[str, Any]

source

Return aggregated execution statistics keyed by task name.

---

NamedTaskConfig

source

Configuration for a single named task queue backend.

Used in TaskConfig.backends to declare multiple task queues that the framework registers as Named DI bindings.

Parameters

Parameter	Type	Description
`name`		Unique backend identifier. Used as the Named() DI key.
`primary`		Whether this is the primary backend. Primary backends also receive the unnamed TaskQueueProtocol binding.
`type`		Backend type ('memory', 'redis', 'rabbitmq', 'postgres').
`redis_url`		Redis connection URL (when type='redis').
`amqp_url`		AMQP connection URL (when type='rabbitmq').
`postgres_dsn`		Postgres DSN (when type='postgres').
`queue_name`		Queue name for this backend instance.

Example

backends:
  - name: primary
    primary: true
    type: redis
    redis_url: "${REDIS_URL}"
  - name: notifications
    type: rabbitmq
    amqp_url: "${AMQP_URL}"

backends:
  - name: primary
    primary: true
    type: redis
    redis_url: "${REDIS_URL}"
  - name: notifications
    type: rabbitmq
    amqp_url: "${AMQP_URL}"

Inject by name

queue: Annotated[TaskQueueProtocol, Named("notifications")]

queue: Annotated[TaskQueueProtocol, Named("notifications")]

---

OnErrorPolicy

source

Controls what happens when ScheduledWorker.run_cycle raises.

Attributes: LOG_AND_CONTINUE: Log the error, then resume the normal schedule. BACKOFF: Log the error and double the sleep before the next cycle (up to 10 * interval_seconds). STOP: Log the error and stop the worker permanently.

---

Priority

source

Task priority levels for queue ordering (higher number = higher priority).

The scale is 0–20 and is separate from the kernel’s ProviderPriority (0–100). Workers dequeue higher-priority jobs before lower-priority ones.

Values: LOW (0): Background tasks that can be deferred indefinitely. NORMAL (5): Default priority for routine task processing. HIGH (10): Time-sensitive tasks processed before NORMAL queue. CRITICAL (20): Highest priority; reserved for urgent system tasks.

---

ProgressInfo

source

Current progress state of a task.

is_complete

property is_complete() -> bool

source

Return True when the current count has reached the total.

elapsed_seconds

property elapsed_seconds() -> float

source

Return seconds elapsed since the task started, or 0 if not started.

---

ProgressSnapshot

source

Immutable point-in-time view of a task's progress.

Parameters

Parameter	Type	Description
`task_id`		Unique identifier for the tracked task.
`current`		Number of units completed so far.
`total`		Total number of units to complete (0 means unknown).
`status`		Current lifecycle state.
`message`		Human-readable status message.
`error`		Error description when ``status`` is ``FAILED``; empty otherwise.

percent

property percent() -> float

source

Completion percentage in the range [0.0, 100.0].

Returns 0.0 when total is unknown (zero).

---

ProgressStatus

source

Lifecycle states for a tracked task.

Values are lowercase strings so they round-trip cleanly through JSON.

---

ProgressStore

source

Abstract storage for progress state.

save

async def save(info: ProgressInfo) -> None

source

Save progress state.

get

async def get(job_id: str) -> ProgressInfo | None

source

Get progress for a job.

delete

async def delete(job_id: str) -> None

source

Remove progress entry.

list_active

async def list_active() -> list[ProgressInfo]

source

List all active (incomplete) progress entries.

---

ProgressTracker

source

Track and report task progress.

Injected into task handlers to enable progress reporting.

__init__

def __init__(
    job_id: str,
    store: ProgressStore
) -> None

source

update

async def update(
    current: int,
    total: int,
    message: str = '',
    **metadata: Any
) -> None

source

Update progress.

Parameters

Parameter	Type	Description
`current`	int	Current progress count.
`total`	int	Total items to process.
`message`	str	Human-readable status message. **metadata: Additional metadata.

complete

async def complete(message: str = 'Complete') -> None

source

Mark progress as complete.

fail

async def fail(message: str) -> None

source

Mark progress as failed.

info

property info() -> ProgressInfo

source

Return the current ProgressInfo snapshot.

---

QueueRateLimiter

source

Per-queue rate limiter.

Manages independent rate limits for named task queues, delegating each limit to a TokenBucket.

Example

limiter = QueueRateLimiter()
limiter.add_limit("emails", rate=5, per=1.0)   # 5 / sec
limiter.add_limit("reports", rate=1, per=60.0) # 1 / min

await limiter.acquire("emails")

limiter = QueueRateLimiter()
limiter.add_limit("emails", rate=5, per=1.0)   # 5 / sec
limiter.add_limit("reports", rate=1, per=60.0) # 1 / min

await limiter.acquire("emails")

add_limit

def add_limit(
    queue: str,
    rate: int,
    per: float = 1.0,
    burst: int | None = None
) -> None

source

Register a rate limit for a named queue.

Parameters

Parameter	Type	Description
`queue`	str	Queue name to apply the limit to.
`rate`	int	Allowed requests per time period.
`per`	float	Length of the time period in seconds.
`burst`	int | None	Maximum burst size (defaults to *rate*).

acquire

async def acquire(queue: str) -> None

source

Block until a token is available for queue.

Parameters

Parameter	Type	Description
`queue`	str	Queue name; a no-op for unlisted queues.

try_acquire

async def try_acquire(queue: str) -> bool

source

Attempt a non-blocking token acquisition for queue.

Parameters

Parameter	Type	Description
`queue`	str	Queue name.

Returns

Type	Description
bool	``True`` if a token was obtained or the queue has no limit; ``False`` if the queue is currently rate-limited.

---

ResultStore

source

Abstract result store backend.

store

async def store(
    job_id: str,
    result: JobResult
) -> None

source

Store a job result.

get

async def get(job_id: str) -> JobResult | None

source

Get a result by job ID. Returns None if not found or expired.

delete

async def delete(job_id: str) -> bool

source

Delete a result. Returns True if deleted.

wait

async def wait(
    job_id: str,
    *,
    timeout: float = 30.0,
    poll_interval: float = 0.5
) -> JobResult | None

source

Wait for a result with timeout.

cleanup_expired

async def cleanup_expired() -> int

source

Remove expired results. Returns count removed.

---

ScheduledJob

source

Represents a scheduled job with cron expression

Attributes: job_template: Template for creating job instances cron_expression: Cron expression string next_run: Unix timestamp of next execution enabled: Whether job is enabled for execution

calculate_next_run

def calculate_next_run() -> float

source

Calculate the next run time based on cron expression

Returns

Type	Description
float	Unix timestamp of next execution

---

ScheduledWorker

source

Periodic worker base class.

Subclass and implement run_cycle. The base class handles the scheduling loop, error policy, graceful shutdown, and task tracking via BackgroundTaskManager.

Class-level defaults (override per subclass or via constructor)

class MyWorker(ScheduledWorker):
    interval_seconds = 60.0
    initial_delay_seconds = 0.0
    max_jitter_seconds = 0.0
    on_error_policy = OnErrorPolicy.LOG_AND_CONTINUE

class MyWorker(ScheduledWorker):
    interval_seconds = 60.0
    initial_delay_seconds = 0.0
    max_jitter_seconds = 0.0
    on_error_policy = OnErrorPolicy.LOG_AND_CONTINUE

Constructor injection is the preferred way to customise at runtime

worker = MyWorker(
    task_manager=task_manager,
    interval_seconds=30.0,
    on_error_policy=OnErrorPolicy.BACKOFF,
)

worker = MyWorker(
    task_manager=task_manager,
    interval_seconds=30.0,
    on_error_policy=OnErrorPolicy.BACKOFF,
)

__init__

def __init__(
    task_manager: BackgroundTaskManager,
    *,
    interval_seconds: float | None = None,
    initial_delay_seconds: float | None = None,
    max_jitter_seconds: float | None = None,
    on_error_policy: OnErrorPolicy | None = None
) -> None

source

Initialise the worker.

Parameters

Parameter	Type	Description
`task_manager`	BackgroundTaskManager	Shared BackgroundTaskManager that owns the task handle and participates in framework shutdown.
`interval_seconds`	float | None	Override interval_seconds.
`initial_delay_seconds`	float | None	Override initial_delay_seconds.
`max_jitter_seconds`	float | None	Override max_jitter_seconds.
`on_error_policy`	OnErrorPolicy | None	Override on_error_policy.

start

async def start() -> None

source

Start the worker loop as a tracked background task.

Safe to call only once. Calling start() while already running is a no-op (logs a warning).

stop

async def stop(grace_seconds: float = 2.0) -> None

source

Signal the worker to stop and wait for it to finish.

Graceful: the stop event is set so the sleeping phase wakes up early, and the current cycle (if any) is allowed to finish. If the task has not completed within grace_seconds, it is forcibly cancelled.

Parameters

Parameter	Type	Description
`grace_seconds`	float	Seconds to wait for a clean stop before cancelling. Default is 2 seconds.

run_cycle

async def run_cycle() -> None

source

Override to implement one unit of periodic work.

Called once per interval. Must not swallow asyncio.CancelledError.

---

StepResult

source

Result from a single workflow step.

---

TaskBackendConfig

source

Configuration for task queue backends.

model_config: ClassVar[ConfigDict] = ConfigDict(extra=“ignore”)

---

TaskChain

source

Sequential task execution — output of each step feeds the next.

Example

chain = TaskChain([ TaskStep(“parse”, parse_data), TaskStep(“validate”, validate_data), TaskStep(“save”, save_data), ]) result = await chain.execute(raw_input)

__init__

def __init__(
    steps: list[TaskStep],
    *,
    stop_on_error: bool = True,
    state_store: WorkflowStateStore | None = None,
    workflow_id: str | None = None
) -> None

source

execute

async def execute(input_data: Any = None) -> Result[WorkflowResult, WorkflowError]

source

Execute all steps in sequence, threading output of each step into the next.

If a workflow_id and state_store were provided at construction, this method will attempt to resume from the last saved checkpoint — any step whose name appears in the persisted state’s completed steps will be skipped, and the last completed step’s output will seed the next step.

---

TaskChord

source

Fan-out then collect — parallel steps + callback with all results.

Example

chord = TaskChord( steps=[TaskStep(“a”, fetch_a), TaskStep(“b”, fetch_b)], callback=TaskStep(“merge”, merge_results), ) result = await chord.execute()

__init__

def __init__(
    steps: list[TaskStep],
    callback: TaskStep
) -> None

source

execute

async def execute(input_data: Any = None) -> Result[WorkflowResult, WorkflowError]

source

Fan out to all group steps then call the callback with their collected results.

---

TaskCompletedEvent

source

Emitted when a task finishes successfully.

Attributes: task_id: Unique identifier of the completed task. worker_id: Identifier of the worker that executed the task.

---

TaskCompletedHook

source

Payload fired when a task finishes successfully.

Attributes: task_name: Name or type label of the completed task. task_id: Unique identifier of the task instance.

---

TaskConfig

source

Hierarchical root configuration for Lexigram Tasks.

Attributes: name: Configuration name (default: “tasks”) enabled: Whether tasks module is enabled backend: Task queue backend configuration worker: Task worker settings scheduler: Task scheduler settings retry: Task retry configuration rate_limit: Task rate limiting timeout: Task timeout settings extra: Extra configuration

is_production

property is_production() -> bool

source

Check if running in production environment.

is_development

property is_development() -> bool

source

Check if running in development environment.

is_test

property is_test() -> bool

source

Check if running in test environment.

validate_for_environment

def validate_for_environment(env: Environment | None = None) -> list[ConfigIssue]

source

Validate task configuration for the given environment.

Parameters

Parameter	Type	Description
`env`	Environment | None	Target environment; resolved from ``LEX_ENV`` when ``None``.

Returns

Type	Description
list[ConfigIssue]	List of ConfigIssue instances.

from_named

def from_named(
    cls,
    entry: NamedTaskConfig
) -> TaskConfig

source

Build a single-backend TaskConfig from a NamedTaskConfig entry.

Used internally by TasksProvider to create per-backend configs.

Parameters

Parameter	Type	Description
`entry`	NamedTaskConfig	The named backend entry to materialise.

Returns

Type	Description
TaskConfig	A TaskConfig configured for the single named backend.

---

TaskDashboard

source

Aggregated task observability dashboard.

__init__

def __init__(
    *,
    history_size: int = 1000,
    window_seconds: float = 300
) -> None

source

record_execution

def record_execution(
    task_name: str,
    *,
    job_id: str = '',
    duration_ms: float = 0.0,
    success: bool = True,
    error: str | None = None,
    worker_id: str | None = None
) -> None

source

Record a task execution.

get_summary

def get_summary() -> dict[str, Any]

source

Get dashboard summary with throughput and error rates.

get_per_task_stats

def get_per_task_stats() -> dict[str, dict[str, Any]]

source

Get per-task-type statistics with latency percentiles.

get_recent_executions

def get_recent_executions(
    *,
    limit: int = 50,
    task_name: str | None = None
) -> list[dict[str, Any]]

source

Get recent execution history.

get_error_breakdown

def get_error_breakdown() -> dict[str, list[dict[str, Any]]]

source

Get recent errors grouped by task type.

reset

def reset() -> None

source

Reset all counters and history.

---

TaskEnqueuedHook

source

Payload fired when a task is placed on the queue.

Attributes: task_name: Name or type label of the enqueued task. queue_name: Name of the queue the task was added to.

---

TaskExecutionContext

source

Context passed through the middleware pipeline.

---

TaskFailedEvent

source

Emitted when a task fails after all retry attempts are exhausted.

Attributes: task_id: Unique identifier of the failed task. error: Human-readable description of the failure reason.

---

TaskFailedHook

source

Payload fired when a task raises an unhandled exception.

Attributes: task_name: Name or type label of the failed task. task_id: Unique identifier of the task instance. reason: Short description or exception message.

---

TaskGroup

source

Parallel task execution — all steps run concurrently.

Example

group = TaskGroup([ TaskStep(“email”, send_email), TaskStep(“sms”, send_sms), ]) result = await group.execute(user_data)

__init__

def __init__(
    steps: list[TaskStep],
    *,
    max_concurrency: int | None = None,
    state_store: WorkflowStateStore | None = None,
    workflow_id: str | None = None
) -> None

source

execute

async def execute(input_data: Any = None) -> Result[WorkflowResult, WorkflowError]

source

Run all steps concurrently, respecting max_concurrency if set.

---

TaskHealth

source

Health status for task processing system

Aggregates health information from queue, workers, and scheduler.

is_healthy

property is_healthy() -> bool

source

Check if system is healthy

success_rate

property success_rate() -> float

source

Calculate job success rate

to_dict

def to_dict() -> dict[str, Any]

source

Convert to dictionary for serialization

---

TaskMiddleware

source

Abstract base for task middleware.

before_execute

async def before_execute(ctx: TaskExecutionContext) -> None

source

Called before task execution. Override to modify or log.

after_execute

async def after_execute(ctx: TaskExecutionContext) -> None

source

Called after task execution. Override to observe or log.

on_error

async def on_error(
    ctx: TaskExecutionContext,
    error: Exception
) -> None

source

Called when task execution fails.

---

TaskMiddlewarePipeline

source

Chains middleware around task execution.

__init__

def __init__() -> None

source

add

def add(middleware: TaskMiddleware) -> None

source

Append a middleware instance to the pipeline.

execute

async def execute(
    job: JobProtocol,
    handler: Any
) -> JobResult

source

Execute a job through the middleware pipeline.

Parameters

Parameter	Type	Description
`job`	JobProtocol	The job to execute.
`handler`	Any	The async handler function.

Returns

Type	Description
JobResult	JobResult from execution.

---

TaskProvider

source

Task processing provider for Lexigram Framework

TaskProvider integrates task processing with the Lexigram Framework, providing dependency injection, lifecycle management, and health monitoring.

Example

from lexigram.app import Application
from lexigram.tasks import TaskProvider, MemoryTaskQueue

app = Application()
queue = MemoryTaskQueue()
provider = TaskProvider(queue, worker_count=4)
app.use(provider)

await app.start()

__init__

def __init__(
    queue: TaskQueueProtocol,
    worker_count: int = 1,
    enable_scheduler: bool = True
)

source

Initialize task provider

Parameters

Parameter	Type	Description
`queue`	TaskQueueProtocol	TaskQueueProtocol implementation to use
`worker_count`	int	Number of workers to create
`enable_scheduler`	bool	Whether to enable job scheduling

from_config

def from_config(
    cls,
    config: TaskConfig,
    **context: Any
) -> TaskProvider

source

Create a TaskProvider from a TaskConfig.

Context kwargs may include a pre-built ‘queue’. If not provided, a MemoryTaskQueue is created as default.

register

async def register(container: ContainerRegistrarProtocol) -> None

source

Register task services with the DI container.

Parameters

Parameter	Type	Description
`container`	ContainerRegistrarProtocol	DI registrar to bind task services into.

boot

async def boot(container: ContainerResolverProtocol) -> None

source

Start the task provider.

Called by the framework on application startup after all registrations.

shutdown

async def shutdown() -> None

source

Shutdown the task provider gracefully

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Check task provider health.

In multi-backend mode the overall status is the worst individual status across all named queues. TaskQueueProtocol does not expose a health_check() method, so liveness is probed via get_task_count().

Returns

Type	Description
HealthCheckResult	HealthCheckResult with current status and metrics.

register_handler

def register_handler(
    task_name: str,
    handler: Callable[Ellipsis, Any]
) -> None

source

Register a task handler

Parameters

Parameter	Type	Description
`task_name`	str	Name of the task type
`handler`	Callable[Ellipsis, Any]	Async handler function

register_scheduled_task

def register_scheduled_task(task_func: Any) -> None

source

Register a decorated task function for scheduling.

Validates that the task has required metadata (_task_name and _cron). Registration is atomic: if scheduling fails, the handler is rolled back.

Parameters

Parameter	Type	Description
`task_func`	Any	Task function decorated with @scheduled

Raises

Exception	Description
TaskRegistrationError	If _task_name or _cron is missing

enqueue_job

async def enqueue_job(job: JobProtocol) -> str

source

Enqueue a job for processing

schedule_job_sync

def schedule_job_sync(
    job_template: JobProtocol | JobTemplateProtocol,
    cron_expression: str,
    job_id: str | None = None
) -> str | None

source

Schedule a job with cron expression

unschedule_job_sync

def unschedule_job_sync(job_id: str) -> bool

source

Remove a scheduled job

get_worker_stats

def get_worker_stats() -> dict[str, Any] | None

source

Get worker pool statistics

get_scheduled_jobs

def get_scheduled_jobs() -> dict[str, Any] | None

source

Get scheduled jobs information

refresh_worker_handlers

def refresh_worker_handlers() -> None

source

Refresh handler mappings in all workers.

Call this after registering new handlers to ensure workers can execute jobs registered after pool creation.

---

TaskQueuedEvent

source

Emitted when a task is accepted into a processing queue.

Attributes: task_id: Unique identifier of the queued task. queue_name: Name of the queue that accepted the task.

---

TaskRateLimitConfig

source

Configuration for rate limiting.

model_config: ClassVar[ConfigDict] = ConfigDict(extra=“ignore”)

---

TaskScheduler

source

Task scheduler for cron-like job scheduling

TaskScheduler manages periodic job execution using cron expressions. Jobs are checked at regular intervals and enqueued when due.

Example

scheduler = TaskScheduler()

# Schedule job
template = JobTemplateProtocol(name="cleanup", args=())
scheduler.schedule_job(template, "0 0 * * *")  # Daily at midnight

# Start scheduler
await scheduler.start_scheduler(queue.enqueue)

__init__

def __init__(
    check_interval: float = 60.0,
    store: SchedulerStore | None = None
)

source

Initialize task scheduler.

Parameters

Parameter	Type	Description
`check_interval`	float	How often to check for due jobs (seconds).
`store`	SchedulerStore | None	Optional SchedulerStore for durable persistence of job definitions and next-run times. When provided, all schedule/unschedule operations are persisted and the scheduler can be restored after restart via restore_from_store.

schedule_job_sync

def schedule_job_sync(
    job_template: JobTemplateProtocol | JobProtocol,
    cron_expression: str,
    job_id: str | None = None
) -> str

source

Schedule a job with cron expression (sync, no persistence).

For the primary async version with store persistence, use schedule_job.

Parameters

Parameter	Type	Description
`job_template`	JobTemplateProtocol | JobProtocol	JobProtocol template or JobProtocol to schedule
`cron_expression`	str	Cron expression (e.g., "0 9 * * 1-5")
`job_id`	str | None	Optional job ID, generated if not provided

Returns

Type	Description
str	JobProtocol ID for the scheduled job

schedule_job

async def schedule_job(
    job_template: JobTemplateProtocol | JobProtocol,
    cron_expression: str,
    job_id: str | None = None
) -> str

source

Schedule a job and persist to store if one is configured.

This is the primary async method. For a synchronous variant without persistence, use schedule_job_sync.

Parameters

Parameter	Type	Description
`job_template`	JobTemplateProtocol | JobProtocol	JobProtocol template or JobProtocol to schedule.
`cron_expression`	str	Cron expression (e.g., ``"0 9 * * 1-5"``).
`job_id`	str | None	Optional job ID, generated if not provided.

Returns

Type	Description
str	JobProtocol ID for the scheduled job.

restore_from_store

async def restore_from_store() -> int

source

Restore all persisted job definitions from the configured store.

Loads every job persisted by previous scheduler instances and re-registers them in scheduled_jobs. Call this once during startup before invoking start_scheduler to resume the full scheduling state.

Returns

Type	Description
int	Number of jobs restored.

Raises

Exception	Description
RuntimeError	If no SchedulerStore was provided at construction.

unschedule_job_sync

def unschedule_job_sync(job_id: str) -> bool

source

Remove a scheduled job (sync, no store deletion).

For the primary async version with store deletion, use unschedule_job.

Parameters

Parameter	Type	Description
`job_id`	str	JobProtocol ID to remove

Returns

Type	Description
bool	True if job was removed, False if not found

unschedule_job

async def unschedule_job(job_id: str) -> bool

source

Remove a scheduled job and delete it from the store.

This is the primary async method. For a synchronous variant without store deletion, use unschedule_job_sync.

Parameters

Parameter	Type	Description
`job_id`	str	JobProtocol ID to remove.

Returns

Type	Description
bool	True if job was removed, False if not found.

get_scheduled_jobs

def get_scheduled_jobs() -> list[ScheduledJob]

source

Get all scheduled jobs

Returns

Type	Description
list[ScheduledJob]	List of all scheduled jobs

get_scheduled_job

def get_scheduled_job(job_id: str) -> ScheduledJob | None

source

Get a specific scheduled job

Parameters

Parameter	Type	Description
`job_id`	str	JobProtocol ID

Returns

Type	Description
ScheduledJob | None	ScheduledJob or None if not found

start_scheduler

async def start_scheduler(enqueue_callback: Callable[[JobProtocol], Any]) -> None

source

Start the scheduler loop

Parameters

Parameter	Type	Description
`enqueue_callback`	Callable[[JobProtocol], Any]	Async function to enqueue jobs when they are due

stop_scheduler

async def stop_scheduler() -> None

source

Stop the scheduler gracefully

enable_job

def enable_job(job_id: str) -> bool

source

Enable a scheduled job

Parameters

Parameter	Type	Description
`job_id`	str	JobProtocol ID

Returns

Type	Description
bool	True if job was enabled

disable_job

def disable_job(job_id: str) -> bool

source

Disable a scheduled job

Parameters

Parameter	Type	Description
`job_id`	str	JobProtocol ID

Returns

Type	Description
bool	True if job was disabled

get_next_run_times

def get_next_run_times() -> dict[str, float]

source

Get next run times for all enabled scheduled jobs

Returns

Type	Description
dict[str, float]	Dictionary mapping job ID to next run timestamp

---

TaskSchedulerConfig

source

Configuration for task scheduler.

model_config: ClassVar[ConfigDict] = ConfigDict(extra=“ignore”)

---

TaskStartedHook

source

Payload fired when a worker picks up and begins executing a task.

Attributes: task_name: Name or type label of the task being executed. task_id: Unique identifier of the task instance.

---

TaskStep

source

A single step in a workflow.

Parameters

Parameter	Type	Description
`name`		Human-readable step name.
`handler`		Async callable that takes input and returns output.
`on_error`		Optional error handler.
`timeout`		Optional timeout in seconds.

---

TaskTimeoutConfig

source

Configuration for task timeout behavior.

---

TaskWorker

source

Individual task worker for job execution

TaskWorker processes jobs from a queue using registered handlers. Each worker runs independently and tracks its own statistics.

Example

worker = TaskWorker("worker-1", queue, handlers)
await worker.start()
# ... worker processes jobs ...
await worker.stop()

__init__

def __init__(
    worker_id: str,
    queue: TaskQueueProtocol,
    handler_registry: dict[str, Callable[Ellipsis, Any]],
    config: TaskWorkerConfig | None = None,
    services: TaskWorkerServices | None = None,
    *,
    middleware_pipeline: TaskMiddlewarePipeline | None = None
)

source

Initialize task worker.

Parameters

Parameter	Type	Description
`worker_id`	str	Unique worker identifier.
`queue`	TaskQueueProtocol	TaskQueueProtocol to pull jobs from.
`handler_registry`	dict[str, Callable[Ellipsis, Any]]	Dict mapping job names to handler functions.
`config`	TaskWorkerConfig | None	Optional worker configuration controlling timeouts and execution behaviour. Defaults to TaskWorkerConfig() when not provided.
`services`	TaskWorkerServices | None	Optional bundle of infrastructure services (DLQ, result store, progress store, dashboard, retry policy, task manager, idempotency manager, logger). Use TaskWorkerServices to compose only the services you need; absent services degrade gracefully. Pass a ``logger`` via ``services.logger`` to inject a custom bound logger; otherwise a module logger is created and bound to ``worker_id`` automatically.
`middleware_pipeline`	TaskMiddlewarePipeline | None	Optional middleware pipeline for cross-cutting concerns.

get_stats

def get_stats() -> WorkerJobStats

source

Get worker statistics

Returns

Type	Description
WorkerJobStats	Current worker statistics

is_busy

def is_busy() -> bool

source

Check if worker is currently processing a job

Returns

Type	Description
bool	True if worker has a job in progress

get_progress_tracker

def get_progress_tracker(job_id: str) -> ProgressTracker | None

source

Get a ProgressTracker for a job (opt-in for handlers).

Handlers can use this to report progress. The worker must have a progress_store configured.

Parameters

Parameter	Type	Description
`job_id`	str	The job ID to track progress for

Returns

Type	Description
ProgressTracker | None	ProgressTracker instance if progress_store is configured, None otherwise

---

TaskWorkerConfig

source

Configuration for task workers.

model_config: ClassVar[ConfigDict] = ConfigDict(extra=“ignore”)

---

TaskWorkerServices

source

Optional infrastructure services injected into a TaskWorker.

Groups all optional infrastructure dependencies so that the TaskWorker constructor stays focused on the mandatory parameters. All fields default to None — omitted services degrade gracefully.

Attributes: container: DI container for resolving task dependencies. dead_letter_queue: DLQ for routing permanently-failed jobs. result_store: Persistent store for job execution results. progress_store: Store for in-progress task progress updates. dashboard: Observability dashboard for worker and job metrics. retry_policy: Retry policy implementation (from lexigram-resilience). task_manager: Kernel task manager for critical shutdown registration. idempotency_manager: Manager for at-most-once job execution. logger: Optional logger instance.

---

TasksModule

source

Background task workers, scheduling, and async job queues.

Call configure to configure the task-processing subsystem.

Usage (in-memory queue for development)

from lexigram.tasks.backends.memory import MemoryTaskQueue

@module(
    imports=[TasksModule.configure(queue=MemoryTaskQueue())]
)
class AppModule(Module):
    pass

from lexigram.tasks.backends.memory import MemoryTaskQueue

@module(
    imports=[TasksModule.configure(queue=MemoryTaskQueue())]
)
class AppModule(Module):
    pass

Usage (Redis queue for production)

from lexigram.tasks.backends.redis import RedisTaskQueue

@module(
    imports=[
        TasksModule.configure(
            queue=RedisTaskQueue(url="redis://localhost"),
            worker_count=4,
        )
    ]
)
class AppModule(Module):
    pass

from lexigram.tasks.backends.redis import RedisTaskQueue

@module(
    imports=[
        TasksModule.configure(
            queue=RedisTaskQueue(url="redis://localhost"),
            worker_count=4,
        )
    ]
)
class AppModule(Module):
    pass

configure

def configure(
    cls,
    queue: Any | None = None,
    worker_count: int = 1,
    enable_scheduler: bool = True
) -> DynamicModule

source

Create a TasksModule with explicit configuration.

Parameters

Parameter	Type	Description
`queue`	Any | None	TaskQueueProtocol implementation. Defaults to MemoryTaskQueue.
`worker_count`	int	Number of concurrent task workers.
`enable_scheduler`	bool	Whether to enable the job scheduler.

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

scope

def scope(
    cls,
    *handlers: type,
    **kwargs: Any
) -> DynamicModule

source

Scope task handler classes into a feature module.

Registers the given task handler classes as providers so they are discovered and wired by the task executor. The parent module graph must already include TasksModule.configure — this does not create a new queue or worker pool.

Uses the anonymous token pattern so both configure() and scope() can coexist in the same compiled graph without a ModuleDuplicateError.

Example

@module(
    imports=[
        TasksModule.configure(queue=RedisTaskQueue(url=...)),
        TasksModule.scope(SendEmailTask, GenerateReportTask),
    ]
)
class NotificationFeatureModule(Module):
    pass

@module(
    imports=[
        TasksModule.configure(queue=RedisTaskQueue(url=...)),
        TasksModule.scope(SendEmailTask, GenerateReportTask),
    ]
)
class NotificationFeatureModule(Module):
    pass

Parameters

Parameter	Type	Description

Returns

Type	Description
DynamicModule	A DynamicModule scoped to this feature.

stub

def stub(
    cls,
    config: Any = None
) -> DynamicModule

source

Return an in-memory TasksModule for unit testing.

Uses a MemoryTaskQueue with a single worker and no job scheduler.

Returns

Type	Description
DynamicModule	A DynamicModule backed by an in-memory task queue.

---

TimeoutMiddleware

source

Enforces execution timeout on tasks.

Uses asyncio.wait_for to enforce timeout on task execution.

__init__

def __init__(
    default_timeout: float = 300.0,
    max_timeout: float | None = None
) -> None

source

before_execute

async def before_execute(ctx: TaskExecutionContext) -> None

source

Compute and store the effective timeout in the execution context metadata.

execute_with_timeout

async def execute_with_timeout(
    handler: Any,
    job: JobProtocol,
    ctx: TaskExecutionContext,
    resolved_args: list[Any] | None = None,
    resolved_kwargs: dict[str, Any] | None = None
) -> JobResult

source

Execute handler with timeout enforcement.

Parameters

Parameter	Type	Description
`handler`	Any	The handler to execute
`job`	JobProtocol	The job being executed
`ctx`	TaskExecutionContext	Execution context
`resolved_args`	list[Any] | None	Pre-resolved positional arguments (from DI)
`resolved_kwargs`	dict[str, Any] | None	Pre-resolved keyword arguments (from DI)

Returns

Type	Description
JobResult	JobResult from execution

---

UniqueTask

source

Decorator for unique task execution.

Prevents duplicate execution of the same task using in-memory locks.

Example

lock_manager = LockManager()
@unique_task(
    lock_manager=lock_manager,
    key_func=lambda user_id: f"sync_user:{user_id}",
    timeout=3600
)
async def sync_user(user_id: str):
    await sync_user_data(user_id)

__init__

def __init__(
    lock_manager: LockManager,
    key_func: Callable[[Any], str],
    timeout: float = 60.0,
    skip_if_locked: bool = True
)

source

Initialize unique task decorator.

Parameters

Parameter	Type	Description
`lock_manager`	LockManager	LockManager instance for creating locks
`key_func`	Callable[[Any], str]	Function to generate lock key from task args
`timeout`	float	Lock timeout in seconds
`skip_if_locked`	bool	If True, skip execution if lock held; if False, wait

---

WorkerJobStats

source

Worker statistics for monitoring performance.

Tracks execution metrics for an individual worker including job counts, timing, and activity tracking.

uptime

property uptime() -> float

source

Get worker uptime in seconds.

record_job

def record_job(result: JobResult) -> None

source

Record job execution statistics.

Parameters

Parameter	Type	Description
`result`	JobResult	JobProtocol execution result.

---

WorkerPool

source

Pool of task workers with monitoring and scaling

WorkerPool manages multiple TaskWorker instances for concurrent job processing. Supports dynamic scaling and aggregate statistics.

Example

pool = WorkerPool(queue, handlers, size=4)
await pool.start()
stats = pool.get_pool_stats()
await pool.scale_to(8)  # Scale up
await pool.stop()

__init__

def __init__(
    queue: TaskQueueProtocol,
    handler_registry: dict[str, Callable[Ellipsis, Any]] | HandlerRegistry,
    size: int = 1,
    logger: Logger | None = None,
    task_manager: TaskManagerProtocol | None = None,
    dead_letter_queue: DeadLetterQueue | None = None,
    hooks: HookRegistryProtocol | None = None,
    container: Any = None
)

source

Initialize worker pool

Parameters

Parameter	Type	Description
`queue`	TaskQueueProtocol	TaskQueueProtocol for workers to pull from
`handler_registry`	dict[str, Callable[Ellipsis, Any]] | HandlerRegistry	Dict mapping job names to handlers, or HandlerRegistry for dynamic lookups (allows handlers registered after pool creation).
`size`	int	Number of workers to create
`task_manager`	TaskManagerProtocol | None	Optional kernel task manager for critical task registration during worker lifecycle.
`dead_letter_queue`	DeadLetterQueue | None	Optional shared DLQ for permanently failed jobs.
`container`	Any	Optional DI container for resolving task dependencies at runtime.

refresh_handlers

def refresh_handlers() -> dict[str, Callable[Ellipsis, Any]]

source

Refresh handlers from registry if using dynamic registry.

start

async def start() -> None

source

Start the worker pool

Creates and starts all workers in the pool.

stop

async def stop(
    drain: bool = False,
    timeout: float | None = 30.0
) -> None

source

Stop the worker pool gracefully.

Parameters

Parameter	Type	Description
`drain`	bool	If ``True``, wait for each worker to finish its current job before stopping. Defaults to ``False`` for a fast cancel-and-stop.
`timeout`	float | None	Per-worker seconds to wait when *drain* is ``True``. ``None`` means wait indefinitely. Defaults to 30 seconds.

get_worker_stats

def get_worker_stats() -> list[WorkerJobStats]

source

Get statistics for all workers

Returns

Type	Description
list[WorkerJobStats]	List of WorkerJobStats for each worker

get_pool_stats

def get_pool_stats() -> dict[str, Any]

source

Get aggregate pool statistics

Returns

Type	Description
dict[str, Any]	Dictionary with pool-wide statistics

scale_to

async def scale_to(new_size: int) -> None

source

Scale the worker pool to a new size

Dynamically add or remove workers.

Parameters

Parameter	Type	Description
`new_size`	int	Target number of workers

---

WorkflowResult

source

Result from a complete workflow execution.

---

WorkflowStatus

source

Enumeration of possible workflow execution states.

---

Functions

chain

chain

def chain(
    *steps: TaskStep,
    stop_on_error: bool = True
) -> TaskChain

source

Create a sequential TaskChain from the given steps.

Each step receives the output of the previous step as its input.

Parameters

Parameter	Type	Description
`stop_on_error`	bool	Whether to abort the pipeline on the first failure. Defaults to ``True``.

Returns

Type	Description
TaskChain	A configured TaskChain ready to ``await .execute()``.

Example

result = await chain(
    TaskStep("validate", validate_input),
    TaskStep("process", process_data),
    TaskStep("persist", save_to_db),
).execute(raw_data)

result = await chain(
    TaskStep("validate", validate_input),
    TaskStep("process", process_data),
    TaskStep("persist", save_to_db),
).execute(raw_data)

---

delay

delay

def delay(fn: Callable[_P, Coroutine[Any, Any, _R]]) -> _DelayedCallable[_P, _R]

source

Decorate an async function to add a ``.delay()`` background-dispatch method.

The decorated function is unchanged when called directly — it still returns a coroutine that must be await-ed. The .delay() method is the only addition; it schedules the coroutine as a background asyncio.Task and stores the reference to prevent premature garbage collection.

Parameters

Parameter	Type	Description
`fn`	Callable[_P, Coroutine[Any, Any, _R]]	An async function (coroutine function) to wrap.

Returns

Type	Description
_DelayedCallable[_P, _R]	A _DelayedCallable that is callable exactly like ``fn`` but also exposes ``.delay(*args, **kwargs) -> asyncio.Task``.

Example

@delay
async def flush_audit_log(batch: list[AuditEntry]) -> None:
    await audit_store.save_many(batch)

# Direct call (awaited in the current task):
await flush_audit_log(batch=entries)

# Background dispatch (does not block):
flush_audit_log.delay(batch=entries)

---

distributed_lock

distributed_lock

async def distributed_lock(
    key: str,
    timeout: float = 60.0,
    lock_manager: LockManager | None = None
) -> AsyncGenerator[InMemoryLock, None]

source

Context manager for in-memory locks.

Deprecated: Use LockManager.acquire() instead for better DI integration.

Parameters

Parameter	Type	Description
`key`	str	Unique lock key
`timeout`	float	Lock timeout in seconds
`lock_manager`	LockManager | None	Optional LockManager instance (creates new if None)

Yields

Type	Description
AsyncGenerator[InMemoryLock, None]	InMemoryLock instance

Example

manager = LockManager()  # Should be from DI container
async with distributed_lock("resource:123", lock_manager=manager):
    await process_resource(123)

---

scheduled

scheduled

def scheduled(
    cron: str,
    name: str | None = None,
    priority: Priority | int = Priority.NORMAL,
    max_retries: int = 3,
    timeout: float | None = None
) -> Callable[[Callable[Ellipsis, Any]], Any]

source

Decorator to define a scheduled task.

Parameters

Parameter	Type	Description
`cron`	str	Cron expression (e.g., "0 9 * * *")
`name`	str | None	Task name
`priority`	Priority | int	Task priority
`max_retries`	int	Maximum retry attempts
`timeout`	float | None	Execution timeout

Returns

Type	Description
Callable[[Callable[Ellipsis, Any]], Any]	Decorated task function with scheduling metadata

---

task

task

def task(
    name: str | None = None,
    priority: Priority | int = Priority.NORMAL,
    max_retries: int = 3,
    timeout: float | None = None,
    queue: str = 'default',
    idempotency_key: str | None = None
) -> Callable[[Callable[Ellipsis, Any]], Any]

source

Decorator to define a task.

Converts a regular function into a task that can be enqueued and executed by workers.

Parameters

Parameter	Type	Description
`name`	str | None	Task name (defaults to function name)
`priority`	Priority | int	Task priority (HIGH, NORMAL, LOW)
`max_retries`	int	Maximum retry attempts on failure
`timeout`	float | None	Execution timeout in seconds
`queue`	str	Queue name for this task
`idempotency_key`	str | None	Optional key for idempotent execution

Returns

Type	Description
Callable[[Callable[Ellipsis, Any]], Any]	Decorated function with .delay() and .apply_async() methods

---

Exceptions

QueueFullError

source

The queue has reached its maximum capacity and cannot accept new tasks.

The caller should back off and retry after a delay, or apply backpressure.

__init__

def __init__(
    message: str = 'Task queue is full',
    *,
    queue_name: str | None = None,
    capacity: int | None = None,
    **kwargs: Any
) -> None

source

---

TaskCancelledError

source

Raised when a task is cancelled

---

TaskError

source

Base exception for all task-related errors

---

TaskExecutionError

source

Raised when a task execution fails

__init__

def __init__(
    message: str,
    original_error: Exception | None = None,
    details: dict[str, Any] | None = None,
    hint: str | None = None
) -> None

source

---

TaskNotFoundError

source

Raised when a task cannot be found

---

TaskTimeoutError

source

Raised when a task execution exceeds timeout

---

TaskValidationError

source

Raised when task parameters are invalid

---