refactor(py-sdk): allow a better way to integrate w/ Silverback

pyproject.toml

@@ -29,6 +29,8 @@ lint = [
     "black",
     "isort",
     "mypy",
+    # NOTE: Be able to lint our silverback add-ons
+    "apepay[bot]"
 ]
 test = [
     "ape-foundry",

sdk/py/apepay/manager.py

@@ -2,10 +2,15 @@
 from collections.abc import Iterator
 from datetime import datetime, timedelta
 from functools import partial, wraps
-from typing import Any, Callable, Union, cast
+from typing import TYPE_CHECKING, Any, Callable, Union, cast
 
 from ape.api import ReceiptAPI
-from ape.contracts.base import ContractCallHandler, ContractInstance, ContractTransactionHandler
+from ape.contracts.base import (
+    ContractCallHandler,
+    ContractEvent,
+    ContractInstance,
+    ContractTransactionHandler,
+)
 from ape.exceptions import ContractLogicError, DecodingError
 from ape.types import AddressType, HexBytes
 from ape.utils import BaseInterfaceModel, cached_property
@@ -18,6 +23,10 @@
 from .utils import time_unit_to_timedelta
 from .validators import Validator
 
+if TYPE_CHECKING:
+    # NOTE: We really only use this for type checking, optional install
+    from silverback import SilverbackApp
+
 MAX_DURATION_SECONDS = int(timedelta.max.total_seconds()) - 1
 
 _ValidatorItem = Union[Validator, ContractInstance, AddressType]
@@ -192,6 +201,71 @@ def create(
             is_creation_event=True,
         )
 
+    def _parse_stream_decorator(self, app: "SilverbackApp", container: ContractEvent):
+
+        def decorator(f):
+
+            @app.on_(container)
+            @wraps(f)
+            def inner(log):
+                return f(Stream(manager=self, creator=log.creator, stream_id=log.stream_id))
+
+            return inner
+
+        return decorator
+
+    def on_stream_created(self, app: "SilverbackApp"):
+        """
+        Usage example::
+
+            app = SilverbackApp()
+            sm = StreamManager(address=...)
+
+            sm.on_stream_created(app)
+            def do_something(stream):
+                ...  # Use `stream` to update your infrastructure
+        """
+        return self._parse_stream_decorator(app, self.contract.StreamCreated)
+
+    def on_stream_funded(self, app: "SilverbackApp"):
+        """
+        Usage example::
+
+            app = SilverbackApp()
+            sm = StreamManager(address=...)
+
+            sm.on_stream_funded(app)
+            def do_something(stream):
+                ...  # Use `stream` to update your infrastructure
+        """
+        return self._parse_stream_decorator(app, self.contract.StreamFunded)
+
+    def on_stream_claimed(self, app: "SilverbackApp"):
+        """
+        Usage example::
+
+            app = SilverbackApp()
+            sm = StreamManager(address=...)
+
+            sm.on_stream_claimed(app)
+            def do_something(stream):
+                ...  # Use `stream` to update your infrastructure
+        """
+        return self._parse_stream_decorator(app, self.contract.Claimed)
+
+    def on_stream_cancelled(self, app: "SilverbackApp"):
+        """
+        Usage example::
+
+            app = SilverbackApp()
+            sm = StreamManager(address=...)
+
+            sm.on_stream_cancelled(app)
+            def do_something(stream):
+                ...  # Use `stream` to update your infrastructure
+        """
+        return self._parse_stream_decorator(app, self.contract.StreamCancelled)
+
     def streams_by_creator(self, creator: AddressType) -> Iterator["Stream"]:
         for stream_id in range(self.contract.num_streams(creator)):
             yield Stream(manager=self, creator=creator, stream_id=stream_id)