`orangeqs.juice.task_manager._scheduler`#

ScheduledTask scheduling for the task manager.

Module Contents#

Classes#

`ServiceTaskLock`	Serializes work per target service, so a service never runs two Tasks at once.
`ScheduledTask`	A submitted, tracked instance of a Task with lifecycle state.
`ScheduledTaskEvent`	A ScheduledTask state-transition record for publishing.
`Scheduler`	Fires ScheduledTasks onto targets, honoring the shared serial line.

Data#

`ScheduledTaskStatus`	Lifecycle state of a one-off ScheduledTask.
`TERMINAL_STATUSES`	Endstates of Scheduled Task.
`DispatchFn`	`(target, task_id, message) -> result`: send an occurrence, await its result.

API#

orangeqs.juice.task_manager._scheduler.ScheduledTaskStatus#

None

Lifecycle state of a one-off ScheduledTask.

orangeqs.juice.task_manager._scheduler.TERMINAL_STATUSES: frozenset[str]#

‘frozenset(…)’

Endstates of Scheduled Task.

orangeqs.juice.task_manager._scheduler.DispatchFn#

None

(target, task_id, message) -> result: send an occurrence, await its result.

class orangeqs.juice.task_manager._scheduler.ServiceTaskLock#

Serializes work per target service, so a service never runs two Tasks at once.

Each target has one slot. Each Task for the same service tries to acquire the slot to be executed. This guarantees that at most a single Task is running on the target service at any given time.

is_busy(target: str) → bool#: Whether target’s slot is currently taken.

async acquire(target: str) → None#: Take target’s slot, waiting if held (returns at once on a free slot).

release(target: str) → None#: Release target’s slot. A no-op if it is not held.

async hold(target: str) → collections.abc.AsyncGenerator[None]#: Hold target’s slot for the block (waits if busy).

class orangeqs.juice.task_manager._scheduler.ScheduledTask(/, **data: Any)#

Bases: pydantic.BaseModel

A submitted, tracked instance of a Task with lifecycle state.

Task describes what to run; ScheduledTask is the queued/scheduled record of running it.

scheduled_task_id: str#

None

Stable identity of the ScheduledTask.

task_id: str | None#

None

The fresh task id of the running occurrence, if any (assigned at dispatch).

target: str#

None

The service that will execute this ScheduledTask’s Task.

task_type: str#

None

The Task type name (for display and routing).

display_name: str#

None

Human-readable name of the Task.

task_data: dict[str, Any]#

None

The serialized task envelope, re-sent with a fresh id per occurrence.

status: orangeqs.juice.task_manager._scheduler.ScheduledTaskStatus#

‘queued’

Current lifecycle state.

run_at: datetime.datetime | None#

None

Earliest time this ScheduledTask should fire.

submitted_at: datetime.datetime#

None

When the ScheduledTask was submitted (dispatch ordering tiebreak).

result: dict[str, Any] | None#

None

The occurrence result, once completed.

is_eligible(now: datetime.datetime) → bool#: Whether this ScheduledTask is a dispatch candidate at now.

class orangeqs.juice.task_manager._scheduler.ScheduledTaskEvent(/, **data: Any)#

Bases: orangeqs.juice.messaging.Event

A ScheduledTask state-transition record for publishing.

measurement: ClassVar[str]#: ‘scheduled_task_event’

topic() → str#: Topic is the unique ScheduledTask id.

scheduled_task_id: Annotated[str, tag]#: None

task_id: str | None#: None

target: str#: None

task_type: str#: None

display_name: str#: None

status: Annotated[orangeqs.juice.task_manager._scheduler.ScheduledTaskStatus, tag]#: None

class orangeqs.juice.task_manager._scheduler.Scheduler(dispatch: orangeqs.juice.task_manager._scheduler.DispatchFn, *, service_task_lock: orangeqs.juice.task_manager._scheduler.ServiceTaskLock | None = None, publish: collections.abc.Callable[[orangeqs.juice.task_manager._scheduler.ScheduledTaskEvent], None] | None = None, now_fn: collections.abc.Callable[[], datetime.datetime] = _utcnow, tick_interval: float = 1.0)#

Fires ScheduledTasks onto targets, honoring the shared serial line.

Driven by :meth:tick, which takes an injected now so dispatch is deterministic to test. Occurrence completion runs on background tasks; :meth:wait_idle awaits them in tests.

submit(scheduled_task: orangeqs.juice.task_manager._scheduler.ScheduledTask) → orangeqs.juice.task_manager._scheduler.ScheduledTask#: Store a ScheduledTask and announce it as queued.

get(scheduled_task_id: str) → orangeqs.juice.task_manager._scheduler.ScheduledTask | None#: Return the Scheduled Task with scheduled_task_id.

scheduled_tasks() → list[orangeqs.juice.task_manager._scheduler.ScheduledTask]#: Return every tracked ScheduledTask, for the dashboard snapshot query.

cancel(scheduled_task_id: str) → bool#

Cancel a Scheduled Task by id.

A queued Schedule stops firing. A running occurrence is not interrupted — the Scheduled Task is marked cancelled and its running Task is left to finish without recording a result.

Returns False if the Scheduled Task is unknown or already terminal.

async tick(now: datetime.datetime | None = None) → None#: Run one pass: dispatch at most one ScheduledTask per free serial slot.

async wait_idle() → None#: Await all in-flight occurrence completions (used by tests).

async run() → None#: Run the scheduler loop until :meth:stop is called.

stop() → None#: Stop the loop and cancel running occurrences.

orangeqs.juice.task_manager._scheduler#

Module Contents#

Classes#

Data#

API#

`orangeqs.juice.task_manager._scheduler`#