՞iUdZddlmZddlZddlmZej eZe hdZ de d<Gdd Z d d  dd Z dd Zy)u3 Genesis Task Queue — High-level TaskManager interface. Provides a clean API for enqueueing tasks by name, monitoring queue depths, and inspecting dead-letter state without requiring callers to import Dramatiq actors directly. Usage ----- from core.task_queue import configure_broker, TaskManager configure_broker() manager = TaskManager() manager.enqueue("send_notification", "email", "user@example.com", {"msg": "Hi"}) depths = manager.get_all_queue_depths() Or use the module-level convenience function:: from core.task_queue import enqueue enqueue("process_epoch_task", "2026-W09", queue="default") # VERIFICATION_STAMP # Story: M6.03 — manager.py — TaskManager and enqueue() # Verified By: parallel-builder # Verified At: 2026-02-25T00:00:00Z # Tests: 9/9 # Coverage: 100% ) annotationsN)Any>lowhighdefaultcriticalzfrozenset[str] VALID_QUEUEScLeZdZdZd d dZdd d dZd d dZddZy) TaskManagera High-level interface for the Genesis task queue. Decouples call sites from Dramatiq internals and provides queue depth inspection. Designed to be instantiated per-request or as a singleton. Parameters ---------- broker : optional A pre-configured Dramatiq broker. When ``None`` (the default) the manager uses whatever broker was registered via ``configure_broker()``. Nc||_yN)_broker)selfbrokers 0/mnt/e/genesis-system/core/task_queue/manager.py__init__zTaskManager.__init__7s  rqueuec|tvr%tjd|tty ddlm}t ||d}|tjd|yt|dr|j|||n6t|d r|j|i|ntjd |ytjd ||y #t$rtjd |YywxYw)u Dispatch a named task to the specified queue. Looks up the actor function by name in ``core.task_queue.workers``, then calls ``.send_with_options()`` (preferred — allows queue override) or ``.send()`` as a fallback. Parameters ---------- task_name : str Name of the registered Dramatiq actor (e.g. ``"send_notification"``). *args : Positional arguments forwarded to the actor. queue : str Destination queue. Must be one of ``critical``, ``high``, ``default``, ``low``. Defaults to ``"default"``. **kwargs : Keyword arguments forwarded to the actor. Returns ------- bool ``True`` if the task was dispatched successfully, ``False`` on any error (unknown actor, serialisation failure, broker unreachable). z9TaskManager.enqueue: invalid queue '%s'. Valid queues: %sFr)workersNz^TaskManager.enqueue: unknown task '%s'. Register it as a @actor in core/task_queue/workers.py.send_with_options)argskwargs queue_namesendzWTaskManager.enqueue: '%s' is not a Dramatiq actor (missing .send / .send_with_options).z2TaskManager.enqueue: dispatched '%s' to queue '%s'Tz6TaskManager.enqueue: unexpected error dispatching '%s') r loggererrorsortedcore.task_queuergetattrhasattrrrinfo Exception exception)r task_namerrrractor_fns renqueuezTaskManager.enqueue>s@  $ LLK|$  %  /w 48H M x!45**VPU*V6* t.v. <  KKD      H   s+C AC 2C C,+C,c|jxs t}|y |j}d|}|j|}||SdS#t$rt j d|YywxYw)a: Return the current number of messages in the named queue. Uses the Redis broker's underlying client to call ``LLEN`` on the Dramatiq queue key (``dramatiq:``). Returns ``0`` when the broker is unavailable or an error occurs so callers can treat this as a safe, non-raising call. Parameters ---------- queue_name : str One of ``critical``, ``high``, ``default``, ``low``. Returns ------- int Queue depth (0 if unknown / unavailable). rz dramatiq:z2TaskManager.get_queue_depth: failed for queue '%s')r_get_configured_brokerclientllenr$rr%)rrr redis_clientkeydepths rget_queue_depthzTaskManager.get_queue_depths(9!7!9 > !==Lj\*C %%c*E!-5 41 4    D   s%AAA'&A'cLdDcic]}||j|c}Scc}w)z Return the depth of all 4 priority queues in a single call. Returns ------- dict[str, int] Mapping of queue name to depth, e.g. ``{"critical": 0, "high": 2, "default": 5, "low": 0}``. )rrrr)r0)rqs rget_all_queue_depthsz TaskManager.get_all_queue_depthss)5[[q4''**[[[s!r )rrreturnNone r&strrrrr7rrr4bool)r)rr7r4int)r4zdict[str, int])__name__ __module__ __qualname____doc__rr(r0r3rrr r )sY  MMM M  M  Mf"H \rr rrcBtj|g|d|i|S)ag Module-level convenience wrapper around ``TaskManager.enqueue()``. Instantiates a transient ``TaskManager`` and delegates immediately. Suitable for one-off dispatch calls where holding a long-lived manager instance is unnecessary. Parameters ---------- task_name : str Name of the Dramatiq actor. *args : Positional arguments for the task. queue : str Destination queue (default: ``"default"``). **kwargs : Keyword arguments for the task. Returns ------- bool ``True`` on successful dispatch, ``False`` on failure. r)r r()r&rrrs rr(r(s(: !;=  IT I I& IIrc< ddlm}|S#t$rYywxYw)z=Return the broker registered via configure_broker(), or None.r get_brokerN)core.task_queue.brokerrBr$rAs rr*r*s$5| s  r6)r4r)r= __future__rloggingtypingr getLoggerr:r frozensetr __annotations__r r(r*r>rrrJs:#   8 $ ))OP nPV\V\BJJ J J J  JJr