#src.fastware.tasks

#src.fastware.tasks

Background task registry with feature-gated lifecycle management, supporting start/stop protocol, factory registration, and graceful shutdown ordering.

Tasks implement the BackgroundTask protocol (start() and stop() methods). Apps register zero-arg factories via TaskRegistry.register(). The registry instantiates and starts tasks during start_all() and stops them during stop_all(). Feature-gated tasks are skipped when their flag is disabled.

#BackgroundTask

Protocol for background tasks managed by the registry.

#start

def start(self) -> None

#stop

def stop(self) -> None

#TaskRegistry

Registry for background tasks with optional feature gating.

Tasks are registered as factories (callables returning BackgroundTask instances). The registry instantiates and starts them during start_all(), and stops them during stop_all().

#register

def register(self, name: str, factory: Callable[[], BackgroundTask], *, feature: str | None=None) -> None

Register a background task factory.

If feature is set, the task only starts when that feature flag is enabled at start_all() time.

Raises ValueError if a task with the same name is already registered.

#start_all

def start_all(self, features: FeatureFlags | None=None) -> None

Instantiate and start all registered tasks, respecting feature gates.

#stop_all

def stop_all(self) -> None

Stop all running tasks.

#list_tasks

def list_tasks(self) -> list[dict[str, object]]

List registered tasks with their running status.

#get_task

def get_task(self, name: str) -> BackgroundTask | None

Return a running task instance by name, or None.