Split Backend into Broker/ResultBackend/EventBackend

Author: levsh

README.md

@@ -77,11 +77,9 @@ import tasks
 logger = logging.getLogger("arrlio")
 logger.setLevel("INFO")
 
-BACKEND = "arrlio.backends.local"
-# BACKEND = "arrlio.backends.rabbitmq"
 
 async def main():
-    app = arrlio.App(arrlio.Config(backend={"module": BACKEND}))
+    app = arrlio.App(arrlio.Config())
 
     async with app:
         await app.consume_tasks()
@@ -103,7 +101,7 @@ async def main():
             ar = await app.send_task(tasks.exception)
             logger.info(await ar.get())
         except Exception as e:
-            print(f"\nThis is example exception for {app.backend}:\n")
+            print(f"\nThis is example exception for {app.result_backend}:\n")
             logger.exception(e)
             print()
 
@@ -130,9 +128,6 @@ import tasks
 logger = logging.getLogger("arrlio")
 logger.setLevel("INFO")
 
-BACKEND = "arrlio.backends.local"
-# BACKEND = "arrlio.backends.rabbitmq"
-
 
 async def main():
     graph = arrlio.Graph("My Graph")
@@ -146,7 +141,6 @@ async def main():
 	# plugins are required
     app = arrlio.App(
         arrlio.Config(
-            backend={"module": BACKEND},
             plugins=[
                 {"module": "arrlio.plugins.events"},
                 {"module": "arrlio.plugins.graphs"},
@@ -178,9 +172,6 @@ import tasks
 logger = logging.getLogger("arrlio")
 logger.setLevel("INFO")
 
-BACKEND = "arrlio.backends.local"
-# BACKEND = "arrlio.backends.rabbitmq"
-
 
 async def main():
     graph = arrlio.Graph("My Graph")
@@ -190,7 +181,6 @@ async def main():
 
     app = arrlio.App(
         arrlio.Config(
-            backend={"module": BACKEND},
             plugins=[
                 {"module": "arrlio.plugins.events"},
                 {"module": "arrlio.plugins.graphs"},

arrlio/__init__.py

@@ -2,11 +2,12 @@
 import logging
 import sys
 
+
 __version__ = importlib.metadata.version("arrlio")
 
 logger = logging.getLogger("arrlio")
 
-log_frmt = logging.Formatter("%(asctime)s %(levelname)-8s %(name)-27s lineno:%(lineno)4d -- %(message)s")
+log_frmt = logging.Formatter("%(asctime)s %(levelname)-8s %(name)-40s lineno:%(lineno)4d -- %(message)s")
 log_hndl = logging.StreamHandler(stream=sys.stderr)
 log_hndl.setFormatter(log_frmt)
 logger.addHandler(log_hndl)
@@ -17,4 +18,5 @@
 from arrlio.models import Event, Graph, Task, TaskInstance, TaskResult  # noqa
 from arrlio.settings import LOG_LEVEL
 
+
 logger.setLevel(LOG_LEVEL)

arrlio/abc.py

@@ -0,0 +1,97 @@
+from abc import ABC, abstractmethod
+from typing import Any, AsyncGenerator, Callable, Coroutine
+
+from arrlio.models import Event, Shared, TaskInstance, TaskResult
+
+
+class AbstractClosable(ABC):
+    @property
+    @abstractmethod
+    def is_closed(self) -> bool:
+        raise NotImplementedError
+
+    @abstractmethod
+    async def init(self):
+        raise NotImplementedError
+
+    @abstractmethod
+    async def close(self):
+        raise NotImplementedError
+
+    async def __aenter__(self):
+        await self.init()
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb):
+        await self.close()
+
+
+class AbstractBroker(AbstractClosable, ABC):
+    @abstractmethod
+    async def send_task(self, task_instance: TaskInstance, **kwds):
+        """Send task to backend."""
+        raise NotImplementedError
+
+    @abstractmethod
+    async def consume_tasks(self, queues: list[str], callback: Callable[[TaskInstance], Coroutine]):
+        """Consume tasks from the queues and invoke `callback` on `arrlio.models.TaskInstance` received."""
+        raise NotImplementedError
+
+    @abstractmethod
+    async def stop_consume_tasks(self, queues: list[str] | None = None):
+        """Stop consuming tasks."""
+        raise NotImplementedError
+
+
+class AbstractResultBackend(AbstractClosable, ABC):
+    @abstractmethod
+    def make_headers(self, task_instance: TaskInstance) -> dict:
+        """Make result backend headers for `arrlio.models.TaskInstance`."""
+        raise NotImplementedError
+
+    @abstractmethod
+    def make_shared(self, task_instance: TaskInstance) -> Shared:
+        """Make result backend shared for `arrio.models.TaskInstance`."""
+        raise NotImplementedError
+
+    @abstractmethod
+    async def allocate_storage(self, task_instance: TaskInstance):
+        """Allocate storage for future task result if needed."""
+        raise NotImplementedError
+
+    @abstractmethod
+    async def push_task_result(self, task_result: TaskResult, task_instance: TaskInstance):
+        """Push `arrlio.models.TaskResult` for `arrlio.models.TaskInstance`."""
+        raise NotImplementedError
+
+    @abstractmethod
+    async def pop_task_result(self, task_instance: TaskInstance) -> AsyncGenerator[TaskResult, None]:
+        """Pop `arrlio.models.TaskResult` for `arrlio.models.TaskInstance`."""
+        raise NotImplementedError
+
+    @abstractmethod
+    async def close_task(self, task_instance: TaskInstance, idx: tuple[str, int] | None = None):
+        """Close task."""
+        raise NotImplementedError
+
+
+class AbstractEventBackend(AbstractClosable, ABC):
+    @abstractmethod
+    async def send_event(self, event: Event):
+        """Send `arrlio.models.Event`."""
+        raise NotImplementedError
+
+    @abstractmethod
+    async def consume_events(
+        self,
+        callback_id: str,
+        callback: Callable[[Event], Any],
+        event_types: list[str] | None = None,
+    ):
+        """Consume events and invoke `callback` on `arrlio.models.Event` received."""
+        raise NotImplementedError
+
+    @abstractmethod
+    async def stop_consume_events(self, callback_id: str | None = None):
+        """Stop consuming events."""
+        raise NotImplementedError

arrlio/backends/base.py

@@ -0,0 +1 @@
+from . import local, rabbitmq