feat: adds persistence layer for app state and job results [SBK-363]

setup.py

@@ -40,6 +40,9 @@
         "IPython",  # Console for interacting
         "ipdb",  # Debugger (Must use `export PYTHONBREAKPOINT=ipdb.set_trace`)
     ],
+    "mongo": [
+        "beanie~=1.23.6",
+    ],
 }
 
 # NOTE: `pip install -e .[dev]` to install package
@@ -49,6 +52,7 @@
     + extras_require["doc"]
     + extras_require["release"]
     + extras_require["dev"]
+    + extras_require["mongo"]
 )
 
 with open("./README.md") as readme:

silverback/__init__.py

@@ -1,8 +1,10 @@
 from .application import SilverbackApp
 from .exceptions import CircuitBreaker, SilverbackException
+from .types import SilverbackStartupState
 
 __all__ = [
     "CircuitBreaker",
     "SilverbackApp",
     "SilverbackException",
+    "SilverbackStartupState",
 ]

silverback/application.py

@@ -75,19 +75,67 @@ def __init__(self, settings: Optional[Settings] = None):
 
     def on_startup(self) -> Callable:
         """
-        Code to execute on worker startup / restart after an error.
+        Code to execute on one worker upon startup / restart after an error.
 
         Usage example::
 
             @app.on_startup()
             def do_something_on_startup(state):
                 ...  # Can provision resources, or add things to `state`.
         """
-        return self.broker.on_event(TaskiqEvents.WORKER_STARTUP)
+        return self.broker.task(task_name="silverback_startup")
 
     def on_shutdown(self) -> Callable:
         """
-        Code to execute on normal worker shutdown.
+        Code to execute on one worker at shutdown.
+
+        Usage example::
+
+            @app.on_shutdown()
+            def do_something_on_shutdown(state):
+                ...  # Update some external service, perhaps using information from `state`.
+        """
+        return self.broker.task(task_name="silverback_shutdown")
+
+    def on_client_startup(self) -> Callable:
+        """
+        Code to execute on client startup / restart after an error.
+
+        Usage example::
+
+            @app.on_startup()
+            def do_something_on_startup(state):
+                ...  # Can provision resources, or add things to `state`.
+        """
+        return self.broker.on_event(TaskiqEvents.CLIENT_STARTUP)
+
+    def on_client_shutdown(self) -> Callable:
+        """
+        Code to execute on client shutdown.
+
+        Usage example::
+
+            @app.on_shutdown()
+            def do_something_on_shutdown(state):
+                ...  # Update some external service, perhaps using information from `state`.
+        """
+        return self.broker.on_event(TaskiqEvents.CLIENT_SHUTDOWN)
+
+    def on_worker_startup(self) -> Callable:
+        """
+        Code to execute on every worker at startup / restart after an error.
+
+        Usage example::
+
+            @app.on_startup()
+            def do_something_on_startup(state):
+                ...  # Can provision resources, or add things to `state`.
+        """
+        return self.broker.on_event(TaskiqEvents.WORKER_STARTUP)
+
+    def on_worker_shutdown(self) -> Callable:
+        """
+        Code to execute on every worker at shutdown.
 
         Usage example::
 
@@ -97,6 +145,24 @@ def do_something_on_shutdown(state):
         """
         return self.broker.on_event(TaskiqEvents.WORKER_SHUTDOWN)
 
+    def get_startup_handler(self) -> Optional[AsyncTaskiqDecoratedTask]:
+        """
+        Get access to the handler for `silverback_startup` events.
+
+        Returns:
+            Optional[AsyncTaskiqDecoratedTask]: Returns decorated task, if one has been created.
+        """
+        return self.broker.available_tasks.get("silverback_startup")
+
+    def get_shutdown_handler(self) -> Optional[AsyncTaskiqDecoratedTask]:
+        """
+        Get access to the handler for `silverback_shutdown` events.
+
+        Returns:
+            Optional[AsyncTaskiqDecoratedTask]: Returns decorated task, if one has been created.
+        """
+        return self.broker.available_tasks.get("silverback_shutdown")
+
     def get_block_handler(self) -> Optional[AsyncTaskiqDecoratedTask]:
         """
         Get access to the handler for `block` events.

silverback/middlewares.py

@@ -1,8 +1,10 @@
 from ape.logging import logger
-from ape.types import ContractLog, HexBytes
+from ape.types import ContractLog
 from ape.utils import ManagerAccessMixin
 from taskiq import TaskiqMessage, TaskiqMiddleware, TaskiqResult
 
+from silverback.utils import hexbytes_dict
+
 
 class SilverbackMiddleware(TaskiqMiddleware, ManagerAccessMixin):
     def __init__(self, *args, **kwargs):
@@ -47,20 +49,10 @@ def _create_label(self, message: TaskiqMessage) -> str:
         return f"{message.task_name}{args}"
 
     def pre_execute(self, message: TaskiqMessage) -> TaskiqMessage:
-        def fix_dict(data: dict) -> dict:
-            fixed_data = {}
-            for name, value in data.items():
-                if isinstance(value, str) and value.startswith("0x"):
-                    fixed_data[name] = HexBytes(value)
-                else:
-                    fixed_data[name] = value
-
-            return fixed_data
-
         if message.task_name == "block":
             # NOTE: Necessary because we don't know the exact block class
             message.args[0] = self.provider.network.ecosystem.decode_block(
-                fix_dict(message.args[0])
+                hexbytes_dict(message.args[0])
             )
 
         elif "event" in message.task_name: