DEV

Author: dandavison

temporalio/client.py

@@ -532,7 +532,7 @@ async def start_workflow(
 
         for l in workflow_event_links:
             print(
-                f"🌈<< worker starting workflow with link: {[google.protobuf.json_format.MessageToJson(l)]}"
+                f"🌈@@ worker starting workflow with link: {google.protobuf.json_format.MessageToJson(l)}"
             )
 
         return await self._impl.start_workflow(

temporalio/nexus/handler.py

@@ -21,6 +21,7 @@
 )
 
 import nexusrpc.handler
+from hyperlinked import hyperlinked, print
 from nexusrpc.handler import _ServiceImpl
 from typing_extensions import Concatenate, overload
 
@@ -152,7 +153,6 @@ async def start_workflow(
     if task_queue is None:
         # TODO(dan): are we handling empty string well elsewhere?
         task_queue = get_task_queue()
-    print(f"🌈 starting workflow {workflow} {id} in task queue {task_queue}")
     completion_callbacks = (
         [
             temporalio.common.NexusCompletionCallback(
@@ -162,6 +162,9 @@ async def start_workflow(
         if options.callback_url
         else []
     )
+    print(f"🌈 starting workflow {workflow} {id} in task queue {task_queue}")
+    for link in options.links:
+        print(f"🌈 link: {link}")
     # We need to pass options (completion_callbacks, links, on_conflict_options) which are
     # deliberately not exposed in any overload, hence the type error.
     workflow_handle = await client.start_workflow(  # type: ignore
@@ -351,25 +354,31 @@ def _nexus_link_to_workflow_event(
     )
     url = urllib.parse.urlparse(link.url)
     if not (match := path_regex.match(url.path)):
+        from hyperlinked import print_stack
+
         logger.warning(
-            f"Invalid Nexus link: {link}. Expected path to match {path_regex.pattern}"
+            hyperlinked(
+                f"@@ Invalid Nexus link: {link}. Expected path to match {path_regex.pattern}"
+            )
         )
+        print_stack()
+        exit(1)
         return None
 
     try:
         query_params = urllib.parse.parse_qs(url.query)
         [reference_type] = query_params.get("referenceType", [])
         if reference_type != "EventReference":
             raise ValueError(
-                f"Expected Nexus link URL query parameter referenceType to be EventReference but got: {reference_type}"
+                f"@@ Expected Nexus link URL query parameter referenceType to be EventReference but got: {reference_type}"
             )
         [event_type_name] = query_params.get("eventType", [])
         event_ref = temporalio.api.common.v1.Link.WorkflowEvent.EventReference(
             event_type=temporalio.api.enums.v1.EventType.Value(event_type_name)
         )
     except ValueError as err:
         logger.warning(
-            f"Failed to parse event type from Nexus link URL query parameters: {link} ({err})"
+            f"@@ Failed to parse event type from Nexus link URL query parameters: {link} ({err})"
         )
         event_ref = None
 

temporalio/worker/_nexus.py

@@ -192,7 +192,7 @@ async def _handle_start_operation(
         )
         for l in request.links:
             print(
-                f"🌈<< worker received task with link: {google.protobuf.json_format.MessageToJson(l)}"
+                f"🌈@@ worker received task with link: {google.protobuf.json_format.MessageToJson(l)}"
             )
         temporalio.nexus.handler._current_context.set(
             temporalio.nexus.handler._Context(