nebius.aio.operation.Operation

class documentation

class Operation(Generic[OperationPb]): (source)

Constructor: Operation(source_method, channel, operation)

Wrap an operation message.

The Operation wrapper normalizes nebius.api.nebius.common.v1.Operation and nebius.api.nebius.common.v1alpha1.Operation representations. Its methods:

inspect operation metadata (id, resource_id, timestamps),
poll/update the operation state via the corresponding operation service, and
wait for completion either asynchronously or synchronously.

The wrapper stores an operation-service client. A nebius.aio.constant_channel.Constant points this client at source_method. The client reuses channel for network and authorization functions.

Built-in channels schedule polling on the SDK loop, so this wrapper can be used from unrelated caller loops. A legacy custom channel without run_async keeps the historical local-awaitable fallback. Its update lock becomes bound to the first caller loop that contends for it, and the same wrapper must not then be used concurrently from another loop.

Example

Operation from a service action (e.g., creating a bucket):

from nebius.sdk import SDK
from nebius.aio.cli_config import Config
from nebius.api.nebius.storage.v1 import (
    BucketServiceClient,
    CreateBucketRequest
)

sdk = SDK(
    config_reader=Config(),
    user_agent_prefix="example-application/1.0",
)
service = BucketServiceClient(sdk)

# Create operation from service action
operation = await service.create(CreateBucketRequest(
    # fill-in necessary fields
))

# Wait for completion
await operation.wait()
print(f"New bucket ID: {operation.resource_id}")

Operation from list of operations:

from nebius.sdk import SDK
from nebius.aio.cli_config import Config
from nebius.api.nebius.storage.v1 import BucketServiceClient
from nebius.api.nebius.common.v1 import ListOperationsRequest

sdk = SDK(
    config_reader=Config(),
    user_agent_prefix="example-application/1.0",
)
service = BucketServiceClient(sdk)

# Get operation service client from the bucket service
operation_service = service.operation_service()
operations_response = await operation_service.list(ListOperationsRequest(
    # fill-in necessary fields
))

# Get first operation from list
if operations_response.operations:
    operation = operations_response.operations[0]

    # Manual update
    await operation.update()
    print(f"Operation status: {operation.status()}")

Parameters
source_method	the originating `service.method` name used to build a constant channel for operation management calls
channel	channel used for network and auth operations
operation	an operation protobuf instance (v1 or v1alpha1)

Method	`__init__`	Create an operation wrapper from the operation protobuf.
Method	`__repr__`	Return a compact string representation useful for debugging.
Method	`done`	Return True when the operation has reached a terminal state.
Method	`progress_tracker`	Return an operation progress tracker when available.
Method	`raw`	Return the underlying operation protobuf object.
Method	`status`	Return the operation's current status object or `None`.
Method	`successful`	Return True when the operation completed successfully.
Method	`sync_update`	Synchronously perform a single update of the operation state.
Method	`sync_wait`	Synchronously wait for the operation to complete.
Async Method	`update`	Fetch the latest operation data from the operation service.
Async Method	`wait`	Asynchronously wait until the operation reaches a terminal state.
Property	`created_at`	Return the operation creation timestamp.
Property	`created_by`	Return the identity that created the operation (string).
Property	`description`	Return the operation description as provided by the service.
Property	`finished_at`	Return the completion time or `None` if the operation is not finished.
Property	`id`	Return the operation identifier (string).
Property	`resource_id`	Return the resource id associated with the operation.
Method	`_check_process`	Reject an operation inherited across `fork` before locking.
Method	`_operation_snapshot`	Return the current operation message under the state lock.
Method	`_set_new_operation`	Replace the wrapped operation object with a new instance.
Async Method	`_update_from_submission`	Submit one update with a caller-captured monotonic start time.
Async Method	`_update_internal`	Fetch and store one operation update on the SDK event loop.
Async Method	`_wait_from_submission`	Submit polling with a caller-captured monotonic start time.
Async Method	`_wait_internal`	Poll the operation on the SDK event loop until it is complete.
Instance Variable	`_channel`	Undocumented
Instance Variable	`_get_request_obj`	Undocumented
Instance Variable	`_operation`	Undocumented
Instance Variable	`_process_id`	Undocumented
Instance Variable	`_service`	Undocumented
Instance Variable	`_state_lock`	Undocumented
Instance Variable	`_update_lock`	Undocumented

def __init__(self, source_method: str, channel: ClientChannelInterface, operation: OperationPb): (source) ¶

Create an operation wrapper from the operation protobuf.

def __repr__(self) -> str: (source) ¶

Return a compact string representation useful for debugging.

def done(self) -> bool: (source) ¶

Return True when the operation has reached a terminal state.

def progress_tracker(self) -> OperationProgressTracker | None: (source) ¶

Return an operation progress tracker when available.

Return None if the operation has no progress tracker. For example, v1alpha1 operations do not have one.

Example

Polling with a single-line progress display:

from asyncio import sleep
from datetime import datetime
from nebius.base.protos.well_known import local_timezone

while not operation.done():
    await operation.update()
    tracker = operation.progress_tracker()
    parts = [f"waiting for operation {operation.id} to complete:"]

    if tracker:
        work = tracker.work_fraction()
        if work is not None:
            parts.append(f"{work:.0%}")

        desc = tracker.description()
        if desc:
            parts.append(desc)

        started = tracker.started_at()
        if started is not None:
            elapsed = datetime.now(local_timezone) - started
            parts.append(f"{elapsed}")

        eta = tracker.estimated_finished_at()
        if eta is not None:
            parts.append(f"eta {eta}")

    print(" ".join(parts), end="\r", flush=True)
    await sleep(1)

print()

def raw(self) -> OperationPb: (source) ¶

Return the underlying operation protobuf object.

Use this to access version-specific fields that are not exposed by the normalized wrapper. The returned object preserves the existing mutable compatibility surface; mutating it concurrently bypasses this wrapper's snapshot and locking guarantees. Callers must serialize such mutation.

Returns
`OperationPb`	Current mutable operation protobuf object.

def status(self) -> RequestStatus | None: (source) ¶

Return the operation's current status object or None.

Returns
`RequestStatus` or nothing	Undocumented

def successful(self) -> bool: (source) ¶

Return True when the operation completed successfully.

def sync_update(self, **kwargs: Unpack[RequestKwargs]): (source) ¶

Synchronously perform a single update of the operation state.

This wraps the coroutine update and runs it via the channel's synchronous runner. An applicable authorization budget bounds the whole authorized flow; otherwise the request budget bounds SDK-loop queueing. Legacy channels whose provider is discoverable only on their owner loop enforce both clocks internally. A small safety margin accommodates scheduling overhead. Mutable metadata and authorization options are copied before the method dispatches work.

Parameters
**kwargs:`Unpack[RequestKwargs]`	additional request keyword arguments see `nebius.aio.request_kwargs.RequestKwargs` for details.
Raises
`ValueError`	If `timeout` or `auth_timeout` is NaN or infinite. Use `None` for an unlimited timeout.

Synchronously wait for the operation to complete.

This helper wraps wait and executes it in the channel's synchronous runner so callers that are not coroutine-based can wait for operation completion.

See wait for parameter details.

async def update(self, **kwargs: Unpack[RequestKwargs]): (source) ¶

Fetch the latest operation data from the operation service.

This coroutine performs a single get operation using the internal operation service client and replaces the wrapped operation object with the returned value.

Parameters
**kwargs:`Unpack[RequestKwargs]`	additional request keyword arguments see `nebius.aio.request_kwargs.RequestKwargs` for details.
Raises
`ValueError`	If `timeout` or `auth_timeout` is NaN or infinite. Use `None` for an unlimited timeout.

Asynchronously wait until the operation reaches a terminal state.

The method repeatedly invokes update at the specified interval until the operation is done or the overall timeout is reached. Certain transient errors (deadline exceeded) are treated as ignorable and will be retried.

Parameters
interval:`float` or `timedelta`	Positive, finite polling interval (seconds or timedelta). This value is ignored when the operation is already terminal.
timeout:optional `float`	overall timeout (seconds) for waiting, or `None` for infinite timeout, default infinite.
poll_iteration_timeout:optional `float` or `None`	timeout used for each polling iteration, will be passed as the `timeout` to each `update` call.
poll_per_retry_timeout:optional `float` or `None`, will be passed as the `per_retry_timeout` to each `update` call.	per-retry timeout for polling requests, will be passed as the `per_retry_timeout` to each `update` call.
poll_retries:`int \| None`	retry count used for polling requests, will be passed as the `retries` to each `update` call.
**kwargs:`Unpack[RequestKwargsForOperation]`	additional request keyword arguments see `nebius.aio.request_kwargs.RequestKwargsForOperation` for details. Mutable metadata and authorization options are copied before polling is submitted to the SDK event loop.
Raises
`TimeoutError`	when the overall timeout is exceeded
`ValueError`	When an unfinished operation receives a non-positive/non-finite polling interval or a non-finite overall timeout. Use `None` for an unlimited timeout.

@property
created_at: datetime = (source) ¶

Return the operation creation timestamp.

If the underlying protobuf does not expose a creation time this helper returns the current time in the local timezone. :rtype: datetime

@property
created_by: str = (source) ¶

Return the identity that created the operation (string).

@property
description: str = (source) ¶

Return the operation description as provided by the service.

@property
finished_at: datetime | None = (source) ¶

Return the completion time or None if the operation is not finished.

@property
id: str = (source) ¶

Return the operation identifier (string).

@property
resource_id: str = (source) ¶

Return the resource id associated with the operation.

def _check_process(self): (source) ¶

Reject an operation inherited across fork before locking.

def _operation_snapshot(self) -> OperationPb: (source) ¶

Return the current operation message under the state lock.

def _set_new_operation(self, operation: OperationPb): (source) ¶

Replace the wrapped operation object with a new instance.

The replacement is only allowed when the new operation has the same protobuf class as the currently wrapped object; otherwise an SDKError is raised.

async def _update_from_submission(self, submitted_at: float, **kwargs: Unpack[RequestKwargs]): (source) ¶

Submit one update with a caller-captured monotonic start time.

Parameters
submitted_at:`float`	Monotonic time when the caller submitted the update. Request and authorization deadlines include all later dispatch delay.
**kwargs:`Unpack[RequestKwargs]`	Additional request options for the operation service.

async def _update_internal(self, *, request_deadline: float | None = None, authorization_deadline: float | None = None, **kwargs: Unpack[RequestKwargs]): (source) ¶

Fetch and store one operation update on the SDK event loop.

Updates are serialized for this operation. A pending response therefore cannot arrive after a newer terminal response and regress the stored operation state. Once a terminal response is stored, later queued updates return without making another request.

Parameters
request_deadline:`float \| None`	Absolute monotonic request deadline captured before SDK-loop dispatch.
authorization_deadline:`float \| None`	Absolute monotonic authorization deadline captured before SDK-loop dispatch.
**kwargs:`Unpack[RequestKwargs]`	Request options for the operation service.

Submit polling with a caller-captured monotonic start time.

Parameters
submitted_at:`float`	Monotonic time when the caller submitted the wait. The overall timeout includes all later dispatch delay.
interval:`float \| timedelta`	Positive delay between polling attempts.
timeout:`float \| None`	Overall wait limit, or `None` for no limit.
poll_iteration_timeout:`float \| UnsetType \| None`	Limit for one polling request.
poll_per_retry_timeout:`float \| UnsetType \| None`	Limit for each retry.
poll_retries:`int \| None`	Retry count for each polling request.
**kwargs:`Unpack[RequestKwargsForOperation]`	Additional request options for the operation service.

Poll the operation on the SDK event loop until it is complete.

A local timeout and a service DEADLINE_EXCEEDED response are transient for one polling iteration. Other errors stop the wait.

Parameters
interval:`float \| timedelta`	Delay between polling attempts, in seconds or as a time delta.
timeout:`float \| None`	Overall wait limit in seconds. Use `None` for no limit.
deadline:`float \| None`	Absolute monotonic deadline captured before dispatch to the SDK loop. This includes runtime queueing and update-lock acquisition in the overall timeout.
poll_iteration_timeout:`float \| UnsetType \| None`	Timeout for one update request.
poll_per_retry_timeout:`float \| UnsetType \| None`	Timeout for each retry of an update request.
poll_retries:`int \| None`	Retry count for each update request.
**kwargs:`Unpack[RequestKwargsForOperation]`	Additional request options for the operation service.
Raises
`TimeoutError`	If the overall wait limit expires.

_channel = (source) ¶

Undocumented

_get_request_obj = (source) ¶

Undocumented

_operation = (source) ¶

Undocumented

_process_id = (source) ¶

Undocumented

_service = (source) ¶

Undocumented

_state_lock = (source) ¶

Undocumented

_update_lock = (source) ¶

Undocumented