nebius.aio.operation

module documentation
(source)

Helpers for working with long-running operations.

The Operation wrapper normalizes different service-operation versions. Its methods poll, wait synchronously, and inspect operation metadata.

The wrapper accepts current v1 and older v1alpha1 operation messages. It routes calls to the applicable operation-service client.

Class CurrentStep Describe one step in an operation progress tracker.
Class Operation Wrap an operation message.
Class OperationMessage Structural fields shared by current and alpha direct operations.
Class OperationProgressTracker Define 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 No class docstring; 0/1 instance variable, 0/11 method, 1/1 static method documented
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',
        bound=OperationMessage)
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