Reorg

dev/install → .dev/install

@@ -1,3 +1,3 @@
 #!/usr/bin/env bash
 pip install -U pip poetry
-poetry install
+poetry install --all-extras

dev/lint → .dev/lint

@@ -12,6 +12,6 @@ fi
 echo "run black"
 black fluid tests ${BLACK_ARG}
 echo "run ruff"
-ruff fluid tests ${RUFF_ARG}
+ruff check fluid tests ${RUFF_ARG}
 echo "run mypy"
-mypy fluid/scheduler tests/scheduler
+mypy fluid tests

.github/workflows/build.yml

@@ -16,13 +16,13 @@ jobs:
       PYPI_PASSWORD: ${{ secrets.PYPI_PASSWORD }}
     strategy:
       matrix:
-        python-version: ["3.10", "3.11"]
+        python-version: ["3.11", "3.12"]
 
     steps:
       - name: checkout code
-        uses: actions/checkout@v2
+        uses: actions/checkout@v3
       - name: Set up Python ${{ matrix.python-version }}
-        uses: actions/setup-python@v2
+        uses: actions/setup-python@v3
         with:
           python-version: ${{ matrix.python-version }}
       - name: Start Redis

Makefile

@@ -14,16 +14,12 @@ clean:			## remove python cache files
 	rm -rf .coverage
 
 
-install: 		## install packages in virtualenv
-	@./dev/install
+install: 		## install dev packages via poetry
+	@./.dev/install
 
 
 lint: 			## run linters
-	poetry run ./dev/lint fix
-
-
-mypy:			## run mypy
-	@poetry run mypy fluid
+	poetry run ./.dev/lint fix
 
 
 test:			## test with coverage

fluid/__init__.py

@@ -1,2 +1,3 @@
 """Reusable server side python modules"""
+
 __version__ = "0.7.0"

fluid/scheduler/cpubound.py

@@ -64,8 +64,7 @@ def __call__(self, executor: TaskExecutor) -> Task:
 
 
 @overload
-def cpu_task(executor: TaskExecutor) -> Task:
-    ...
+def cpu_task(executor: TaskExecutor) -> Task: ...
 
 
 @overload
@@ -77,8 +76,7 @@ def cpu_task(
     randomize: RandomizeType | None = None,
     max_concurrency: int = 1,
     priority: TaskPriority = TaskPriority.medium,
-) -> CpuTaskConstructor:
-    ...
+) -> CpuTaskConstructor: ...
 
 
 def cpu_task(

fluid/scheduler/crontab.py

@@ -1,5 +1,6 @@
 """Originally from https://github.com/coleifer/huey
 """
+
 import re
 from abc import ABC, abstractmethod
 from datetime import datetime

fluid/scheduler/task.py

@@ -3,24 +3,19 @@
 import inspect
 import logging
 import os
-from importlib import import_module
 from typing import (
     TYPE_CHECKING,
     Any,
     Callable,
     Coroutine,
     NamedTuple,
     TypeVar,
-    cast,
     overload,
 )
 from uuid import uuid4
 
-from openapi.spec.utils import trim_docstring
-
 from fluid import log
-from fluid.tools_aiohttp.node import WorkerApplication
-from fluid.utils import microseconds
+from fluid.tools.text import trim_docstring
 
 from .constants import TaskPriority
 from .crontab import Scheduler
@@ -49,18 +44,6 @@ class ImproperlyConfigured(RuntimeError):
     pass
 
 
-def create_task_app() -> WorkerApplication:
-    if not TASK_MANAGER_APP:
-        raise ImproperlyConfigured("missing TASK_MANAGER_APP environment variable")
-    bits = TASK_MANAGER_APP.split(":")
-    if len(bits) != 2:
-        raise ImproperlyConfigured(
-            "TASK_MANAGER_APP must be of the form <module>:<function>"
-        )
-    mod = import_module(bits[0])
-    return cast(WorkerApplication, getattr(mod, bits[1])())
-
-
 class TaskContext(NamedTuple):
     task_manager: "TaskManager"
     task_run: TaskRun
@@ -111,16 +94,6 @@ class Task(NamedTuple):
     async def register(self, broker: "Broker") -> None:
         pass
 
-    async def __call__(
-        self,
-        task_manager: TaskManager | None = None,
-        **kwargs: Any,
-    ) -> Any:
-        if task_manager is None:
-            task_manager = create_task_app()["task_manager"]
-        context = self.create_context(task_manager, **kwargs)
-        return await self.executor(context)
-
     def create_context(
         self,
         task_manager: TaskManager,
@@ -144,8 +117,7 @@ def create_context(
 
 
 @overload
-def task(executor: TaskExecutor) -> Task:
-    ...
+def task(executor: TaskExecutor) -> Task: ...
 
 
 @overload
@@ -157,8 +129,7 @@ def task(
     randomize: RandomizeType | None = None,
     max_concurrency: int = 1,
     priority: TaskPriority = TaskPriority.medium,
-) -> TaskConstructor:
-    ...
+) -> TaskConstructor: ...
 
 
 # implementation of the task decorator

fluid/settings.py

@@ -1,11 +1,21 @@
 import os
 
-from .tools.text import str2bool
+from .tools.text import to_bool
 
 # Workers
 STOPPING_GRACE_PERIOD: int = int(os.getenv("STOPPING_GRACE_PERIOD") or "10")
 
+# Redis
+
 # Database
 DBPOOL_MAX_SIZE: int = int(os.getenv("DBPOOL_MAX_SIZE") or "10")
 DBPOOL_MAX_OVERFLOW: int = int(os.getenv("DBPOOL_MAX_OVERFLOW") or "10")
-DBECHO: bool = str2bool(os.getenv("DBECHO") or "no")
+DBECHO: bool = to_bool(os.getenv("DBECHO") or "no")
+
+
+# Flamegraph
+STACK_SAMPLER_PERIOD: int = int(os.getenv("STACK_SAMPLER_PERIOD", "60"))
+FLAMEGRAPH_EXECUTABLE: str = os.getenv("FLAMEGRAPH_EXECUTABLE") or "/bin/flamegraph.pl"
+FLAMEGRAPH_DATA_BUCKET, FLAMEGRAPH_DATA_PATH = os.getenv(
+    "FLAMEGRAPH_DATA_BUCKET_PATH", "replace/me"
+).split("/")

fluid/tools/dispatcher.py

@@ -1,15 +1,15 @@
 import asyncio
 from abc import ABC, abstractmethod
 from collections import defaultdict
-from typing import Any, Awaitable, Callable, Dict, Generic, TypeVar
+from typing import Awaitable, Callable, Dict, Generic, TypeVar
 
 MessageType = TypeVar("MessageType")
 MessageHandlerType = TypeVar("MessageHandlerType")
 
 
 class BaseDispatcher(Generic[MessageType, MessageHandlerType], ABC):
     def __init__(self) -> None:
-        self._msg_handlers: Dict[str, Dict[str, MessageHandlerType]] = defaultdict(
+        self._msg_handlers: Dict[str, dict[str, MessageHandlerType]] = defaultdict(
             dict,
         )
 
@@ -18,11 +18,20 @@ def register_handler(
         message_type: str,
         handler: MessageHandlerType,
         tag: str = "",
-    ) -> None:
+    ) -> MessageHandlerType | None:
+        previous = self._msg_handlers[message_type].get(tag)
         self._msg_handlers[message_type][tag] = handler
+        return previous
 
-    def unregister_handler(self, message_type: str, tag: str = "") -> None:
-        self._msg_handlers[message_type].pop(tag)
+    def unregister_handler(
+        self, message_type: str, tag: str = ""
+    ) -> MessageHandlerType | None:
+        return self._msg_handlers[message_type].pop(tag, None)
+
+    def on(
+        self, handler: MessageHandlerType, tag: str = ""
+    ) -> MessageHandlerType | None:
+        return self.register_handler("*", handler, tag)
 
     def get_handlers(
         self,
@@ -45,10 +54,6 @@ def dispatch(self, message: MessageType) -> int:
                 handler(message)
         return len(handlers or ())
 
-    @abstractmethod
-    def create_message(self, **kwargs: Any) -> MessageType:
-        """create a message"""
-
 
 class AsyncDispatcher(
     BaseDispatcher[MessageType, Callable[[MessageType], Awaitable[None]]],
@@ -59,3 +64,8 @@ async def dispatch(self, message: MessageType) -> int:
         if handlers:
             await asyncio.gather(*[handler(message) for handler in handlers.values()])
         return len(handlers or ())
+
+
+class SimpleDispatcher(Dispatcher[MessageType]):
+    def message_type(self, message: MessageType) -> str:
+        return "*"