# Copyright 2025 Softwell S.r.l.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Task capability: the task backbone as a mixin over the base server (◆D22, D16).
``TaskMixin`` is a capability composed over ``BaseServer`` AFTER ``StorageMixin``
(it needs ``server.storage``) and BEFORE ``BaseServer`` (the loop hook must sit
inside the D16 chain so it wraps ``Lifespan``). Its cooperative ``__init__`` peels
``tasks=`` and forwards everything else down the chain (mirroring
``StorageMixin.__init__``); a composition WITHOUT the mixin has no ``tasks``
attribute at all.
``tasks=`` is the on/off switch AND the tuning carrier: absent or truthy → the
``TaskManager`` is armed and the worker loop runs at startup; ``tasks=False`` →
no manager, and the lifespan passes straight through (no loop); a dict —
``{"enabled": ..., "tick_seconds": ..., "mount": ...}`` — peels ``enabled`` and
stashes the rest as ``tasks_config`` for the manager to apply. Storage/session
are on when their mixin is present, so tasks matches: a composed ``AsgiServer``
runs the worker loop out of the box.
``TaskConfigElements`` is the ``config_grammar`` companion (D16: the element is
declared by the class that peels the kwarg). The site recipe composes it
explicitly (``config/elements.py``) so ``server(...).tasks(enabled=,
tick_seconds=, mount=)`` lifts to this mixin's ``tasks=`` kwarg through
``element_kwargs`` — which enforces strict-unknown children (``ConfigError``).
The manager is built LAZILY on first ``tasks`` access, never in ``__init__``: the
cooperative chain runs ``TaskMixin.__init__`` (composed after ``StorageMixin``) from
INSIDE ``StorageMixin.__init__``, before that mixin has assigned ``server.storage`` —
so the manager, which opens its spool over ``server.storage``, cannot be built until
the whole chain has completed. Lazy construction defers it to the first use (the
lifespan hook, or a caller reaching ``server.tasks``), when the server is fully live.
The loop is server-owned, hooked in ``__call__`` — ``lifespan.py`` is NEVER touched
(ratified). ``__call__`` intercepts the ``lifespan`` scope exactly like
``CommunicationMixin.__call__``: pre-receive ``lifespan.startup``, ``manager.start()``,
replay the startup down ``super().__call__`` (so the base ``Lifespan`` still runs the
app hooks and acks the protocol), and ``await manager.stop()`` in ``finally`` when the
protocol completes at shutdown. Every other scope — and the disabled case — passes
straight through.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
from genro_builders.builder import element
from .manager import TaskManager
if TYPE_CHECKING:
from ..types import Message, Receive, Scope, Send
__all__ = ["TaskMixin", "TaskConfigElements"]
[docs]
class TaskConfigElements:
"""Config elements owned by ``TaskMixin`` (its ``config_grammar`` companion).
Recipes compose this mixin explicitly (``config/elements.py``) — no
auto-discovery. ``element_kwargs`` validates children against it
(strict-unknown → ``ConfigError``).
"""
@element(sub_tags="*", parent_tags="server")
def tasks(self) -> None:
"""The task backbone: ``enabled`` (bool, default True — the on/off
switch), ``tick_seconds`` (float — the scheduler tick), ``mount``
(str — explicit task-store mount, overriding the by-keys choice).
Server-domain, so it lives under ``server``. The attributes lift to
the server's ``tasks=`` kwarg as a dict."""
[docs]
class TaskMixin:
"""Task capability mixin, composed after ``StorageMixin`` and before the base.
Constructor kwargs peeled here: ``tasks`` — the on/off switch (default on)
or the ``{enabled, tick_seconds, mount}`` tuning dict lifted from the
``tasks()`` config element. When enabled, builds the ``TaskManager`` and
starts/stops its worker loop around the ASGI ``lifespan`` protocol.
"""
config_grammar = TaskConfigElements
def __init__(self, **kwargs: Any) -> None:
tasks: Any = kwargs.pop("tasks", True)
super().__init__(**kwargs)
if isinstance(tasks, dict):
self._tasks_config: dict[str, Any] = dict(tasks)
self._tasks_enabled: bool = bool(self._tasks_config.pop("enabled", True))
else:
self._tasks_config = {}
self._tasks_enabled = bool(tasks)
self._task_manager: TaskManager | None = None # built lazily (see tasks)
@property
def tasks(self) -> TaskManager:
"""The task manager this server owns (built on first access); disabled is an error.
Lazy: the manager opens its spool over ``server.storage``, which is not yet
set while the cooperative ``__init__`` chain is still running — so the first
access after the server is live builds it.
"""
if not self._tasks_enabled:
raise RuntimeError("tasks are disabled (tasks=False at init)")
if self._task_manager is None:
self._task_manager = TaskManager(self)
return self._task_manager
@property
def tasks_enabled(self) -> bool:
"""Whether the task backbone is armed (``tasks`` not ``False`` at init)."""
return self._tasks_enabled
@property
def tasks_config(self) -> dict[str, Any]:
"""The tuning peeled from a ``tasks=`` dict (``tick_seconds``, ``mount``).
Empty when ``tasks=`` was a plain switch; the ``TaskManager`` applies it
at build time.
"""
return self._tasks_config
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
"""Hook the lifespan to start/stop the worker loop (the D16 proof).
When tasks are enabled the loop starts on ``lifespan.startup`` (replayed to
the base handler so the protocol stays intact) and stops when the protocol
completes at shutdown. Disabled, or any non-lifespan scope, passes straight
through.
"""
if scope["type"] != "lifespan" or not self._tasks_enabled:
await super().__call__(scope, receive, send)
return
manager = self.tasks
startup: Message = await receive()
manager.start()
replayed = False
async def replaying_receive() -> Message:
nonlocal replayed
if not replayed:
replayed = True
return startup
return await receive()
try:
await super().__call__(scope, replaying_receive, send)
finally:
await manager.stop()
if __name__ == "__main__":
from ..application import BaseApplication
from ..server import BaseServer
from ..storage_mixin import StorageMixin
class DemoServer(TaskMixin, StorageMixin, BaseServer):
pass
armed = DemoServer(primary=BaseApplication())
assert armed.tasks_enabled is True
manager = armed.tasks # built lazily here
assert isinstance(manager, TaskManager)
assert armed.tasks is manager # same instance on re-access
disabled = DemoServer(primary=BaseApplication(), tasks=False)
assert disabled.tasks_enabled is False
try:
disabled.tasks
except RuntimeError as error:
assert "disabled" in str(error)
assert not hasattr(BaseServer(primary=BaseApplication()), "tasks_enabled")