viam.operations

Attributes

P

T

METADATA_KEY

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.

Module Contents

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

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[source]
__str__() str[source]
viam.operations.P
viam.operations.T
viam.operations.METADATA_KEY = 'opid'
viam.operations.opid_from_metadata(metadata: Mapping[str, str] | None) uuid.UUID | None[source]
viam.operations.run_with_operation(func: Callable[P, Coroutine[Any, Any, T]]) Callable[P, Coroutine[Any, Any, T]][source]

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 (for example 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