temporalio.bridge._visitor_functions.BoundedVisitorFunctions

class documentation

class BoundedVisitorFunctions(VisitorFunctions): (source)

Constructor: BoundedVisitorFunctions(inner, concurrency_limit)

View In Hierarchy

Wraps VisitorFunctions to cap concurrent payload visits via a semaphore.

After the full traversal, call drain() to await all in-flight tasks.

Method __init__ Create a bounded wrapper around the given visitor functions.
Method checkpoint Return a marker for tasks scheduled after this point.
Async Method drain Wait for all in-flight background tasks to complete.
Async Method drain_since Wait for tasks scheduled after checkpoint to finish.
Async Method visit_payload Visit a single payload once capacity is available.
Async Method visit_payloads Visit a sequence of payloads once capacity is available.
Async Method visit_system_nexus_envelope Visit a system Nexus envelope payload once capacity is available.
Async Method _drain_tasks Wait for the given tasks, cancelling all tasks if one fails.
Instance Variable _inner Undocumented
Instance Variable _sem Undocumented
Instance Variable _tasks Undocumented
def __init__(self, inner: VisitorFunctions, concurrency_limit: int): (source)

Create a bounded wrapper around the given visitor functions.

def checkpoint(self) -> int: (source)
overrides temporalio.bridge._visitor_functions.VisitorFunctions.checkpoint

Return a marker for tasks scheduled after this point.

async def drain(self): (source)

Wait for all in-flight background tasks to complete.

On cancellation or error, cancels all remaining tasks and awaits them so their finally blocks run before this coroutine returns.

async def drain_since(self, checkpoint: int): (source)
overrides temporalio.bridge._visitor_functions.VisitorFunctions.drain_since

Wait for tasks scheduled after checkpoint to finish.

This lets system-envelope traversal finish mutating its decoded value before that value is serialized again, without waiting for unrelated visits that were already in progress.

async def visit_payload(self, payload: Payload): (source)
overrides temporalio.bridge._visitor_functions.VisitorFunctions.visit_payload

Visit a single payload once capacity is available.

async def visit_payloads(self, payloads: PayloadSequence): (source)
overrides temporalio.bridge._visitor_functions.VisitorFunctions.visit_payloads

Visit a sequence of payloads once capacity is available.

async def visit_system_nexus_envelope(self, payload: Payload): (source)
overrides temporalio.bridge._visitor_functions.VisitorFunctions.visit_system_nexus_envelope

Visit a system Nexus envelope payload once capacity is available.

async def _drain_tasks(self, tasks: list[asyncio.Task[None]]): (source)

Wait for the given tasks, cancelling all tasks if one fails.

_inner = (source)

Undocumented

_sem = (source)

Undocumented

_tasks: list[asyncio.Task[None]] = (source)

Undocumented