API Reference — Aquilia Documentation
Comprehensive guide and documentation for API Reference in the Aquilia framework. View API reference, examples, and implementation patterns.
Background Tasks / API Reference Tasks API Reference Complete interface specifications for background task decorators, coordinates, workers, and schedulers, compiled from the actual implementation in aquilia/tasks/. Table of Contents , , , , , , , , , , ].map((item, i) => ( • ))} @task Decorator to register an async function as a background task. Can be used with or without parentheses. from aquilia.tasks import task, Priority, every, cron def task( fn=None, *, name: str | None = None, queue: str = "default", priority: Priority = Priority.NORMAL, max_retries: int = 3, retry_delay: float = 1.0, retry_backoff: float = 2.0, retry_max_delay: float = 300.0, timeout: float = 300.0, tags: list[str] | None = None, schedule: Schedule | None = None, ) -> _TaskDescriptor Parameters Parameter Type Default Description ))} TaskManager Central coordinator for creating, routing, enqueuing, and querying background tasks. class TaskManager: def __init__( self, *, backend: TaskBackend | None = None, # MemoryBackend() by default num_workers: int = 4, default_queue: str = "default", cleanup_interval: float = 300.0, # Seconds between cleanup runs cleanup_max_age: float = 3600.0, # Job TTL after termination scheduler_tick: float = 15.0, # Scheduler tick frequency ) -> None async def start(self) -> None: """Start worker threads, cleanup loops, and scheduled task loops.""" async def stop(self, timeout: float = 10.0) -> None: """Gracefully stop all worker tasks and loops, waiting up to timeout.""" async def enqueue( self, func: Callable | _TaskDescriptor, *args: Any, queue: str | None = None, priority: Priority | None = None, delay: float | None = None, # Delay execution in seconds max_retries: int | None = None, timeout: float | None = None, tags: list[str] | None = None, metadata: dict[str, Any] | None = None, **kwargs: Any, ) -> str: """Enqueue task and return job ID string.""" async def get_job(self, job_id: str) -> Job | None: """Retrieve job dataclass by ID.""" async def list_jobs( self, *, queue: str | None = None, state: JobState | None = None, limit: int = 100, offset: int = 0, ) -> list[Job]: """List jobs matching filters, ordered created_at DESC.""" async def cancel(self, job_id: str) -> bool: """Cancel a pending/running job. Returns True if successful.""" async def retry_job(self, job_id: str) -> bool: """Manually force-retry a failed/dead/cancelled job.""" async def flush(self, queue: str | None = None) -> int: """Clear all tasks (or filtered by queue). Returns number of cleared jobs.""" async def get_stats(self) -> dict[str, Any]: """Return comprehensive TaskManager metrics, uptime, and Chart.js datasets.""" Job Dataclass representing the immutable job configuration and mutable execution state. @dataclass class Job: id: str # Hexadecimal UUID prefix (16 chars) name: str # Human readable task name queue: str = "default" priority: Priority = Priority.NORMAL func_ref: str = "" # module:qualname path args: tuple[Any, ...] = () kwargs: dict[str, Any] = field(default_factory=dict) state: JobState = JobState.PENDING result: JobResult | None = None max_retries: int = 3 retry_count: int = 0 created_at: datetime = datetime.now(timezone.utc) started_at: datetime | None = None completed_at: datetime | None = None scheduled_at: datetime | None = None # Set when executing with delay timeout: float = 300.0 metadata: dict[str, Any] = field(default_factory=dict) tags: list[str] = field(default_factory=list) @property def is_terminal(self) -> bool: """Returns True if state is COMPLETED, FAILED, CANCELLED, or DEAD.""" @property def is_runnable(self) -> bool: """Returns True if job is pending/scheduled and scheduled_at has passed.""" @property def next_retry_delay(self) -> float: """Computes next backoff delay with exponential backoff and random jitter.""" @property def can_retry(self) -> bool: """Returns True if retry_count float | None: """Duration of job execution in milliseconds.""" @property def fingerprint(self) -> str: """SHA-256 fingerprint for enqueued parameter deduplication.""" JobResult Container representing task execution outcomes, exceptions, and execution metrics. @dataclass class JobResult: success: bool value: Any = None # Return value (converted to repr() string on dict serialization) error: str | None = None # Exception message error_type: str | None = None # Name of Exception class traceback: str | None = None # Formatted traceback string duration_ms: float = 0.0 # Millisecond execution time Priority Integer enumeration specifying job urgency. Lower values represent higher priority. class Priority(int, Enum): CRITICAL = 0 HIGH = 1 NORMAL = 2 LOW = 3 JobState Lifecycle states for a task job. class JobState(str, Enum): PENDING = "pending" # Waiting in queue SCHEDULED = "scheduled" # Waiting for delayed timestamp RUNNING = "running" # Undergoing worker processing COMPLETED = "completed" # Executed successfully FAILED = "failed" # Failed, pending retry RETRYING = "retrying" # Rescheduled for retry CANCELLED = "cancelled" # Terminated by admin action DEAD = "dead" # Exhausted all retries (sent to dead letter) Schedule Helpers Helper methods to generate periodic task schedules for the scheduler loop. every() def every( *, seconds: float = 0, minutes: float = 0, hours: float = 0, days: float = 0, ) -> IntervalSchedule cron() def cron(expression: str) -> CronSchedule Accepts standard 5-field cron syntax: "minute hour dom month dow". Registry Queries Access internally mapped tasks registered via decorators. def get_registered_tasks() -> dict[str, _TaskDescriptor]: """Retrieve mapping of all task names to their descriptors.""" def get_periodic_tasks() -> dict[str, _TaskDescriptor]: """Retrieve mapping of only scheduled/periodic tasks.""" def get_task(name: str) -> _TaskDescriptor | None: """Look up a task descriptor by its registered name.""" Integration Builder Option Configures background tasks at the workspace level. # from aquilia.integrations import Integration @staticmethod def tasks( backend: str = "memory", num_workers: int = 4, default_queue: str = "default", cleanup_interval: float = 300.0, cleanup_max_age: float = 3600.0, max_retries: int = 3, retry_delay: float = 1.0, retry_backoff: float = 2.0, retry_max_delay: float = 300.0, default_timeout: float = 300.0, auto_start: bool = True, dead_letter_max: int = 1000, scheduler_tick: float = 15.0, enabled: bool = True, ) -> dict[str, Any] Structured Faults Specific errors raised by the tasks subsystem under the "tasks" fault domain. Fault Triggers ))} )
Go to Homepage