viam.operations

Module Contents

Classes

Operation

An Operation represents a running operation.

Functions

opid_from_metadata(→ Optional[uuid.UUID])
run_with_operation(→ Callable[P, Coroutine[Any, Any, T]])	Run a component function with an Operation.

Attributes

P
T
METADATA_KEY

class viam.operations.Operation(method: str, cancel_event: asyncio.Event, opid: uuid.UUID | None = None)

An Operation represents a running operation.

Every request made to a robot’s components will create a new Operation on the server. For custom components built with this python-sdk, you should check whether the operation has been cancelled to prevent long-running tasks from leaking.

ARG_NAME = 'viam_operation'

id: uuid.UUID

method: str

time_started: float

async is_cancelled() → bool

__str__() → str

Return str(self).

viam.operations.P

viam.operations.T

viam.operations.METADATA_KEY = 'opid'

viam.operations.opid_from_metadata(metadata: Mapping[str, str] | None) → uuid.UUID | None

viam.operations.run_with_operation(func: Callable[P, Coroutine[Any, Any, T]]) → Callable[P, Coroutine[Any, Any, T]]

Run a component function with an Operation. Running a function with an Operation will allow the function to know if/when the calling task was cancelled and take appropriate action (.e.g. stop long running tasks and exit early).

If a timeout is provided to the function, the operation will cancel when the timeout is reached.

An example use case is if a gRPC client disconnects after making a request. Rather than continue to run a task for a receiver that is no longer there, the component can cancel and clean up, saving resources.

Parameters:

func (Callable[..., Coroutine[Any, Any, T]]) – The function to be called with an Operation. This function MUST accept **kwargs or a parameter whose name is equal the value of Operation.ARG_NAME

Returns:

The return of the function

Return type:

T