diff --git a/changelog/added/boomeranging-bellhop.md b/changelog/added/boomeranging-bellhop.md new file mode 100644 index 0000000000..ecdaa4c79c --- /dev/null +++ b/changelog/added/boomeranging-bellhop.md @@ -0,0 +1 @@ +Nexus links now convert to and from `Link.Callback` instead of raising `NotImplementedError`. diff --git a/temporalio/nexus/_link_conversion.py b/temporalio/nexus/_link_conversion.py index 8457cdf4e0..672847ea1f 100644 --- a/temporalio/nexus/_link_conversion.py +++ b/temporalio/nexus/_link_conversion.py @@ -29,6 +29,7 @@ class _LinkType(str, Enum): WORKFLOW = temporalio.api.common.v1.Link.Workflow.DESCRIPTOR.full_name NEXUS_OPERATION = temporalio.api.common.v1.Link.NexusOperation.DESCRIPTOR.full_name ACTIVITY = temporalio.api.common.v1.Link.Activity.DESCRIPTOR.full_name + CALLBACK = temporalio.api.common.v1.Link.Callback.DESCRIPTOR.full_name @dataclass(frozen=True) @@ -66,12 +67,27 @@ def regex(self) -> re.Pattern[str]: _LinkType.ACTIVITY: _LinkPath("activities", "details"), } +_EXECUTION_TYPE_SEGMENTS: dict[temporalio.api.enums.v1.ExecutionType.ValueType, str] = { + temporalio.api.enums.v1.ExecutionType.EXECUTION_TYPE_WORKFLOW: "workflows", + temporalio.api.enums.v1.ExecutionType.EXECUTION_TYPE_ACTIVITY: "activities", + temporalio.api.enums.v1.ExecutionType.EXECUTION_TYPE_NEXUS_OPERATION: "nexus-operations", +} +_SEGMENT_EXECUTION_TYPES = {v: k for k, v in _EXECUTION_TYPE_SEGMENTS.items()} + +# A callback is addressed by its execution's type as well as its IDs, so its path does not fit +# _LinkPath. +_CALLBACK_PATH_REGEX = re.compile( + r"^/namespaces/(?P[^/]+)/(?P[^/]+)" + r"/(?P[^/]+)/(?P[^/]+)/callbacks/(?P[^/]+)$" +) + LINK_EVENT_ID_PARAM_NAME = "eventID" LINK_EVENT_TYPE_PARAM_NAME = "eventType" LINK_REQUEST_ID_PARAM_NAME = "requestID" LINK_REFERENCE_TYPE_PARAM_NAME = "referenceType" LINK_REASON_PARAM_NAME = "reason" +LINK_COMPONENT_PATH_PARAM_NAME = "componentPath" EVENT_REFERENCE_TYPE = "EventReference" REQUEST_ID_REFERENCE_TYPE = "RequestIdReference" @@ -124,13 +140,16 @@ def nexus_link_to_temporal_link( case _LinkType.ACTIVITY: return nexus_link_to_activity_link(nexus_link) + case _LinkType.CALLBACK: + return nexus_link_to_callback_link(nexus_link) + def temporal_link_to_nexus_link( temporal_link: temporalio.api.common.v1.Link, ) -> nexusrpc.Link | None: """Convert a Temporal API Link into a nexusrpc link. - Returns None when the Temporal link variant is missing. + Returns None when the Temporal link variant is missing or cannot be represented. """ match temporal_link.WhichOneof("variant"): case "workflow_event": @@ -149,8 +168,7 @@ def temporal_link_to_nexus_link( raise NotImplementedError("batch_job links are not supported") case "callback": - # TODO(Nexus team): Support callback links. - raise NotImplementedError("callback links are not supported") + return callback_link_to_nexus_link(temporal_link.callback) case None: logger.warning("Invalid Temporal link: missing variant") @@ -246,6 +264,37 @@ def activity_link_to_nexus_link( ) +def callback_link_to_nexus_link( + callback: temporalio.api.common.v1.Link.Callback, +) -> nexusrpc.Link | None: + """Convert a Callback link into a nexusrpc link. + + Returns None when the callback's execution type is unsupported. + """ + execution_type = _EXECUTION_TYPE_SEGMENTS.get(callback.execution.type) + if execution_type is None: + logger.warning( + f"Invalid Temporal callback link: unsupported execution type {callback.execution.type}" + ) + return None + segments = [ + callback.namespace, + execution_type, + callback.execution.business_id, + callback.execution.run_id, + "callbacks", + callback.request_id, + ] + path = "/namespaces/" + "/".join(urllib.parse.quote(s, safe="") for s in segments) + query_params = urllib.parse.urlencode( + [(LINK_COMPONENT_PATH_PARAM_NAME, p) for p in callback.component_path] + ) + return nexusrpc.Link( + url=_temporal_nexus_url(path, query_params=query_params), + type=_LinkType.CALLBACK.value, + ) + + def _build_link_url( link_type: _LinkType, namespace: str, @@ -281,17 +330,25 @@ class _ParsedLinkUrl: query_params: dict[str, list[str]] +def _parse_temporal_url(link: nexusrpc.Link) -> urllib.parse.ParseResult | None: + """Return the parsed link URL, or None if it does not carry the temporal scheme.""" + url = urllib.parse.urlparse(link.url) + if url.scheme != _URL_SCHEME: + logger.warning( + f"Invalid Nexus link: {link}. Expected scheme {_URL_SCHEME!r}, got {url.scheme!r}" + ) + return None + return url + + def _parse_link_url(link: nexusrpc.Link, link_type: _LinkType) -> _ParsedLinkUrl | None: """Return the parts of a link URL, or None if the URL is not a link of this type. The path is matched exactly, so the workflow and workflow-event shapes -- which differ only by the trailing /history -- cannot be mistaken for one another. """ - url = urllib.parse.urlparse(link.url) - if url.scheme != _URL_SCHEME: - logger.warning( - f"Invalid Nexus link: {link}. Expected scheme {_URL_SCHEME!r}, got {url.scheme!r}" - ) + url = _parse_temporal_url(link) + if url is None: return None regex = _LINK_PATHS[link_type].regex match = regex.match(url.path) @@ -421,6 +478,45 @@ def nexus_link_to_activity_link( ) +def nexus_link_to_callback_link( + nexus_link: nexusrpc.Link, +) -> temporalio.api.common.v1.Link | None: + """Convert a Nexus Callback link into a Temporal Callback link.""" + url = _parse_temporal_url(nexus_link) + if url is None: + return None + match = _CALLBACK_PATH_REGEX.match(url.path) + if not match: + logger.warning( + f"Invalid Nexus link: {nexus_link}. Expected path to match {_CALLBACK_PATH_REGEX.pattern}" + ) + return None + namespace, execution_type, business_id, run_id, request_id = ( + urllib.parse.unquote(g) for g in match.groups() + ) + execution_type_value = _SEGMENT_EXECUTION_TYPES.get(execution_type) + if execution_type_value is None: + logger.warning( + f"Invalid Nexus link: {nexus_link}. Unsupported execution type {execution_type!r}" + ) + return None + # parse_qs drops empty values by default, which would remove an empty component path segment + # and shift the ones after it. + query_params = urllib.parse.parse_qs(url.query, keep_blank_values=True) + return temporalio.api.common.v1.Link( + callback=temporalio.api.common.v1.Link.Callback( + namespace=namespace, + execution=temporalio.api.common.v1.Execution( + type=execution_type_value, + business_id=business_id, + run_id=run_id, + ), + component_path=query_params.get(LINK_COMPONENT_PATH_PARAM_NAME, []), + request_id=request_id, + ) + ) + + def _event_type_to_param( event_type: temporalio.api.enums.v1.EventType.ValueType, ) -> str: diff --git a/tests/nexus/test_link_conversion.py b/tests/nexus/test_link_conversion.py index 55973f94ec..cf8e1a1173 100644 --- a/tests/nexus/test_link_conversion.py +++ b/tests/nexus/test_link_conversion.py @@ -8,6 +8,28 @@ import temporalio.api.enums.v1 import temporalio.nexus._link_conversion +_CALLBACK_TYPE = temporalio.api.common.v1.Link.Callback.DESCRIPTOR.full_name +_WORKFLOW_EXECUTION = temporalio.api.enums.v1.ExecutionType.EXECUTION_TYPE_WORKFLOW + + +def _callback_link( + execution_type: temporalio.api.enums.v1.ExecutionType.ValueType, + business_id: str, + component_path: list[str] | None = None, +) -> temporalio.api.common.v1.Link: + return temporalio.api.common.v1.Link( + callback=temporalio.api.common.v1.Link.Callback( + namespace="ns", + execution=temporalio.api.common.v1.Execution( + type=execution_type, + business_id=business_id, + run_id="run-id", + ), + component_path=component_path, + request_id="request-id", + ) + ) + @pytest.mark.parametrize( ["query_param_str", "expected_event_ref"], @@ -485,3 +507,96 @@ def test_link_with_non_temporal_scheme_is_ignored(converter: str, url: str): """A link must carry the temporal scheme; the path alone does not make it one of ours.""" link = nexusrpc.Link(url=url, type="ignored") assert getattr(temporalio.nexus._link_conversion, converter)(link) is None + + +@pytest.mark.parametrize( + ["url", "expected_link"], + [ + ( + "temporal:///namespaces/ns/workflows/wf-id/run-id/callbacks/request-id", + _callback_link(_WORKFLOW_EXECUTION, "wf-id"), + ), + ( + "temporal:///namespaces/ns/activities/act-id/run-id/callbacks/request-id", + _callback_link( + temporalio.api.enums.v1.ExecutionType.EXECUTION_TYPE_ACTIVITY, "act-id" + ), + ), + ( + "temporal:///namespaces/ns/nexus-operations/op-id/run-id/callbacks/request-id", + _callback_link( + temporalio.api.enums.v1.ExecutionType.EXECUTION_TYPE_NEXUS_OPERATION, + "op-id", + ), + ), + ( + "temporal:///namespaces/ns/workflows/wf-id/run-id/callbacks/request-id" + "?componentPath=A%3AB&componentPath=C%2FD&componentPath=E_F", + _callback_link(_WORKFLOW_EXECUTION, "wf-id", ["A:B", "C/D", "E_F"]), + ), + # Segments end in a user-supplied ID, so a comma inside one is reachable. + ( + "temporal:///namespaces/ns/workflows/wf-id/run-id/callbacks/request-id" + "?componentPath=Updates&componentPath=id%2Cwith%2Ccommas", + _callback_link(_WORKFLOW_EXECUTION, "wf-id", ["Updates", "id,with,commas"]), + ), + # Only a path whose order differs from its sorted order proves order is preserved. + ( + "temporal:///namespaces/ns/workflows/wf-id/run-id/callbacks/request-id" + "?componentPath=zulu&componentPath=alpha&componentPath=", + _callback_link(_WORKFLOW_EXECUTION, "wf-id", ["zulu", "alpha", ""]), + ), + ( + "temporal:///namespaces/ns%2Fwith-slash/workflows/wf-id%2Fcallbacks%2Ffake" + "/run-id/callbacks/request%20id%3Fx%3D1", + temporalio.api.common.v1.Link( + callback=temporalio.api.common.v1.Link.Callback( + namespace="ns/with-slash", + execution=temporalio.api.common.v1.Execution( + type=_WORKFLOW_EXECUTION, + business_id="wf-id/callbacks/fake", + run_id="run-id", + ), + request_id="request id?x=1", + ) + ), + ), + ], +) +def test_link_conversion_callback_to_link_and_back( + url: str, expected_link: temporalio.api.common.v1.Link +): + link = nexusrpc.Link(url=url, type=_CALLBACK_TYPE) + assert ( + temporalio.nexus._link_conversion.temporal_link_to_nexus_link(expected_link) + == link + ) + assert ( + temporalio.nexus._link_conversion.nexus_link_to_temporal_link(link) + == expected_link + ) + + +def test_callback_link_with_unsupported_execution_type_is_ignored(): + assert ( + temporalio.nexus._link_conversion.temporal_link_to_nexus_link( + _callback_link( + temporalio.api.enums.v1.ExecutionType.EXECUTION_TYPE_UNSPECIFIED, + "wf-id", + ) + ) + is None + ) + + +@pytest.mark.parametrize( + "url", + [ + "https:///namespaces/ns/workflows/wf-id/run-id/callbacks/request-id", + "temporal:///namespaces/ns/workflows/wf-id/run-id/callbacks", + "temporal:///namespaces/ns/schedules/sched-id/run-id/callbacks/request-id", + ], +) +def test_invalid_callback_nexus_link_is_ignored(url: str): + link = nexusrpc.Link(url=url, type=_CALLBACK_TYPE) + assert temporalio.nexus._link_conversion.nexus_link_to_temporal_link(link) is None