Tasks (lexigram-tasks)

Background task processing for Lexigram Framework — Scheduling, workers, and job queues

---

Overview

lexigram-tasks provides background task execution, scheduling, and async job queues with support for Redis, RabbitMQ, PostgreSQL, and in-memory backends. It features configurable retry policies, dead-letter queues, priority queues, rate limiting, cron scheduling, and Prometheus metrics. All services are wired via TasksProvider, which registers task protocols with the DI container.

---

Install

uv add lexigram-tasks
# Optional extras
uv add "lexigram-tasks[redis]"     # Redis backend
uv add "lexigram-tasks[rabbitmq]"  # RabbitMQ backend

Quick Start

from lexigram import Application
from lexigram.di.module import Module, module

# Import the module from the package
from lexigram.tasks import TasksModule

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

app = Application(modules=[AppModule])
if __name__ == "__main__":
    app.run()

Configuration

Zero-config usage: Call TasksModule.configure() with no arguments to use defaults.

Option 1 — YAML file

application.yaml

tasks:
  backend:
    type: redis
    redis_url: redis://localhost:6379/0
  worker:
    worker_count: 4
    max_concurrent_tasks: 10
  scheduler:
    enabled: true
    timezone: UTC

Option 2 — Profiles + Environment Variables (recommended)

export LEX_TASKS__ENABLED=true
# Environment variables for each field

Option 3 — Python

from lexigram.tasks.config import TaskConfig
from lexigram.tasks import TasksModule

config = TaskConfig(
    backend=TaskBackendConfig(type="redis", redis_url="redis://localhost:6379/0"),
    worker=TaskWorkerConfig(worker_count=4, max_concurrent_tasks=10),
)
TasksModule.configure(queue=queue, worker_count=4, enable_scheduler=True)

Config reference

Field	Default	Env var	Description
backend.type	memory	LEX_TASKS__BACKEND__TYPE	Queue backend (redis, rabbitmq, postgres, memory)
backend.redis_url	redis://localhost:6379/0	LEX_TASKS__BACKEND__REDIS_URL	Redis connection URL
backend.amqp_url	amqp://localhost	LEX_TASKS__BACKEND__AMQP_URL	AMQP connection URL
worker.worker_count	4	LEX_TASKS__WORKER__WORKER_COUNT	Number of worker processes
worker.max_concurrent_tasks	10	LEX_TASKS__WORKER__MAX_CONCURRENT_TASKS	Max tasks executed in parallel per worker
worker.default_timeout	300	LEX_TASKS__WORKER__DEFAULT_TIMEOUT	Task execution timeout in seconds
worker.max_retries	3	LEX_TASKS__WORKER__MAX_RETRIES	Maximum retry attempts per task
scheduler.enabled	true	LEX_TASKS__SCHEDULER__ENABLED	Enable cron-based task scheduling
scheduler.timezone	UTC	LEX_TASKS__SCHEDULER__TIMEZONE	Timezone for cron schedule evaluation

Module Factory Methods

Method	Description
TasksModule.configure(queue, worker_count, enable_scheduler)	Configure with explicit queue and worker settings
TasksModule.stub()	Minimal config for testing

Key Features

• Multiple backends — Redis, RabbitMQ, PostgreSQL (transactional), in-memory
• Retry policies — Configurable retries, backoff, max_delay per task
• Dead-letter queue — Failed jobs routed to DLQ for inspection / replay
• Priority queues — urgent, default, bulk built-in; custom queues supported
• Rate limiting — Per-queue and per-task throughput caps
• Concurrency — Bounded worker pool with backpressure
• Cron scheduling — Cron-expression task scheduling via @scheduled decorator
• Observability — Prometheus metrics for queue depth, latency, error rate
• Health checks — /health/tasks endpoint via lexigram-monitor

Testing

async with Application.boot(modules=[TasksModule.stub()]) as app:
    # your test code
    ...

Key Source Files

File	What it contains
src/lexigram/tasks/module.py	TasksModule class with factory methods
src/lexigram/tasks/di/provider.py	TasksProvider — wires task protocols into DI container
src/lexigram/tasks/config.py	TaskConfig and sub-config classes
src/lexigram/tasks/backends/	Backend implementations (memory, redis, rabbitmq, postgres)
src/lexigram/tasks/scheduling/	Cron scheduler and scheduled task decorators