temporalio.nexus._link_conversion

module documentation
(source)

Undocumented

Function activity_link_to_nexus_link Convert an Activity link into a nexusrpc link.
Function nexus_link_to_activity_link Convert a Nexus Activity link into a Temporal Activity link.
Function nexus_link_to_nexus_operation_link Convert a nexus link into a Temporal NexusOperation link
Function nexus_link_to_temporal_link Convert a nexusrpc link into a Temporal API Link.
Function nexus_link_to_workflow_event_link Convert a nexus link into a Temporal WorkflowEvent link
Function nexus_link_to_workflow_link Convert a nexus link into a Temporal Workflow link.
Function nexus_operation_to_nexus_link Convert a NexusOperation link into a nexusrpc link
Function temporal_link_to_nexus_link Convert a Temporal API Link into a nexusrpc link.
Function workflow_event_to_nexus_link Convert a WorkflowEvent link into a nexusrpc link
Function workflow_execution_started_event_link_from_workflow_handle Create a WorkflowEvent link corresponding to a started workflow
Function workflow_to_nexus_link Convert a Workflow link into a nexusrpc link.
Constant EVENT_REFERENCE_TYPE Undocumented
Constant LINK_EVENT_ID_PARAM_NAME Undocumented
Constant LINK_EVENT_TYPE_PARAM_NAME Undocumented
Constant LINK_REASON_PARAM_NAME Undocumented
Constant LINK_REFERENCE_TYPE_PARAM_NAME Undocumented
Constant LINK_REQUEST_ID_PARAM_NAME Undocumented
Constant REQUEST_ID_REFERENCE_TYPE Undocumented
Variable logger Undocumented
Class _LinkPath The URL path shape for one link type.
Class _LinkType Undocumented
Class _ParsedLinkUrl Undocumented
Function _build_link_url Build the URL for a link of the given type.
Function _event_reference_to_query_params Undocumented
Function _event_type_constant_case_to_pascal_case Convert a CONSTANT_CASE string to PascalCase.
Function _event_type_pascal_case_to_constant_case Convert a PascalCase string to CONSTANT_CASE.
Function _event_type_to_param Render an event type as the short PascalCase name used on the wire.
Function _optional_single_query_param Undocumented
Function _parse_link_url Return the parts of a link URL, or None if the URL is not a link of this type.
Function _query_params_to_event_reference Return an EventReference from the query params or raise ValueError.
Function _query_params_to_event_type Return the event type named in the query params, or raise ValueError.
Function _query_params_to_request_id_reference Return an EventReference from the query params or raise ValueError.
Function _request_id_reference_to_query_params Undocumented
Function _temporal_nexus_url Undocumented
Constant _LINK_PATHS Undocumented
Constant _URL_SCHEME Undocumented
def activity_link_to_nexus_link(activity: temporalio.api.common.v1.Link.Activity) -> nexusrpc.Link: (source)

Convert an Activity link into a nexusrpc link.

def nexus_link_to_activity_link(nexus_link: nexusrpc.Link) -> temporalio.api.common.v1.Link | None: (source)

Convert a Nexus Activity link into a Temporal Activity link.

def nexus_link_to_nexus_operation_link(nexus_link: nexusrpc.Link) -> temporalio.api.common.v1.Link | None: (source)

Convert a nexus link into a Temporal NexusOperation link

This is used when propagating links from a Nexus start operation request to a StartNexusOperation request.

def nexus_link_to_temporal_link(nexus_link: nexusrpc.Link) -> temporalio.api.common.v1.Link | None: (source)

Convert a nexusrpc link into a Temporal API Link.

Returns None when the Nexus link type is invalid or unknown.

def nexus_link_to_workflow_event_link(link: nexusrpc.Link) -> temporalio.api.common.v1.Link | None: (source)

Convert a nexus link into a Temporal WorkflowEvent link

This is used when propagating links from a Nexus start operation request to a StartWorklow request.

def nexus_link_to_workflow_link(link: nexusrpc.Link) -> temporalio.api.common.v1.Link | None: (source)

Convert a nexus link into a Temporal Workflow link.

def nexus_operation_to_nexus_link(op_link: temporalio.api.common.v1.Link.NexusOperation) -> nexusrpc.Link: (source)

Convert a NexusOperation link into a nexusrpc link

Used when propagating links from a StartNexusOperation response to a Nexus start operation response.

def temporal_link_to_nexus_link(temporal_link: temporalio.api.common.v1.Link) -> nexusrpc.Link | None: (source)

Convert a Temporal API Link into a nexusrpc link.

Returns None when the Temporal link variant is missing.

def workflow_event_to_nexus_link(workflow_event: temporalio.api.common.v1.Link.WorkflowEvent) -> nexusrpc.Link: (source)

Convert a WorkflowEvent link into a nexusrpc link

Used when propagating links from a StartWorkflow response to a Nexus start operation response.

def workflow_execution_started_event_link_from_workflow_handle(handle: temporalio.client.WorkflowHandle[Any, Any], request_id: str) -> temporalio.api.common.v1.Link.WorkflowEvent: (source)

Create a WorkflowEvent link corresponding to a started workflow

def workflow_to_nexus_link(workflow: temporalio.api.common.v1.Link.Workflow) -> nexusrpc.Link: (source)

Convert a Workflow link into a nexusrpc link.

EVENT_REFERENCE_TYPE: str = (source)

Undocumented

Value
'EventReference'
LINK_EVENT_ID_PARAM_NAME: str = (source)

Undocumented

Value
'eventID'
LINK_EVENT_TYPE_PARAM_NAME: str = (source)

Undocumented

Value
'eventType'
LINK_REASON_PARAM_NAME: str = (source)

Undocumented

Value
'reason'
LINK_REFERENCE_TYPE_PARAM_NAME: str = (source)

Undocumented

Value
'referenceType'
LINK_REQUEST_ID_PARAM_NAME: str = (source)

Undocumented

Value
'requestID'
REQUEST_ID_REFERENCE_TYPE: str = (source)

Undocumented

Value
'RequestIdReference'
logger = (source)

Undocumented

def _build_link_url(link_type: _LinkType, namespace: str, id_value: str, run_id: str, *, query_params: str | None = '') -> str: (source)

Build the URL for a link of the given type.

def _event_reference_to_query_params(event_ref: temporalio.api.common.v1.Link.WorkflowEvent.EventReference) -> str: (source)

Undocumented

def _event_type_constant_case_to_pascal_case(s: str) -> str: (source)

Convert a CONSTANT_CASE string to PascalCase.

>>> _event_type_constant_case_to_pascal_case("NEXUS_OPERATION_SCHEDULED")
"NexusOperationScheduled"
def _event_type_pascal_case_to_constant_case(s: str) -> str: (source)

Convert a PascalCase string to CONSTANT_CASE.

>>> _event_type_pascal_case_to_constant_case("NexusOperationScheduled")
"NEXUS_OPERATION_SCHEDULED"
def _event_type_to_param(event_type: temporalio.api.enums.v1.EventType.ValueType) -> str: (source)

Render an event type as the short PascalCase name used on the wire.

def _optional_single_query_param(query_params: dict[str, list[str]], param_name: str) -> str: (source)

Undocumented

def _parse_link_url(link: nexusrpc.Link, link_type: _LinkType) -> _ParsedLinkUrl | None: (source)

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.

def _query_params_to_event_reference(query_params: dict[str, list[str]]) -> temporalio.api.common.v1.Link.WorkflowEvent.EventReference: (source)

Return an EventReference from the query params or raise ValueError.

def _query_params_to_event_type(query_params: dict[str, list[str]]) -> temporalio.api.enums.v1.EventType.ValueType: (source)

Return the event type named in the query params, or raise ValueError.

Both the prefixed proto name and the short PascalCase name are accepted, since either may arrive on the wire.

def _query_params_to_request_id_reference(query_params: dict[str, list[str]]) -> temporalio.api.common.v1.Link.WorkflowEvent.RequestIdReference: (source)

Return an EventReference from the query params or raise ValueError.

def _request_id_reference_to_query_params(request_id_ref: temporalio.api.common.v1.Link.WorkflowEvent.RequestIdReference) -> str: (source)

Undocumented

def _temporal_nexus_url(path: str, *, query_params: str | None = '') -> str: (source)

Undocumented

_LINK_PATHS: dict[_LinkType, _LinkPath] = (source)

Undocumented

Value
{_LinkType.WORKFLOW_EVENT: _LinkPath('workflows', 'history'),
 _LinkType.WORKFLOW: _LinkPath('workflows', None),
 _LinkType.NEXUS_OPERATION: _LinkPath('nexus-operations', 'details',
                                      run_id_required=False),
 _LinkType.ACTIVITY: _LinkPath('activities', 'details')}
_URL_SCHEME: str = (source)

Undocumented

Value
'temporal'