src/genro_asgi/tasks/mixin.py¶
Source from this local checkout, regenerated when the reader rebuilds.
Line links use #L<number>; a GitHub line range opens its first line.
1 # Copyright 2025 Softwell S.r.l.2 #3 # Licensed under the Apache License, Version 2.0 (the "License");4 # you may not use this file except in compliance with the License.5 # You may obtain a copy of the License at6 #7 # https://www.apache.org/licenses/LICENSE-2.08 #9 # Unless required by applicable law or agreed to in writing, software10 # distributed under the License is distributed on an "AS IS" BASIS,11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12 # See the License for the specific language governing permissions and13 # limitations under the License.14 15 """Task capability: the task backbone as a mixin over the base server (◆D22, D16).16 17 ``TaskMixin`` is a capability composed over ``BaseServer`` AFTER ``StorageMixin``18 (it needs ``server.storage``) and BEFORE ``BaseServer`` (the loop hook must sit19 inside the D16 chain so it wraps ``Lifespan``). Its cooperative ``__init__`` peels20 ``tasks=`` and forwards everything else down the chain (mirroring21 ``StorageMixin.__init__``); a composition WITHOUT the mixin has no ``tasks``22 attribute at all.23 24 ``tasks=`` is the on/off switch AND the tuning carrier: absent or truthy → the25 ``TaskManager`` is armed and the worker loop runs at startup; ``tasks=False`` →26 no manager, and the lifespan passes straight through (no loop); a dict —27 ``{"enabled": ..., "tick_seconds": ..., "mount": ...}`` — peels ``enabled`` and28 stashes the rest as ``tasks_config`` for the manager to apply. Storage/session29 are on when their mixin is present, so tasks matches: a composed ``AsgiServer``30 runs the worker loop out of the box.31 32 ``TaskGrammar`` is this mixin's grammar companion (D16: the element is declared33 by the class that peels the kwarg). The server's grammar composes it explicitly34 (``config/elements.py``) so ``server(...).tasks(enabled=, tick_seconds=,35 mount=)`` lifts to this mixin's ``tasks=`` kwarg — the element is a void one, so36 a stray child is rejected at the recipe line by the grammar itself.37 38 The manager is built LAZILY on first ``tasks`` access, never in ``__init__``: the39 cooperative chain runs ``TaskMixin.__init__`` (composed after ``StorageMixin``) from40 INSIDE ``StorageMixin.__init__``, before that mixin has assigned ``server.storage`` —41 so the manager, which opens its spool over ``server.storage``, cannot be built until42 the whole chain has completed. Lazy construction defers it to the first use (the43 lifespan hook, or a caller reaching ``server.tasks``), when the server is fully live.44 45 The loop is server-owned, hooked in ``__call__`` — ``lifespan.py`` is NEVER touched46 (ratified). ``__call__`` intercepts the ``lifespan`` scope exactly like47 ``CommunicationMixin.__call__``: pre-receive ``lifespan.startup``, ``manager.start()``,48 replay the startup down ``super().__call__`` (so the base ``Lifespan`` still runs the49 app hooks and acks the protocol), and ``await manager.stop()`` in ``finally`` when the50 protocol completes at shutdown. Every other scope — and the disabled case — passes51 straight through.52 """53 54 from __future__ import annotations55 56 from typing import TYPE_CHECKING, Any57 58 from genro_bag import BagResolver59 from genro_builders.builder import element60 61 from .manager import TaskManager62 63 if TYPE_CHECKING:64 from ..types import Message, Receive, Scope, Send65 66 __all__ = ["TaskGrammar", "TaskMixin"]67 68 69 class TaskGrammar:70 """Config grammar owned by ``TaskMixin`` (the class that peels ``tasks=``).71 72 The server's grammar composes this mixin explicitly73 (``config/elements.py``) — no auto-discovery.74 """75 76 @element(sub_tags="", parent_tags="server")77 def tasks(78 self,79 enabled: bool = True,80 tick_seconds: float | BagResolver = None,81 mount: str | BagResolver = None,82 ) -> None:83 """The task backbone: ``enabled`` (default True — the on/off switch),84 ``tick_seconds`` (the scheduler tick), ``mount`` (explicit task-store85 mount, overriding the by-keys choice). Server-domain, so it lives under86 ``server``. The attributes lift to the server's ``tasks=`` kwarg as a87 dict."""88 89 90 class TaskMixin:91 """Task capability mixin, composed after ``StorageMixin`` and before the base.92 93 Constructor kwargs peeled here: ``tasks`` — the on/off switch (default on)94 or the ``{enabled, tick_seconds, mount}`` tuning dict lifted from the95 ``tasks()`` config element. When enabled, builds the ``TaskManager`` and96 starts/stops its worker loop around the ASGI ``lifespan`` protocol.97 """98 99 grammar: type = TaskGrammar100 101 def __init__(self, **kwargs: Any) -> None:102 tasks: Any = kwargs.pop("tasks", True)103 super().__init__(**kwargs)104 if isinstance(tasks, dict):105 self._tasks_config: dict[str, Any] = dict(tasks)106 self._tasks_enabled: bool = bool(self._tasks_config.pop("enabled", True))107 else:108 self._tasks_config = {}109 self._tasks_enabled = bool(tasks)110 self._task_manager: TaskManager | None = None # built lazily (see tasks)111 112 @property113 def tasks(self) -> TaskManager:114 """The task manager this server owns (built on first access); disabled is an error.115 116 Lazy: the manager opens its spool over ``server.storage``, which is not yet117 set while the cooperative ``__init__`` chain is still running — so the first118 access after the server is live builds it.119 """120 if not self._tasks_enabled:121 raise RuntimeError("tasks are disabled (tasks=False at init)")122 if self._task_manager is None:123 self._task_manager = TaskManager(self)124 return self._task_manager125 126 @property127 def tasks_enabled(self) -> bool:128 """Whether the task backbone is armed (``tasks`` not ``False`` at init)."""129 return self._tasks_enabled130 131 @property132 def tasks_config(self) -> dict[str, Any]:133 """The tuning peeled from a ``tasks=`` dict (``tick_seconds``, ``mount``).134 135 Empty when ``tasks=`` was a plain switch; the ``TaskManager`` applies it136 at build time.137 """138 return self._tasks_config139 140 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:141 """Hook the lifespan to start/stop the worker loop (the D16 proof).142 143 When tasks are enabled the loop starts on ``lifespan.startup`` (replayed to144 the base handler so the protocol stays intact) and stops when the protocol145 completes at shutdown. Disabled, or any non-lifespan scope, passes straight146 through.147 """148 if scope["type"] != "lifespan" or not self._tasks_enabled:149 await super().__call__(scope, receive, send)150 return151 manager = self.tasks152 startup: Message = await receive()153 manager.start()154 replayed = False155 156 async def replaying_receive() -> Message:157 nonlocal replayed158 if not replayed:159 replayed = True160 return startup161 return await receive()162 163 try:164 await super().__call__(scope, replaying_receive, send)165 finally:166 await manager.stop()