nebius.aio.operation.Operation

class documentation

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

Constructor: Operation(source_method, channel, operation)

A convenience wrapper around operation protobufs.

The Operation wrapper normalizes nebius.api.nebius.common.v1.Operation and nebius.api.nebius.common.v1alpha1.Operation representations and provides helpers to:

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 bound to a nebius.aio.constant_channel.Constant that points at the provided source_method and reuses the provided channel for network/auth behaviors.

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())
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())
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	`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 timestamp for the operation or `None` if the operation hasn't finished yet.
Property	`id`	Return the operation identifier (string).
Property	`resource_id`	Return the resource id associated with the operation.
Method	`_set_new_operation`	Replace the wrapped operation object with a new instance.
Instance Variable	`_channel`	Undocumented
Instance Variable	`_get_request_obj`	Undocumented
Instance Variable	`_operation`	Undocumented
Instance Variable	`_service`	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 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.

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. A small safety margin is added to the provided timeout to allow for scheduling overhead.

Parameters
**kwargs:`Unpack[RequestKwargs]`	additional request keyword arguments see `nebius.aio.request_kwargs.RequestKwargs` for details.

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.

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`	polling interval (seconds or timedelta)
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.
Raises
`TimeoutError`	when the overall timeout is exceeded

@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 timestamp for the operation or None if the operation hasn't finished yet.

@property
id: str = (source) ¶

Return the operation identifier (string).

@property
resource_id: str = (source) ¶

Return the resource id associated with the operation.

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.

_channel = (source) ¶

Undocumented

_get_request_obj: type[GetOperationRequest | OldGet] = (source) ¶

Undocumented

_operation: Operation | Old = (source) ¶

Undocumented

_service: OperationServiceClient | OldClient = (source) ¶

Undocumented