nebius.aio.operation

module documentation
(source)

Helpers for working with long-running operations.

This module provides an Operation wrapper that normalizes different versions of the service operation protobuf and exposes convenient helpers for polling, synchronous waiting, and inspecting operation metadata.

The wrapper accepts operation protobufs from either the current v1 API or an older v1alpha1 variant and routes calls to the corresponding operation service client.

Class CurrentStep Wrapper describing a step of an operation progress tracker.
Class Operation A convenience wrapper around operation protobufs.
Class OperationProgressTracker Protocol describing operation-level progress tracking.
Function wrap_progress_tracker Return a progress tracker wrapper for an operation if available.
Type Variable OperationPb A convenience wrapper around operation protobufs. Either nebius.api.nebius.common.v1.Operation or nebius.api.nebius.common.v1alpha1.Operation, or their protobuf classes.
Type Variable T Undocumented
Class _ProgressTrackerWrapper Undocumented
Function _check_presence Undocumented
Function _get_timestamp Undocumented
Function _get_work_done Undocumented
def wrap_progress_tracker(operation: Operation[OperationPb] | None) -> OperationProgressTracker | None: (source)

Return a progress tracker wrapper for an operation if available.

This helper is exposed as Operation.progress_tracker and performs the presence checks needed to avoid accessing default/absent fields on protobuf wrappers.

Example

Using the helper directly:

tracker = wrap_progress_tracker(operation)
if tracker is not None:
    print(tracker.description())
OperationPb = (source)

A convenience wrapper around operation protobufs. Either nebius.api.nebius.common.v1.Operation or nebius.api.nebius.common.v1alpha1.Operation, or their protobuf classes.

Value
TypeVar('OperationPb')
T = (source)

Undocumented

Value
TypeVar('T')
def _check_presence(message: object, field: str) -> bool: (source)

Undocumented

def _get_timestamp(message: object, field: str) -> datetime | None: (source)

Undocumented

def _get_work_done(message: object) -> ProgressTracker.WorkDone | None: (source)

Undocumented