temporalio.contrib.opentelemetry._otel_interceptor._TracingWorkflowInboundInterceptor

class documentation

class _TracingWorkflowInboundInterceptor(temporalio.worker.WorkflowInboundInterceptor): (source)

Constructor: _TracingWorkflowInboundInterceptor(next)

View In Hierarchy

Tracing interceptor for workflow calls.

def __init__(self, next: temporalio.worker.WorkflowInboundInterceptor): (source)
overrides temporalio.worker.WorkflowInboundInterceptor.__init__

Initialize a tracing workflow interceptor.

async def execute_workflow(self, input: temporalio.worker.ExecuteWorkflowInput) -> Any: (source)
overrides temporalio.worker.WorkflowInboundInterceptor.execute_workflow
async def handle_query(self, input: temporalio.worker.HandleQueryInput) -> Any: (source)
overrides temporalio.worker.WorkflowInboundInterceptor.handle_query
async def handle_signal(self, input: temporalio.worker.HandleSignalInput): (source)
overrides temporalio.worker.WorkflowInboundInterceptor.handle_signal
async def handle_update_handler(self, input: temporalio.worker.HandleUpdateInput) -> Any: (source)
overrides temporalio.worker.WorkflowInboundInterceptor.handle_update_handler
def handle_update_validator(self, input: temporalio.worker.HandleUpdateInput): (source)
overrides temporalio.worker.WorkflowInboundInterceptor.handle_update_validator
def init(self, outbound: temporalio.worker.WorkflowOutboundInterceptor): (source)
overrides temporalio.worker.WorkflowInboundInterceptor.init
def _top_level_workflow_context(self, input: _InputWithHeaders) -> Iterator[None]: (source)

Undocumented

def _workflow_maybe_span(self, name: str) -> Iterator[None]: (source)

Undocumented

_add_temporal_spans: bool = (source)

Undocumented