Rework event and instance id naming pattern

Since Invocation is a Suspension, it's becoming clear that we can have lifecycle events for all suspensions. But "suspension_id" being the id just because everything is a suspension, even when the type is more specialized, seems suboptimal.

Instead, since the event_id is much less likely to be accessed, it can have the name "event_id", leaving the "id" to be the instance id, or aggregate id in event sourcing. This has the benefit of avoiding the "aggregate" verbage, which IMO is not very approachable as a concept. I think it's a little unintuitive that event.id will be the id of the object of the event rather than the event, but I think it's outweighed by the benefits of doing it this way.

Author: ryanhiebert

qio/consumer.py

@@ -42,7 +42,7 @@ def __iter__(self) -> Iterator[Invocation]:
 
     def start(self, invocation: Invocation):
         """Signal that the invocation is starting."""
-        self.__stream.publish(InvocationStarted(invocation_id=invocation.id))
+        self.__stream.publish(InvocationStarted(id=invocation.id))
 
     def suspend(
         self,
@@ -52,10 +52,10 @@ def suspend(
     ):
         """Signal that the invocation has suspended."""
         if suspension:
-            self.__stream.publish(InvocationSuspended(invocation_id=invocation.id))
+            self.__stream.publish(InvocationSuspended(id=invocation.id))
             self.__stream.publish_local(
                 LocalInvocationSuspended(
-                    invocation_id=invocation.id,
+                    id=invocation.id,
                     suspension=suspension,
                     invocation=invocation,
                     generator=generator,
@@ -65,42 +65,32 @@ def suspend(
 
     def resolve(self, invocation: Invocation, generator: Generator, value: Any):
         """Signal that a suspension has resolved to a value."""
-        self.__stream.publish(
-            InvocationContinued(invocation_id=invocation.id, value=value)
-        )
+        self.__stream.publish(InvocationContinued(id=invocation.id, value=value))
         self.__stream.publish_local(
-            LocalInvocationContinued(
-                invocation_id=invocation.id, generator=generator, value=value
-            )
+            LocalInvocationContinued(id=invocation.id, generator=generator, value=value)
         )
         self.__receiver.unpause(self.__invocations[invocation])
 
     def throw(self, invocation: Invocation, generator: Generator, exception: Exception):
         """Signal that a suspension has thrown an exception."""
-        self.__stream.publish(
-            InvocationThrew(invocation_id=invocation.id, exception=exception)
-        )
+        self.__stream.publish(InvocationThrew(id=invocation.id, exception=exception))
         self.__stream.publish_local(
             LocalInvocationThrew(
-                invocation_id=invocation.id, generator=generator, exception=exception
+                id=invocation.id, generator=generator, exception=exception
             )
         )
         self.__receiver.unpause(self.__invocations[invocation])
 
     def resume(self, invocation: Invocation):
         """Signal that the invocation is resuming."""
-        self.__stream.publish(InvocationResumed(invocation_id=invocation.id))
+        self.__stream.publish(InvocationResumed(id=invocation.id))
 
     def succeed(self, invocation: Invocation, value: Any):
         """Signal that the invocation has succeeded."""
-        self.__stream.publish(
-            InvocationSucceeded(invocation_id=invocation.id, value=value)
-        )
+        self.__stream.publish(InvocationSucceeded(id=invocation.id, value=value))
         self.__receiver.finish(self.__invocations.pop(invocation))
 
     def error(self, invocation: Invocation, exception: Exception):
         """Signal that the invocation has errored."""
-        self.__stream.publish(
-            InvocationErrored(invocation_id=invocation.id, exception=exception)
-        )
+        self.__stream.publish(InvocationErrored(id=invocation.id, exception=exception))
         self.__receiver.finish(self.__invocations.pop(invocation))

qio/invocation.py

@@ -74,12 +74,12 @@ def deserialize(serialized: bytes, /) -> Invocation:
 
 @dataclass(eq=False, kw_only=True)
 class InvocationEvent:
-    id: str = field(default_factory=random_id)
+    event_id: str = field(default_factory=random_id)
     timestamp: datetime = field(default_factory=lambda: datetime.now(tz=UTC))
-    invocation_id: str
+    id: str
 
     def __repr__(self):
-        return f"<{type(self).__name__} {self.invocation_id}>"
+        return f"<{type(self).__name__} {self.id}>"
 
 
 @dataclass(eq=False, kw_only=True, repr=False)
@@ -92,10 +92,7 @@ def __repr__(self):
         params_repr = ", ".join(
             (*map(repr, self.args), *(f"{k}={v!r}" for k, v in self.kwargs.items())),
         )
-        return (
-            f"<{type(self).__name__} {self.invocation_id} "
-            f"{self.routine}({params_repr})>"
-        )
+        return f"<{type(self).__name__} {self.id} {self.routine}({params_repr})>"
 
 
 @dataclass(eq=False, kw_only=True, repr=False)
@@ -117,18 +114,15 @@ class LocalInvocationSuspended(BaseInvocationSuspended):
     invocation: Invocation
 
     def __repr__(self):
-        return (
-            f"<{type(self).__name__} {self.invocation_id}"
-            f" suspension={self.suspension!r}>"
-        )
+        return f"<{type(self).__name__} {self.id} suspension={self.suspension!r}>"
 
 
 @dataclass(eq=False, kw_only=True)
 class BaseInvocationContinued(InvocationEvent):
     value: Any
 
     def __repr__(self):
-        return f"<{type(self).__name__} {self.invocation_id} value={self.value!r}>"
+        return f"<{type(self).__name__} {self.id} value={self.value!r}>"
 
 
 @dataclass(eq=False, kw_only=True, repr=False)
@@ -145,9 +139,7 @@ class BaseInvocationThrew(InvocationEvent):
     exception: Exception
 
     def __repr__(self):
-        return (
-            f"<{type(self).__name__} {self.invocation_id} exception={self.exception!r}>"
-        )
+        return f"<{type(self).__name__} {self.id} exception={self.exception!r}>"
 
 
 @dataclass(eq=False, kw_only=True, repr=False)
@@ -172,14 +164,12 @@ class InvocationSucceeded(InvocationCompleted):
     value: Any
 
     def __repr__(self):
-        return f"<{type(self).__name__} {self.invocation_id} value={self.value!r}>"
+        return f"<{type(self).__name__} {self.id} value={self.value!r}>"
 
 
 @dataclass(eq=False, kw_only=True)
 class InvocationErrored(InvocationCompleted):
     exception: Exception
 
     def __repr__(self):
-        return (
-            f"<{type(self).__name__} {self.invocation_id} exception={self.exception!r}>"
-        )
+        return f"<{type(self).__name__} {self.id} exception={self.exception!r}>"

qio/monitor.py

@@ -63,51 +63,51 @@ def handle_invocation_event(
         match event:
             case InvocationSubmitted():
                 table.add_row(
-                    event.invocation_id,
+                    event.id,
                     event.routine,
                     "Submitted",
                     self.__qio.routine(event.routine).queue,
-                    key=event.invocation_id,
+                    key=event.id,
                 )
             case InvocationStarted():
                 table.update_cell(
-                    event.invocation_id,
+                    event.id,
                     self.__column_keys[2],
                     "Started",
                 )
             case InvocationSuspended():
                 table.update_cell(
-                    event.invocation_id,
+                    event.id,
                     self.__column_keys[2],
                     "Suspended",
                 )
             case InvocationContinued():
                 table.update_cell(
-                    event.invocation_id,
+                    event.id,
                     self.__column_keys[2],
                     "Continued",
                 )
             case InvocationThrew():
                 table.update_cell(
-                    event.invocation_id,
+                    event.id,
                     self.__column_keys[2],
                     "Threw",
                 )
             case InvocationResumed():
                 table.update_cell(
-                    event.invocation_id,
+                    event.id,
                     self.__column_keys[2],
                     "Resumed",
                 )
             case InvocationSucceeded():
                 table.update_cell(
-                    event.invocation_id,
+                    event.id,
                     self.__column_keys[2],
                     "Succeeded",
                 )
             case InvocationErrored():
                 table.update_cell(
-                    event.invocation_id,
+                    event.id,
                     self.__column_keys[2],
                     "Errored",
                 )

qio/qio.py

@@ -129,13 +129,11 @@ def resolver():
                     break
 
                 match event:
-                    case InvocationSucceeded(invocation_id=invocation_id, value=value):
+                    case InvocationSucceeded(id=invocation_id, value=value):
                         if invocation_id in waiting:
                             future = waiting.pop(invocation_id)
                             future.set_result(value)
-                    case InvocationErrored(
-                        invocation_id=invocation_id, exception=exception
-                    ):
+                    case InvocationErrored(id=invocation_id, exception=exception):
                         if invocation_id in waiting:
                             future = waiting.pop(invocation_id)
                             future.set_exception(exception)
@@ -161,7 +159,7 @@ def submit(self, invocation: Invocation, /):
         routine = self.routine(invocation.routine)
         self.__stream.publish(
             InvocationSubmitted(
-                invocation_id=invocation.id,
+                id=invocation.id,
                 routine=invocation.routine,
                 args=invocation.args,
                 kwargs=invocation.kwargs,

qio/sample_test.py

@@ -27,7 +27,7 @@ def test_integration():
         )
         try:
             while event := events.get():
-                if event.invocation_id == invocation.id:
+                if event.id == invocation.id:
                     break
         finally:
             if worker.poll() is None:  # Process is still running