Update Debouncer API (#462)

On reflection, the debounce key should be passed into the `debounce`
function instead of the Debouncer constructor because a Debouncer may be
used with multiple different keys (for example, if the key is customer
ID, the same debouncer may dynamically debounce different customers).

Author: kraftp

dbos/_debouncer.py

@@ -147,7 +147,6 @@ def __init__(
         self,
         workflow_name: str,
         *,
-        debounce_key: str,
         debounce_timeout_sec: Optional[float] = None,
         queue: Optional[Queue] = None,
     ):
@@ -157,13 +156,11 @@ def __init__(
             "queue_name": queue.name if queue else None,
             "workflow_name": workflow_name,
         }
-        self.debounce_key = debounce_key
 
     @staticmethod
     def create(
         workflow: Callable[P, R],
         *,
-        debounce_key: str,
         debounce_timeout_sec: Optional[float] = None,
         queue: Optional[Queue] = None,
     ) -> "Debouncer[P, R]":
@@ -172,7 +169,6 @@ def create(
             raise TypeError("Only workflow functions may be debounced, not methods")
         return Debouncer[P, R](
             get_dbos_func_name(workflow),
-            debounce_key=debounce_key,
             debounce_timeout_sec=debounce_timeout_sec,
             queue=queue,
         )
@@ -181,7 +177,6 @@ def create(
     def create_async(
         workflow: Callable[P, Coroutine[Any, Any, R]],
         *,
-        debounce_key: str,
         debounce_timeout_sec: Optional[float] = None,
         queue: Optional[Queue] = None,
     ) -> "Debouncer[P, R]":
@@ -190,13 +185,16 @@ def create_async(
             raise TypeError("Only workflow functions may be debounced, not methods")
         return Debouncer[P, R](
             get_dbos_func_name(workflow),
-            debounce_key=debounce_key,
             debounce_timeout_sec=debounce_timeout_sec,
             queue=queue,
         )
 
     def debounce(
-        self, debounce_period_sec: float, *args: P.args, **kwargs: P.kwargs
+        self,
+        debounce_key: str,
+        debounce_period_sec: float,
+        *args: P.args,
+        **kwargs: P.kwargs,
     ) -> "WorkflowHandle[R]":
         from dbos._dbos import DBOS, _get_dbos_instance
 
@@ -232,9 +230,7 @@ def assign_debounce_ids() -> tuple[str, str]:
         while True:
             try:
                 # Attempt to enqueue a debouncer for this workflow.
-                deduplication_id = (
-                    f"{self.options['workflow_name']}-{self.debounce_key}"
-                )
+                deduplication_id = f"{self.options['workflow_name']}-{debounce_key}"
                 with SetEnqueueOptions(deduplication_id=deduplication_id):
                     with SetWorkflowTimeout(None):
                         internal_queue.enqueue(
@@ -284,6 +280,7 @@ def get_deduplicated_workflow() -> Optional[str]:
 
     async def debounce_async(
         self,
+        debounce_key: str,
         debounce_period_sec: float,
         *args: P.args,
         **kwargs: P.kwargs,
@@ -292,7 +289,7 @@ async def debounce_async(
 
         dbos = _get_dbos_instance()
         handle = await asyncio.to_thread(
-            self.debounce, debounce_period_sec, *args, **kwargs
+            self.debounce, debounce_key, debounce_period_sec, *args, **kwargs
         )
         return WorkflowHandleAsyncPolling(handle.workflow_id, dbos)
 
@@ -304,7 +301,6 @@ def __init__(
         client: DBOSClient,
         workflow_options: EnqueueOptions,
         *,
-        debounce_key: str,
         debounce_timeout_sec: Optional[float] = None,
         queue: Optional[Queue] = None,
     ):
@@ -314,11 +310,10 @@ def __init__(
             "queue_name": queue.name if queue else None,
             "workflow_name": workflow_options["workflow_name"],
         }
-        self.debounce_key = debounce_key
         self.client = client
 
     def debounce(
-        self, debounce_period_sec: float, *args: Any, **kwargs: Any
+        self, debounce_key: str, debounce_period_sec: float, *args: Any, **kwargs: Any
     ) -> "WorkflowHandle[R]":
 
         ctxOptions: ContextOptions = {
@@ -337,7 +332,7 @@ def debounce(
             try:
                 # Attempt to enqueue a debouncer for this workflow.
                 deduplication_id = (
-                    f"{self.debouncer_options['workflow_name']}-{self.debounce_key}"
+                    f"{self.debouncer_options['workflow_name']}-{debounce_key}"
                 )
                 debouncer_options: EnqueueOptions = {
                     "workflow_name": DEBOUNCER_WORKFLOW_NAME,
@@ -390,10 +385,10 @@ def debounce(
                     )
 
     async def debounce_async(
-        self, debounce_period_sec: float, *args: Any, **kwargs: Any
+        self, deboucne_key: str, debounce_period_sec: float, *args: Any, **kwargs: Any
     ) -> "WorkflowHandleAsync[R]":
         handle: "WorkflowHandle[R]" = await asyncio.to_thread(
-            self.debounce, debounce_period_sec, *args, **kwargs
+            self.debounce, deboucne_key, debounce_period_sec, *args, **kwargs
         )
         return WorkflowHandleClientAsyncPolling[R](
             handle.workflow_id, self.client._sys_db

tests/test_debouncer.py

@@ -33,18 +33,18 @@ def debouncer_test() -> None:
 
         debounce_period = 2
 
-        debouncer = Debouncer.create(workflow, debounce_key="key")
-        first_handle = debouncer.debounce(debounce_period, first_value)
-        debouncer = Debouncer.create(workflow, debounce_key="key")
-        second_handle = debouncer.debounce(debounce_period, second_value)
+        debouncer = Debouncer.create(workflow)
+        first_handle = debouncer.debounce("key", debounce_period, first_value)
+        debouncer = Debouncer.create(workflow)
+        second_handle = debouncer.debounce("key", debounce_period, second_value)
         assert first_handle.workflow_id == second_handle.workflow_id
         assert first_handle.get_result() == second_value
         assert second_handle.get_result() == second_value
 
-        debouncer = Debouncer.create(workflow, debounce_key="key")
-        third_handle = debouncer.debounce(debounce_period, third_value)
-        debouncer = Debouncer.create(workflow, debounce_key="key")
-        fourth_handle = debouncer.debounce(debounce_period, fourth_value)
+        debouncer = Debouncer.create(workflow)
+        third_handle = debouncer.debounce("key", debounce_period, third_value)
+        debouncer = Debouncer.create(workflow)
+        fourth_handle = debouncer.debounce("key", debounce_period, fourth_value)
         assert third_handle.workflow_id != first_handle.workflow_id
         assert third_handle.workflow_id == fourth_handle.workflow_id
         assert third_handle.get_result() == fourth_value
@@ -53,7 +53,7 @@ def debouncer_test() -> None:
         # Test SetWorkflowID works
         wfid = generate_uuid()
         with SetWorkflowID(wfid):
-            handle = debouncer.debounce(debounce_period, first_value)
+            handle = debouncer.debounce("key", debounce_period, first_value)
         assert handle.workflow_id == wfid
         assert handle.get_result() == first_value
 
@@ -83,19 +83,18 @@ def workflow(x: int) -> int:
     # Set a huge period but small timeout, verify workflows start after the timeout
     debouncer = Debouncer.create(
         workflow,
-        debounce_key="key",
         debounce_timeout_sec=2,
     )
     long_debounce_period = 10000000
 
-    first_handle = debouncer.debounce(long_debounce_period, first_value)
-    second_handle = debouncer.debounce(long_debounce_period, second_value)
+    first_handle = debouncer.debounce("key", long_debounce_period, first_value)
+    second_handle = debouncer.debounce("key", long_debounce_period, second_value)
     assert first_handle.workflow_id == second_handle.workflow_id
     assert first_handle.get_result() == second_value
     assert second_handle.get_result() == second_value
 
-    third_handle = debouncer.debounce(long_debounce_period, third_value)
-    fourth_handle = debouncer.debounce(long_debounce_period, fourth_value)
+    third_handle = debouncer.debounce("key", long_debounce_period, third_value)
+    fourth_handle = debouncer.debounce("key", long_debounce_period, fourth_value)
     assert third_handle.workflow_id != first_handle.workflow_id
     assert third_handle.workflow_id == fourth_handle.workflow_id
     assert third_handle.get_result() == fourth_value
@@ -104,12 +103,11 @@ def workflow(x: int) -> int:
     # Submit first with a long period then with a short one, verify workflows start on time
     debouncer = Debouncer.create(
         workflow,
-        debounce_key="key",
     )
     short_debounce_period = 1
 
-    first_handle = debouncer.debounce(long_debounce_period, first_value)
-    second_handle = debouncer.debounce(short_debounce_period, second_value)
+    first_handle = debouncer.debounce("key", long_debounce_period, first_value)
+    second_handle = debouncer.debounce("key", short_debounce_period, second_value)
     assert fourth_handle.workflow_id != first_handle.workflow_id
     assert first_handle.workflow_id == second_handle.workflow_id
     assert first_handle.get_result() == second_value
@@ -125,14 +123,14 @@ def workflow(x: int) -> int:
     first_value, second_value, third_value, fourth_value = 0, 1, 2, 3
 
     # Set a huge period but small timeout, verify workflows start after the timeout
-    debouncer_one = Debouncer.create(workflow, debounce_key="key_one")
-    debouncer_two = Debouncer.create(workflow, debounce_key="key_two")
+    debouncer_one = Debouncer.create(workflow)
+    debouncer_two = Debouncer.create(workflow)
     debounce_period = 2
 
-    first_handle = debouncer_one.debounce(debounce_period, first_value)
-    second_handle = debouncer_one.debounce(debounce_period, second_value)
-    third_handle = debouncer_two.debounce(debounce_period, third_value)
-    fourth_handle = debouncer_two.debounce(debounce_period, fourth_value)
+    first_handle = debouncer_one.debounce("key_one", debounce_period, first_value)
+    second_handle = debouncer_one.debounce("key_one", debounce_period, second_value)
+    third_handle = debouncer_two.debounce("key_two", debounce_period, third_value)
+    fourth_handle = debouncer_two.debounce("key_two", debounce_period, fourth_value)
     assert first_handle.workflow_id == second_handle.workflow_id
     assert first_handle.workflow_id != third_handle.workflow_id
     assert third_handle.workflow_id == fourth_handle.workflow_id
@@ -151,20 +149,20 @@ def workflow(x: int) -> int:
     first_value, second_value, third_value, fourth_value = 0, 1, 2, 3
     queue = Queue("test-queue")
 
-    debouncer = Debouncer.create(workflow, debounce_key="key", queue=queue)
+    debouncer = Debouncer.create(workflow, queue=queue)
     debounce_period_sec = 2
 
-    first_handle = debouncer.debounce(debounce_period_sec, first_value)
-    second_handle = debouncer.debounce(debounce_period_sec, second_value)
+    first_handle = debouncer.debounce("key", debounce_period_sec, first_value)
+    second_handle = debouncer.debounce("key", debounce_period_sec, second_value)
     assert first_handle.workflow_id == second_handle.workflow_id
     assert first_handle.get_result() == second_value
     assert second_handle.get_result() == second_value
     assert second_handle.get_status().queue_name == queue.name
 
     # Test SetWorkflowTimeout works
     with SetWorkflowTimeout(5.0):
-        third_handle = debouncer.debounce(debounce_period_sec, third_value)
-        fourth_handle = debouncer.debounce(debounce_period_sec, fourth_value)
+        third_handle = debouncer.debounce("key", debounce_period_sec, third_value)
+        fourth_handle = debouncer.debounce("key", debounce_period_sec, fourth_value)
     assert third_handle.workflow_id != first_handle.workflow_id
     assert third_handle.workflow_id == fourth_handle.workflow_id
     assert third_handle.get_result() == fourth_value
@@ -176,7 +174,7 @@ def workflow(x: int) -> int:
     # Test SetWorkflowID works
     wfid = str(uuid.uuid4())
     with SetWorkflowID(wfid):
-        handle = debouncer.debounce(debounce_period_sec, first_value)
+        handle = debouncer.debounce("key", debounce_period_sec, first_value)
     assert handle.workflow_id == wfid
     assert handle.get_result() == first_value
     assert handle.get_status().queue_name == queue.name
@@ -187,7 +185,7 @@ def workflow(x: int) -> int:
     with SetEnqueueOptions(
         priority=1, deduplication_id="test", app_version=test_version
     ):
-        handle = debouncer.debounce(debounce_period_sec, first_value)
+        handle = debouncer.debounce("key", debounce_period_sec, first_value)
     assert handle.get_result() == first_value
     assert handle.get_status().queue_name == queue.name
     assert handle.get_status().app_version == test_version
@@ -202,17 +200,25 @@ async def workflow_async(x: int) -> int:
 
     first_value, second_value, third_value, fourth_value = 0, 1, 2, 3
 
-    debouncer = Debouncer.create_async(workflow_async, debounce_key="key")
+    debouncer = Debouncer.create_async(workflow_async)
     debounce_period_sec = 2
 
-    first_handle = await debouncer.debounce_async(debounce_period_sec, first_value)
-    second_handle = await debouncer.debounce_async(debounce_period_sec, second_value)
+    first_handle = await debouncer.debounce_async(
+        "key", debounce_period_sec, first_value
+    )
+    second_handle = await debouncer.debounce_async(
+        "key", debounce_period_sec, second_value
+    )
     assert first_handle.workflow_id == second_handle.workflow_id
     assert await first_handle.get_result() == second_value
     assert await second_handle.get_result() == second_value
 
-    third_handle = await debouncer.debounce_async(debounce_period_sec, third_value)
-    fourth_handle = await debouncer.debounce_async(debounce_period_sec, fourth_value)
+    third_handle = await debouncer.debounce_async(
+        "key", debounce_period_sec, third_value
+    )
+    fourth_handle = await debouncer.debounce_async(
+        "key", debounce_period_sec, fourth_value
+    )
     assert third_handle.workflow_id != first_handle.workflow_id
     assert third_handle.workflow_id == fourth_handle.workflow_id
     assert await third_handle.get_result() == fourth_value
@@ -232,24 +238,24 @@ def workflow(x: int) -> int:
         "workflow_name": workflow.__qualname__,
         "queue_name": queue.name,
     }
-    debouncer = DebouncerClient(client, options, debounce_key="key")
+    debouncer = DebouncerClient(client, options)
     debounce_period_sec = 2
 
     first_handle: WorkflowHandle[int] = debouncer.debounce(
-        debounce_period_sec, first_value
+        "key", debounce_period_sec, first_value
     )
     second_handle: WorkflowHandle[int] = debouncer.debounce(
-        debounce_period_sec, second_value
+        "key", debounce_period_sec, second_value
     )
     assert first_handle.workflow_id == second_handle.workflow_id
     assert first_handle.get_result() == second_value
     assert second_handle.get_result() == second_value
 
     third_handle: WorkflowHandle[int] = debouncer.debounce(
-        debounce_period_sec, third_value
+        "key", debounce_period_sec, third_value
     )
     fourth_handle: WorkflowHandle[int] = debouncer.debounce(
-        debounce_period_sec, fourth_value
+        "key", debounce_period_sec, fourth_value
     )
     assert third_handle.workflow_id != first_handle.workflow_id
     assert third_handle.workflow_id == fourth_handle.workflow_id
@@ -258,7 +264,9 @@ def workflow(x: int) -> int:
 
     wfid = str(uuid.uuid4())
     options["workflow_id"] = wfid
-    handle: WorkflowHandle[int] = debouncer.debounce(debounce_period_sec, first_value)
+    handle: WorkflowHandle[int] = debouncer.debounce(
+        "key", debounce_period_sec, first_value
+    )
     assert handle.workflow_id == wfid
     assert handle.get_result() == first_value
 
@@ -277,24 +285,24 @@ async def workflow_async(x: int) -> int:
         "workflow_name": workflow_async.__qualname__,
         "queue_name": queue.name,
     }
-    debouncer = DebouncerClient(client, options, debounce_key="key")
+    debouncer = DebouncerClient(client, options)
     debounce_period_sec = 2
 
     first_handle: WorkflowHandleAsync[int] = await debouncer.debounce_async(
-        debounce_period_sec, first_value
+        "key", debounce_period_sec, first_value
     )
     second_handle: WorkflowHandleAsync[int] = await debouncer.debounce_async(
-        debounce_period_sec, second_value
+        "key", debounce_period_sec, second_value
     )
     assert first_handle.workflow_id == second_handle.workflow_id
     assert await first_handle.get_result() == second_value
     assert await second_handle.get_result() == second_value
 
     third_handle: WorkflowHandleAsync[int] = await debouncer.debounce_async(
-        debounce_period_sec, third_value
+        "key", debounce_period_sec, third_value
     )
     fourth_handle: WorkflowHandleAsync[int] = await debouncer.debounce_async(
-        debounce_period_sec, fourth_value
+        "key", debounce_period_sec, fourth_value
     )
     assert third_handle.workflow_id != first_handle.workflow_id
     assert third_handle.workflow_id == fourth_handle.workflow_id
@@ -304,7 +312,7 @@ async def workflow_async(x: int) -> int:
     wfid = str(uuid.uuid4())
     options["workflow_id"] = wfid
     handle: WorkflowHandleAsync[int] = await debouncer.debounce_async(
-        debounce_period_sec, first_value
+        "key", debounce_period_sec, first_value
     )
     assert handle.workflow_id == wfid
     assert await handle.get_result() == first_value