Flaxon 0.3.0 and Teloce-Py 0.3.0 are available on PyPI. Release instructions are maintained on this site; source links identify the framework snapshot. Check version and upgrade guidance.

Tasks API

Task

Task definition.

Constructor

Task(
    name: str,
    func: Callable,
    *,
    retry_policy: RetryPolicy | None = None,
    timeout: int | None = None,
    queue: str = "default",
    priority: int = 0
)
````

## Methods

### run

```python
async def run(
    *args: Any,
    **kwargs: Any
) -> Any

Run the task.

cancel

def cancel() -> None

Cancel the task.

to_result

def to_result() -> TaskResult

Convert the task into a TaskResult.

Properties

Property	Type	Description
id	str	Task ID
name	str	Task name
status	TaskStatus	Task status
created_at	datetime	Created timestamp
started_at	datetime	Started timestamp
completed_at	datetime	Completed timestamp
retry_count	int	Retry count

TaskStatus

Task status enum.

Value	Description
PENDING	Task is pending
RUNNING	Task is running
COMPLETED	Task completed
FAILED	Task failed
RETRY	Task will retry
CANCELLED	Task cancelled
TIMEOUT	Task timed out

TaskQueue

Task queue.

Constructor

TaskQueue(
    name: str = "default",
    max_size: int = 1000
)

Methods

push

async def push(
    task: Task
) -> None

Push a task to the queue.

pop

async def pop(
    timeout: float | None = None
) -> Task | None

Pop a task from the queue.

get

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

Get a task by ID.

remove

async def remove(
    task_id: str
) -> bool

Remove a task.

cancel

async def cancel(
    task_id: str
) -> bool

Cancel a task.

clear

async def clear() -> None

Clear the queue.

size

async def size() -> int

Get queue size.

pending_count

async def pending_count() -> int

Get pending task count.

running_count

async def running_count() -> int

Get running task count.

completed_count

async def completed_count() -> int

Get completed task count.

failed_count

async def failed_count() -> int

Get failed task count.

Worker

Task worker.

Constructor

Worker(
    registry: TaskRegistry,
    queue: TaskQueue | None = None,
    concurrency: int = 10,
    queue_name: str = "default"
)

Methods

start

async def start() -> None

Start the worker.

stop

def stop() -> None

Stop the worker.

shutdown

def shutdown() -> None

Shutdown the worker.

is_running

def is_running() -> bool

Check whether the worker is running.

Scheduler

Task scheduler.

Constructor

Scheduler(
    queue: TaskQueue
)

Methods

schedule

def schedule(
    task: Task,
    delay: int | None = None,
    at: datetime | None = None,
    interval: int | None = None
) -> None

Schedule a task.

start

async def start() -> None

Start the scheduler.

stop

async def stop() -> None

Stop the scheduler.

RetryPolicy

Task retry configuration.

Constructor

RetryPolicy(
    max_retries: int = 3,
    delay: float = 1.0,
    backoff: float = 2.0,
    max_delay: float = 60.0,
    random_jitter: float = 0.1,
    retry_on: list[type[Exception]] | None = None
)

Methods

should_retry

def should_retry(
    retry_count: int,
    error: Exception
) -> bool

Check if retry should be attempted.

get_delay

def get_delay(
    retry_count: int
) -> float

Get retry delay.

TaskResult

Task execution result.

Attributes

Attribute	Type	Description
id	str	Task ID
name	str	Task name
status	TaskStatus	Task status
result	Any	Task result
error	str	None
retry_count	int	Retry count

Methods

is_pending

def is_pending() -> bool

Check if pending.

is_running

def is_running() -> bool

Check if running.

is_completed

def is_completed() -> bool

Check if completed.

is_failed

def is_failed() -> bool

Check if failed.

is_retry

def is_retry() -> bool

Check if retrying.

is_cancelled

def is_cancelled() -> bool

Check if cancelled.

is_timeout

def is_timeout() -> bool

Check if timed out.

is_done

def is_done() -> bool

Check if task is finished.

get_duration

def get_duration() -> float | None

Get task execution duration.

to_dict

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

Convert result to dictionary.

Admin Durable Jobs

The Admin production layer exposes a small persistent job contract for work such as thumbnail generation. It uses the configured AdminStore and records retry state, errors, and completion:

from flaxon.admin import DurableJobStore, DurableJobWorker

jobs = DurableJobStore(admin.store)
worker = DurableJobWorker(jobs)
worker.register("media.thumbnail", make_thumbnail)
job = jobs.enqueue("media.thumbnail", {"relative": "images/photo.jpg"}, max_attempts=4)
await worker.run_once()

Handlers are registered by name so the worker can be started separately from the web process. For multiple workers, use a shared persistent store and add a distributed claim/lease implementation appropriate to the database or Redis deployment.

Use this capability in a full-stack app

Flaxon serves Python APIs and Teloce interfaces in one project. Keep validation, authorization and privileged operations on the server; components call those endpoints and present their results. Follow the full-stack lessons and Teloce integration reference to connect this capability to pages, components and application modules.