nebius

Name	nebius JSON
Version	0.2.53 JSON
	download
home_page	None
Summary	Nebius Python SDK
upload_time	2025-08-19 12:33:23
maintainer	None
docs_url	None
author	None
requires_python	>=3.10
license	The MIT License (MIT) Copyright (c) 2025 Nebius B.V. Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
keywords
VCS
bugtrack_url
requirements	No requirements were recorded.
Travis-CI	No Travis.
coveralls test coverage	No coveralls.

            # Nebius Python® SDK

The **Nebius Python® SDK** is a comprehensive client library for interacting with [nebius.com](https://nebius.com) services.
Built on gRPC, it supports all APIs defined in the [Nebius API repository](https://github.com/nebius/api).
This SDK simplifies resource management, authentication, and communication with Nebius services, making it a valuable tool for developers.

> **Note**: "Python" and the Python logos are trademarks or registered trademarks of the Python Software Foundation, used by Nebius B.V. with permission from the Foundation.

### Full documentation and reference

To see all the services and their methods, look into the [API reference](https://nebius.github.io/pysdk/apiReference.html).

### Installation

```bash
pip install nebius
```

If you've received this module in a zip archive or checked out from git, install it as follows:

```bash
pip install ./path/to/your/pysdk
```

### Example

Working examples in `src/examples`.
Try it out as follows:
```bash
NEBIUS_IAM_TOKEN=$(nebius iam get-access-token) python -m ./path/to/your/pysdk/src/examples/basic.py your-project-id
```

### How-to

#### Initialize

```python
from nebius.sdk import SDK

sdk = SDK()
```

This will initialize the [SDK](https://nebius.github.io/pysdk/nebius.sdk.SDK.html) with an IAM token from a `NEBIUS_IAM_TOKEN` env var.
If you want to use different ways of authorization, read further.

See the following how-to's on how to provide your crerentials:

##### Initialize using an IAM Token

You can also initialize the `SDK` by providing the token directly or from the other environment variable, here are examples how to do that:

```python
import os
from nebius.sdk import SDK
from nebius.aio.token.static import Bearer, EnvBearer  # [1]
from nebius.aio.token.token import Token  # [2]

sdk = SDK(credentials=os.environ.get("NEBIUS_IAM_TOKEN", ""))
#or
sdk = SDK(credentials=Bearer(os.environ.get("NEBIUS_IAM_TOKEN", "")))
#or
sdk = SDK(credentials=EnvBearer("NEBIUS_IAM_TOKEN"))
#or
sdk = SDK(credentials=Bearer(Token(os.environ.get("NEBIUS_IAM_TOKEN", ""))))
```
[[1](https://nebius.github.io/pysdk/nebius.aio.token.static.html), [2](https://nebius.github.io/pysdk/nebius.aio.token.token.html)]

Now, your application will get token from the local Env variable, as in the example above, but provided in several other ways.

##### Initialize using CLI config

If you've set up [Nebius AI Cloud CLI](https://docs.nebius.com/cli), you can initialize the `SDK` using [CLI config](https://nebius.github.io/pysdk/nebius.aio.cli_config.Config.html):

```python
from nebius.sdk import SDK
from nebius.aio.cli_config import Config

sdk = SDK(config_reader=Config())
```

This will also import the domain if the endpoint parameter is in the config and the domain wasn't set explicitly.

*Keep in mind, that it will get the token from the `NEBIUS_IAM_TOKEN` environment variable if it is set, or use `NEBIUS_PROFILE` for selecting the profile, the same way CLI does. To stop that from happening, set `Config(no_env=True)`*

Config reader also helps with getting the default parent ID if necessary:

```python
from nebius.aio.cli_config import Config

print(f"My default parent ID: {Config().parent_id}")
```

Check the [`Config` documentation](https://nebius.github.io/pysdk/nebius.aio.cli_config.Config.html) for more settings like file path, profile name or environment variables names.

##### Initialize with the private key file

If you have a private key and a service account, you may want to authorize using them. Here is an example of how to do it.

Replace in the IDs in the following example with your service account and public key ID pair, related to the private key you have.
You need to have a `private_key.pem` file on your machine, modify the file path in the example accordingly.

```python
from nebius.sdk import SDK
from nebius.base.service_account.pk_file import Reader as PKReader  # [1]

sdk = SDK(
    credentials=PKReader(
        filename="location/of/your/private_key.pem",
        public_key_id="public-key-id",
        service_account_id="your-service-account-id",
    ),
)
#or without importing PKReader:
sdk = SDK(
    service_account_private_key_file_name="location/of/your/private_key.pem",
    service_account_public_key_id="public-key-id",
    service_account_id="your-service-account-id",
)
```
[[1](https://nebius.github.io/pysdk/nebius.base.service_account.pk_file.Reader.html)]

##### Initialize with a credentials file

Assuming you have a joint credentials file with a private key and all the IDs inside.

```python
from nebius.sdk import SDK
from nebius.base.service_account.credentials_file import Reader as CredentialsReader  # [1]

sdk = SDK(
    credentials=CredentialsReader(
        filename="location/of/your/credentials.json",
    ),
)
#or without importing CredentialsReader:
sdk = SDK(
    credentials_file_name="location/of/your/credentials.json",
)
```
[[1](https://nebius.github.io/pysdk/nebius.base.service_account.credentials_file.Reader.html)]

#### Test the SDK

Now as you've initialized the SDK, you may want to test whether your credentials are ok, everything works and you have a good connection.

To test the SDK, we provide a convenient method [`SDK.whoami`](https://nebius.github.io/pysdk/nebius.sdk.SDK.html#whoami), that will return the basic information about the profile, you've authenticated with:

```python
import asyncio

async def my_call():
    async with sdk:
        print(await sdk.whoami())

asyncio.run(my_call)
```

It is important to close the SDK, so all the coroutines and tasks will be gracefully stopped and gathered. It can either be achieved by using `async with`, or by explicitly calling `sdk.close()`:

```python
import asyncio

async def my_call():
    try:
        print(await sdk.whoami())
        # Other calls to SDK
    finally:
        await sdk.close()

asyncio.run(my_call)
```

SDK is created with asyncio in mind, so the best way to call methods of it is to use an async context. But if you haven't started async loop, you can run it synchronously:

```python
try:
    print(sdk.whoami().wait())
    # Other calls to SDK
finally:
    sdk.sync_close()
```

*Keep in mind, that this may lead to some problems or infinite locks, even if timeouts have been added. Moreover, synchronous methods won't run inside an async call stack, if you haven't provided a dedicated separate loop for the SDK. And even when the loop is provided, there might be issues or deadlocks.*

Closing the SDK is not strictly necessary, but forgetting to add it may lead to a bunch of annoying errors of unterminated tasks.

#### Test token renewal

If you use service account, credentials file or something like that, SDK will renew your tokens under the hood. This renewal process will normally report to log if any errors occur. However it might be good to get the repored errors as request errors. In that case, pass special options to the request like so:


```python
import asyncio

from nebius.aio.authorization.options import options_to_metadata
from nebius.aio.token.renewable import (
    OPTION_RENEW_REQUEST_TIMEOUT,
    OPTION_RENEW_REQUIRED,
    OPTION_RENEW_SYNCHRONOUS,
)

async def my_call():
    try:
        await sdk.whoami(
            metadata=options_to_metadata(
                {
                    OPTION_RENEW_REQUIRED: "true",
                    OPTION_RENEW_SYNCHRONOUS: "true",
                    OPTION_RENEW_REQUEST_TIMEOUT: ".9",
                }
            )
        )
    except Exception as err:
        print(f"something is wrong with your token: {err=}")
    finally:
        await sdk.close()

asyncio.run(my_call)
```

You can pass these options to any request, or add them to your metadata with `add_options_to_metadata`.

#### Call some method

Now as you have your SDK initialized and tested, you may work with our services and call their methods with it. Here and further we assume, that the [SDK](https://nebius.github.io/pysdk/nebius.sdk.SDK.html) is initialized and is located in the `sdk` variable. We also omit closing the SDK.

All the services API classes are located in submodules of `nebius.api.nebius`. [The reference can be found here](https://nebius.github.io/pysdk/apiReference.html). The `nebius.api.nebius` also includes all the raw gRPC and ProtoBuf classes.

As an example how to use the API, let's receive a bucket from a storage service by its ID:

```python
import asyncio

from nebius.api.nebius.storage.v1 import GetBucketRequest
from nebius.api.nebius.storage.v1 import BucketServiceClient

async def my_call():
    service = BucketServiceClient(sdk)
    return await service.get(GetBucketRequest(
        id="some-bucket-id",
    ))

asyncio.run(my_call())
```

Same thing, but synchronously:

```python
import asyncio

from nebius.api.nebius.storage.v1 import BucketServiceClient, GetBucketRequest

service = BucketServiceClient(sdk)
result = service.get(GetBucketRequest(
    id="some-bucket-id",
)).wait()
```

##### Parent ID

Some methods may include `parent_id` in the requests, for certain methods this field is populated automatically:
* Methods `list` and `get_by_name` with an empty `parent_id`
* All other methods, except `update`, with an empty `metadata.parent_id`

The `parent_id` will only be set if it was preset at the initialization, either from the [CLI `Config`](https://nebius.github.io/pysdk/nebius.aio.cli_config.Config.html) or from the `parent_id` attribute from the [SDK](https://nebius.github.io/pysdk/nebius.sdk.SDK.html). You can disable it, retaining the CLI Config, if you set it up with `no_parent_id=True`.

##### Operations

Many core methods return a [`nebius.aio.Operation`](https://nebius.github.io/pysdk/nebius.aio.operation.Operation.html) object, representing a time-consuming asynchronous operation. For example, the `create` request from the `BucketServiceClient` above is one of such cases. The [`nebius.aio.Operation`](https://nebius.github.io/pysdk/nebius.aio.operation.Operation.html) is a wrapper class that provides convenient methods for working with operations. It can be awaited util completion.

Here is an async example:

```python
from nebius.api.nebius.storage.v1 import BucketServiceClient, CreateBucketRequest

service = BucketServiceClient(sdk)
operation = await service.create(CreateBucketRequest(
    # fill-in necessary fields
))
await operation.wait()
print(f"New bucket ID: {operation.resource_id}")
```

Or synchronously:

```python
from nebius.api.nebius.storage.v1 import BucketServiceClient, CreateBucketRequest

service = BucketServiceClient(sdk)
operation = service.create(CreateBucketRequest(
    # fill-in necessary fields
)).wait()
operation.wait_sync()
print(f"New bucket ID: {operation.resource_id}")
```

##### Operations service

If you need to get an operation or list operations, you will need an [`OperationServiceClient`](https://nebius.github.io/pysdk/nebius.api.nebius.common.v1.OperationServiceClient.html).

The [`OperationServiceClient`](https://nebius.github.io/pysdk/nebius.api.nebius.common.v1.OperationServiceClient.html), despite being located in [`nebius.api.nebius.common.v1`](https://nebius.github.io/pysdk/nebius.api.nebius.common.v1.html), **cannot** be used the same way as other services. The real operation service must be acquired from the source service of your operation using [`service.operation_service()`](https://nebius.github.io/pysdk/nebius.aio.client.ClientWithOperations.html#operation_service) method.

Example of listing operations and getting operation by service (only async for brevity):
```python
from nebius.api.nebius.common.v1 import GetOperationRequest, ListOperationsRequest
from nebius.api.nebius.storage.v1 import BucketServiceClient

service = BucketServiceClient(sdk)
op_service = service.operation_service()

resp = await op_service.list(ListOperationsRequest(resource_id="your-bucket-id"))
op_id = resp.operations[0].id # The elements of resp.operations are not of type Operation!
real_operation = await op_service.get(GetOperationRequest(id=op_id))

# Get returns the real operation that can be awaited.
await real_operation.wait()
```

**NOTE** As you can see from the example, only [`get`](https://nebius.github.io/pysdk/nebius.api.nebius.common.v1.OperationServiceClient.html#get) will return a fully functional [`Operation`](https://nebius.github.io/pysdk/nebius.aio.operation.Operation.html). Other methods like [`list`](https://nebius.github.io/pysdk/nebius.api.nebius.common.v1.OperationServiceClient.html#list) or [`list_operations_by_parent`](https://nebius.github.io/pysdk/nebius.api.nebius.compute.v1.DiskServiceClient.html#list_operations_by_parent) from Compute will contain an internal [`Operation`](https://nebius.github.io/pysdk/nebius.api.nebius.common.v1.Operation.html) representation object, that cannot be awaited or polled as a normal [`Operation`](https://nebius.github.io/pysdk/nebius.aio.operation.Operation.html).

##### Retrieve additional metadata

Sometimes you need more than just a result of your request. For instance, if you have problems, you may want to provide more information about the request to the Nebius support team. Service methods do not return basic coroutines, they return [`Request`](https://nebius.github.io/pysdk/nebius.aio.request.Request.html) objects, that can provide more information about the request itself.

Here is an example how to retrieve the request ID and the trace ID for referencing. In most cases, the error will contain them already, but maybe you want to reference a successful request as well. The example:

```python
request = service.get(req)  # Note, that we don't await immediately

# all three can be awaited in any order, or simultaneously
response = await request
request_id = await request.request_id()
trace_id = await request.trace_id()

log.info(f"Server answered: {response}; Request ID: {request_id} and Trace ID: {trace_id}")
```

Or in the case of a synchronous context:

```python
request = service.get(req)  # Note, that we don't await immediately

# all three can be called in any order, the first call will start the request and wait till completion
response = request.wait()
request_id = request.request_id_sync()
trace_id = request.trace_id_sync()

log.info(f"Server answered: {response}; Request ID: {request_id} and Trace ID: {trace_id}")
```

##### Parse errors

Sometimes things go wrong. There are many `Exception`s a request can raise, but some of them are created on a server and sent back. These exceptions will derive from the [`nebius.aio.service_error.RequestError`](https://nebius.github.io/pysdk/nebius.aio.service_error.RequestError.html). This error will contain a request status and additional information from the server, if there was any.

You can simply print the `RequestError` to see all the info in a readable format, or you can parse it and retrieve the [`nebius.aio.service_error.RequestStatusExtended`](https://nebius.github.io/pysdk/nebius.aio.service_error.RequestStatusExtended.html) located in the `err.status` of the excepted error `err`. It will contain all the information in the structured form.

```python
from nebius.aio.service_error import RequestError

try:
    response = await service.get(req)
except RequestError as err:
    log.exception(f"Caught request error {err}")
```

Do not forget to save both the request ID and the trace ID from the output, in case you will have to submit something to the support.

### Calling `update` methods on resources

Any `update` method on resources requires either to pass a manually constructed [`x-resetmask`](https://nebius.github.io/pysdk/nebius.base.fieldmask.Mask.html) or to send a full resource specification, previously obtained by the corresponding `get` method and then modified by your code. Here are both examples:

#### Sending a full specification

Here is an example of doubling the limit on the bucket. In this example, we receive the specification, change it and then send it back.

```python
from nebius.api.nebius.storage.v1 import UpdateBucketRequest

bucket = await service.get(req)
bucket.spec.max_size_bytes *= 2  # Example of the change
operation = await service.update(
    UpdateBucketRequest(
        metadata=bucket.metadata,
        spec=bucket.spec,
    ),
)
```

This operation respects the resource version, thus if somebody was modifying the same resource at the same time, one of your requests will not be accepted. You may omit resource version check by resetting the `metadata.resource_version`. Simply set it to **0** and your update will be applied in any situation:

```python
from nebius.api.nebius.storage.v1 import UpdateBucketRequest

bucket = await service.get(req)
bucket.spec.max_size_bytes *= 2  # Example of the change
bucket.metadata.resource_version = 0  # This will skip version check and fully overwrite the resource
operation = await service.update(
    UpdateBucketRequest(
        metadata=bucket.metadata,
        spec=bucket.spec,
    ),
)
```

This will **fully replace** the bucket specification with the one you've sent, overwriting any changes that could have been made by any concurrent updates.


#### Updating with manually set `X-ResetMask`

You may want to send partial updates without requesting a full specification beforehand, if your update does not require incremental changes, but only value replacements. This process will require manual setting of the `X-ResetMask` in the metadata, if you need to set any value to its default (in terms of ProtoBuf). Any unset or default fields without the mask set, will not be overwritten.

Here is an example of resetting the limit on the bucket:

```python
from nebius.api.nebius.storage.v1 import UpdateBucketRequest
from nebius.api.nebius.common.v1 import ResourceMetadata
from nebius.base.metadata import Metadata

md = Metadata()
md["X-ResetMask"] = "spec.max_size_bytes"
operation = await service.update(
    UpdateBucketRequest(
        metadata=ResourceMetadata(
            id="some-bucket-id",  # Required to identify the resource
        )
    ),
    metadata=md,
)
```

This example will only reset `max_size_bytes` in the bucket, clearing the limit, but won't unset or change anything else.

> **Note**: Our internal field masks have more granularity than google ones, so they are incompatible. You can read more on the masks in the Nebius API documentation.

> **Note**: Please read the API documentation before modifying lists and maps using manually set masks.

### Contributing

Contributions are welcome! Please refer to the [contributing guidelines](CONTRIBUTING.md) for more information.

### License

This project is licensed under the MIT License. See the [LICENSE](LICENSE) file for details.

Copyright (c) 2025 Nebius B.V.

            

Raw data

            {
    "_id": null,
    "home_page": null,
    "name": "nebius",
    "maintainer": null,
    "docs_url": null,
    "requires_python": ">=3.10",
    "maintainer_email": null,
    "keywords": null,
    "author": null,
    "author_email": "Daniil Drizhuk <complynx@nebius.com>, Ivan Kornilov <ivan.kornilov@nebius.com>, Marat Reymers <marat-reymers@nebius.com>, Andrei Zenkov <andrei.zenkov@nebius.com>",
    "download_url": "https://files.pythonhosted.org/packages/36/dd/cd7291e670badf39c3a0544d4224c26b4cab28dea3d9d547e1b726cc82d6/nebius-0.2.53.tar.gz",
    "platform": null,
    "description": "# Nebius Python\u00ae SDK\n\nThe **Nebius Python\u00ae SDK** is a comprehensive client library for interacting with [nebius.com](https://nebius.com) services.\nBuilt on gRPC, it supports all APIs defined in the [Nebius API repository](https://github.com/nebius/api).\nThis SDK simplifies resource management, authentication, and communication with Nebius services, making it a valuable tool for developers.\n\n> **Note**: \"Python\" and the Python logos are trademarks or registered trademarks of the Python Software Foundation, used by Nebius B.V. with permission from the Foundation.\n\n### Full documentation and reference\n\nTo see all the services and their methods, look into the [API reference](https://nebius.github.io/pysdk/apiReference.html).\n\n### Installation\n\n```bash\npip install nebius\n```\n\nIf you've received this module in a zip archive or checked out from git, install it as follows:\n\n```bash\npip install ./path/to/your/pysdk\n```\n\n### Example\n\nWorking examples in `src/examples`.\nTry it out as follows:\n```bash\nNEBIUS_IAM_TOKEN=$(nebius iam get-access-token) python -m ./path/to/your/pysdk/src/examples/basic.py your-project-id\n```\n\n### How-to\n\n#### Initialize\n\n```python\nfrom nebius.sdk import SDK\n\nsdk = SDK()\n```\n\nThis will initialize the [SDK](https://nebius.github.io/pysdk/nebius.sdk.SDK.html) with an IAM token from a `NEBIUS_IAM_TOKEN` env var.\nIf you want to use different ways of authorization, read further.\n\nSee the following how-to's on how to provide your crerentials:\n\n##### Initialize using an IAM Token\n\nYou can also initialize the `SDK` by providing the token directly or from the other environment variable, here are examples how to do that:\n\n```python\nimport os\nfrom nebius.sdk import SDK\nfrom nebius.aio.token.static import Bearer, EnvBearer  # [1]\nfrom nebius.aio.token.token import Token  # [2]\n\nsdk = SDK(credentials=os.environ.get(\"NEBIUS_IAM_TOKEN\", \"\"))\n#or\nsdk = SDK(credentials=Bearer(os.environ.get(\"NEBIUS_IAM_TOKEN\", \"\")))\n#or\nsdk = SDK(credentials=EnvBearer(\"NEBIUS_IAM_TOKEN\"))\n#or\nsdk = SDK(credentials=Bearer(Token(os.environ.get(\"NEBIUS_IAM_TOKEN\", \"\"))))\n```\n[[1](https://nebius.github.io/pysdk/nebius.aio.token.static.html), [2](https://nebius.github.io/pysdk/nebius.aio.token.token.html)]\n\nNow, your application will get token from the local Env variable, as in the example above, but provided in several other ways.\n\n##### Initialize using CLI config\n\nIf you've set up [Nebius AI Cloud CLI](https://docs.nebius.com/cli), you can initialize the `SDK` using [CLI config](https://nebius.github.io/pysdk/nebius.aio.cli_config.Config.html):\n\n```python\nfrom nebius.sdk import SDK\nfrom nebius.aio.cli_config import Config\n\nsdk = SDK(config_reader=Config())\n```\n\nThis will also import the domain if the endpoint parameter is in the config and the domain wasn't set explicitly.\n\n*Keep in mind, that it will get the token from the `NEBIUS_IAM_TOKEN` environment variable if it is set, or use `NEBIUS_PROFILE` for selecting the profile, the same way CLI does. To stop that from happening, set `Config(no_env=True)`*\n\nConfig reader also helps with getting the default parent ID if necessary:\n\n```python\nfrom nebius.aio.cli_config import Config\n\nprint(f\"My default parent ID: {Config().parent_id}\")\n```\n\nCheck the [`Config` documentation](https://nebius.github.io/pysdk/nebius.aio.cli_config.Config.html) for more settings like file path, profile name or environment variables names.\n\n##### Initialize with the private key file\n\nIf you have a private key and a service account, you may want to authorize using them. Here is an example of how to do it.\n\nReplace in the IDs in the following example with your service account and public key ID pair, related to the private key you have.\nYou need to have a `private_key.pem` file on your machine, modify the file path in the example accordingly.\n\n```python\nfrom nebius.sdk import SDK\nfrom nebius.base.service_account.pk_file import Reader as PKReader  # [1]\n\nsdk = SDK(\n    credentials=PKReader(\n        filename=\"location/of/your/private_key.pem\",\n        public_key_id=\"public-key-id\",\n        service_account_id=\"your-service-account-id\",\n    ),\n)\n#or without importing PKReader:\nsdk = SDK(\n    service_account_private_key_file_name=\"location/of/your/private_key.pem\",\n    service_account_public_key_id=\"public-key-id\",\n    service_account_id=\"your-service-account-id\",\n)\n```\n[[1](https://nebius.github.io/pysdk/nebius.base.service_account.pk_file.Reader.html)]\n\n##### Initialize with a credentials file\n\nAssuming you have a joint credentials file with a private key and all the IDs inside.\n\n```python\nfrom nebius.sdk import SDK\nfrom nebius.base.service_account.credentials_file import Reader as CredentialsReader  # [1]\n\nsdk = SDK(\n    credentials=CredentialsReader(\n        filename=\"location/of/your/credentials.json\",\n    ),\n)\n#or without importing CredentialsReader:\nsdk = SDK(\n    credentials_file_name=\"location/of/your/credentials.json\",\n)\n```\n[[1](https://nebius.github.io/pysdk/nebius.base.service_account.credentials_file.Reader.html)]\n\n#### Test the SDK\n\nNow as you've initialized the SDK, you may want to test whether your credentials are ok, everything works and you have a good connection.\n\nTo test the SDK, we provide a convenient method [`SDK.whoami`](https://nebius.github.io/pysdk/nebius.sdk.SDK.html#whoami), that will return the basic information about the profile, you've authenticated with:\n\n```python\nimport asyncio\n\nasync def my_call():\n    async with sdk:\n        print(await sdk.whoami())\n\nasyncio.run(my_call)\n```\n\nIt is important to close the SDK, so all the coroutines and tasks will be gracefully stopped and gathered. It can either be achieved by using `async with`, or by explicitly calling `sdk.close()`:\n\n```python\nimport asyncio\n\nasync def my_call():\n    try:\n        print(await sdk.whoami())\n        # Other calls to SDK\n    finally:\n        await sdk.close()\n\nasyncio.run(my_call)\n```\n\nSDK is created with asyncio in mind, so the best way to call methods of it is to use an async context. But if you haven't started async loop, you can run it synchronously:\n\n```python\ntry:\n    print(sdk.whoami().wait())\n    # Other calls to SDK\nfinally:\n    sdk.sync_close()\n```\n\n*Keep in mind, that this may lead to some problems or infinite locks, even if timeouts have been added. Moreover, synchronous methods won't run inside an async call stack, if you haven't provided a dedicated separate loop for the SDK. And even when the loop is provided, there might be issues or deadlocks.*\n\nClosing the SDK is not strictly necessary, but forgetting to add it may lead to a bunch of annoying errors of unterminated tasks.\n\n#### Test token renewal\n\nIf you use service account, credentials file or something like that, SDK will renew your tokens under the hood. This renewal process will normally report to log if any errors occur. However it might be good to get the repored errors as request errors. In that case, pass special options to the request like so:\n\n\n```python\nimport asyncio\n\nfrom nebius.aio.authorization.options import options_to_metadata\nfrom nebius.aio.token.renewable import (\n    OPTION_RENEW_REQUEST_TIMEOUT,\n    OPTION_RENEW_REQUIRED,\n    OPTION_RENEW_SYNCHRONOUS,\n)\n\nasync def my_call():\n    try:\n        await sdk.whoami(\n            metadata=options_to_metadata(\n                {\n                    OPTION_RENEW_REQUIRED: \"true\",\n                    OPTION_RENEW_SYNCHRONOUS: \"true\",\n                    OPTION_RENEW_REQUEST_TIMEOUT: \".9\",\n                }\n            )\n        )\n    except Exception as err:\n        print(f\"something is wrong with your token: {err=}\")\n    finally:\n        await sdk.close()\n\nasyncio.run(my_call)\n```\n\nYou can pass these options to any request, or add them to your metadata with `add_options_to_metadata`.\n\n#### Call some method\n\nNow as you have your SDK initialized and tested, you may work with our services and call their methods with it. Here and further we assume, that the [SDK](https://nebius.github.io/pysdk/nebius.sdk.SDK.html) is initialized and is located in the `sdk` variable. We also omit closing the SDK.\n\nAll the services API classes are located in submodules of `nebius.api.nebius`. [The reference can be found here](https://nebius.github.io/pysdk/apiReference.html). The `nebius.api.nebius` also includes all the raw gRPC and ProtoBuf classes.\n\nAs an example how to use the API, let's receive a bucket from a storage service by its ID:\n\n```python\nimport asyncio\n\nfrom nebius.api.nebius.storage.v1 import GetBucketRequest\nfrom nebius.api.nebius.storage.v1 import BucketServiceClient\n\nasync def my_call():\n    service = BucketServiceClient(sdk)\n    return await service.get(GetBucketRequest(\n        id=\"some-bucket-id\",\n    ))\n\nasyncio.run(my_call())\n```\n\nSame thing, but synchronously:\n\n```python\nimport asyncio\n\nfrom nebius.api.nebius.storage.v1 import BucketServiceClient, GetBucketRequest\n\nservice = BucketServiceClient(sdk)\nresult = service.get(GetBucketRequest(\n    id=\"some-bucket-id\",\n)).wait()\n```\n\n##### Parent ID\n\nSome methods may include `parent_id` in the requests, for certain methods this field is populated automatically:\n* Methods `list` and `get_by_name` with an empty `parent_id`\n* All other methods, except `update`, with an empty `metadata.parent_id`\n\nThe `parent_id` will only be set if it was preset at the initialization, either from the [CLI `Config`](https://nebius.github.io/pysdk/nebius.aio.cli_config.Config.html) or from the `parent_id` attribute from the [SDK](https://nebius.github.io/pysdk/nebius.sdk.SDK.html). You can disable it, retaining the CLI Config, if you set it up with `no_parent_id=True`.\n\n##### Operations\n\nMany core methods return a [`nebius.aio.Operation`](https://nebius.github.io/pysdk/nebius.aio.operation.Operation.html) object, representing a time-consuming asynchronous operation. For example, the `create` request from the `BucketServiceClient` above is one of such cases. The [`nebius.aio.Operation`](https://nebius.github.io/pysdk/nebius.aio.operation.Operation.html) is a wrapper class that provides convenient methods for working with operations. It can be awaited util completion.\n\nHere is an async example:\n\n```python\nfrom nebius.api.nebius.storage.v1 import BucketServiceClient, CreateBucketRequest\n\nservice = BucketServiceClient(sdk)\noperation = await service.create(CreateBucketRequest(\n    # fill-in necessary fields\n))\nawait operation.wait()\nprint(f\"New bucket ID: {operation.resource_id}\")\n```\n\nOr synchronously:\n\n```python\nfrom nebius.api.nebius.storage.v1 import BucketServiceClient, CreateBucketRequest\n\nservice = BucketServiceClient(sdk)\noperation = service.create(CreateBucketRequest(\n    # fill-in necessary fields\n)).wait()\noperation.wait_sync()\nprint(f\"New bucket ID: {operation.resource_id}\")\n```\n\n##### Operations service\n\nIf you need to get an operation or list operations, you will need an [`OperationServiceClient`](https://nebius.github.io/pysdk/nebius.api.nebius.common.v1.OperationServiceClient.html).\n\nThe [`OperationServiceClient`](https://nebius.github.io/pysdk/nebius.api.nebius.common.v1.OperationServiceClient.html), despite being located in [`nebius.api.nebius.common.v1`](https://nebius.github.io/pysdk/nebius.api.nebius.common.v1.html), **cannot** be used the same way as other services. The real operation service must be acquired from the source service of your operation using [`service.operation_service()`](https://nebius.github.io/pysdk/nebius.aio.client.ClientWithOperations.html#operation_service) method.\n\nExample of listing operations and getting operation by service (only async for brevity):\n```python\nfrom nebius.api.nebius.common.v1 import GetOperationRequest, ListOperationsRequest\nfrom nebius.api.nebius.storage.v1 import BucketServiceClient\n\nservice = BucketServiceClient(sdk)\nop_service = service.operation_service()\n\nresp = await op_service.list(ListOperationsRequest(resource_id=\"your-bucket-id\"))\nop_id = resp.operations[0].id # The elements of resp.operations are not of type Operation!\nreal_operation = await op_service.get(GetOperationRequest(id=op_id))\n\n# Get returns the real operation that can be awaited.\nawait real_operation.wait()\n```\n\n**NOTE** As you can see from the example, only [`get`](https://nebius.github.io/pysdk/nebius.api.nebius.common.v1.OperationServiceClient.html#get) will return a fully functional [`Operation`](https://nebius.github.io/pysdk/nebius.aio.operation.Operation.html). Other methods like [`list`](https://nebius.github.io/pysdk/nebius.api.nebius.common.v1.OperationServiceClient.html#list) or [`list_operations_by_parent`](https://nebius.github.io/pysdk/nebius.api.nebius.compute.v1.DiskServiceClient.html#list_operations_by_parent) from Compute will contain an internal [`Operation`](https://nebius.github.io/pysdk/nebius.api.nebius.common.v1.Operation.html) representation object, that cannot be awaited or polled as a normal [`Operation`](https://nebius.github.io/pysdk/nebius.aio.operation.Operation.html).\n\n##### Retrieve additional metadata\n\nSometimes you need more than just a result of your request. For instance, if you have problems, you may want to provide more information about the request to the Nebius support team. Service methods do not return basic coroutines, they return [`Request`](https://nebius.github.io/pysdk/nebius.aio.request.Request.html) objects, that can provide more information about the request itself.\n\nHere is an example how to retrieve the request ID and the trace ID for referencing. In most cases, the error will contain them already, but maybe you want to reference a successful request as well. The example:\n\n```python\nrequest = service.get(req)  # Note, that we don't await immediately\n\n# all three can be awaited in any order, or simultaneously\nresponse = await request\nrequest_id = await request.request_id()\ntrace_id = await request.trace_id()\n\nlog.info(f\"Server answered: {response}; Request ID: {request_id} and Trace ID: {trace_id}\")\n```\n\nOr in the case of a synchronous context:\n\n```python\nrequest = service.get(req)  # Note, that we don't await immediately\n\n# all three can be called in any order, the first call will start the request and wait till completion\nresponse = request.wait()\nrequest_id = request.request_id_sync()\ntrace_id = request.trace_id_sync()\n\nlog.info(f\"Server answered: {response}; Request ID: {request_id} and Trace ID: {trace_id}\")\n```\n\n##### Parse errors\n\nSometimes things go wrong. There are many `Exception`s a request can raise, but some of them are created on a server and sent back. These exceptions will derive from the [`nebius.aio.service_error.RequestError`](https://nebius.github.io/pysdk/nebius.aio.service_error.RequestError.html). This error will contain a request status and additional information from the server, if there was any.\n\nYou can simply print the `RequestError` to see all the info in a readable format, or you can parse it and retrieve the [`nebius.aio.service_error.RequestStatusExtended`](https://nebius.github.io/pysdk/nebius.aio.service_error.RequestStatusExtended.html) located in the `err.status` of the excepted error `err`. It will contain all the information in the structured form.\n\n```python\nfrom nebius.aio.service_error import RequestError\n\ntry:\n    response = await service.get(req)\nexcept RequestError as err:\n    log.exception(f\"Caught request error {err}\")\n```\n\nDo not forget to save both the request ID and the trace ID from the output, in case you will have to submit something to the support.\n\n### Calling `update` methods on resources\n\nAny `update` method on resources requires either to pass a manually constructed [`x-resetmask`](https://nebius.github.io/pysdk/nebius.base.fieldmask.Mask.html) or to send a full resource specification, previously obtained by the corresponding `get` method and then modified by your code. Here are both examples:\n\n#### Sending a full specification\n\nHere is an example of doubling the limit on the bucket. In this example, we receive the specification, change it and then send it back.\n\n```python\nfrom nebius.api.nebius.storage.v1 import UpdateBucketRequest\n\nbucket = await service.get(req)\nbucket.spec.max_size_bytes *= 2  # Example of the change\noperation = await service.update(\n    UpdateBucketRequest(\n        metadata=bucket.metadata,\n        spec=bucket.spec,\n    ),\n)\n```\n\nThis operation respects the resource version, thus if somebody was modifying the same resource at the same time, one of your requests will not be accepted. You may omit resource version check by resetting the `metadata.resource_version`. Simply set it to **0** and your update will be applied in any situation:\n\n```python\nfrom nebius.api.nebius.storage.v1 import UpdateBucketRequest\n\nbucket = await service.get(req)\nbucket.spec.max_size_bytes *= 2  # Example of the change\nbucket.metadata.resource_version = 0  # This will skip version check and fully overwrite the resource\noperation = await service.update(\n    UpdateBucketRequest(\n        metadata=bucket.metadata,\n        spec=bucket.spec,\n    ),\n)\n```\n\nThis will **fully replace** the bucket specification with the one you've sent, overwriting any changes that could have been made by any concurrent updates.\n\n\n#### Updating with manually set `X-ResetMask`\n\nYou may want to send partial updates without requesting a full specification beforehand, if your update does not require incremental changes, but only value replacements. This process will require manual setting of the `X-ResetMask` in the metadata, if you need to set any value to its default (in terms of ProtoBuf). Any unset or default fields without the mask set, will not be overwritten.\n\nHere is an example of resetting the limit on the bucket:\n\n```python\nfrom nebius.api.nebius.storage.v1 import UpdateBucketRequest\nfrom nebius.api.nebius.common.v1 import ResourceMetadata\nfrom nebius.base.metadata import Metadata\n\nmd = Metadata()\nmd[\"X-ResetMask\"] = \"spec.max_size_bytes\"\noperation = await service.update(\n    UpdateBucketRequest(\n        metadata=ResourceMetadata(\n            id=\"some-bucket-id\",  # Required to identify the resource\n        )\n    ),\n    metadata=md,\n)\n```\n\nThis example will only reset `max_size_bytes` in the bucket, clearing the limit, but won't unset or change anything else.\n\n> **Note**: Our internal field masks have more granularity than google ones, so they are incompatible. You can read more on the masks in the Nebius API documentation.\n\n> **Note**: Please read the API documentation before modifying lists and maps using manually set masks.\n\n### Contributing\n\nContributions are welcome! Please refer to the [contributing guidelines](CONTRIBUTING.md) for more information.\n\n### License\n\nThis project is licensed under the MIT License. See the [LICENSE](LICENSE) file for details.\n\nCopyright (c) 2025 Nebius B.V.\n",
    "bugtrack_url": null,
    "license": "The MIT License (MIT)\n        \n        Copyright (c) 2025 Nebius B.V.\n        \n        Permission is hereby granted, free of charge, to any person obtaining a copy\n        of this software and associated documentation files (the \"Software\"), to deal\n        in the Software without restriction, including without limitation the rights\n        to use, copy, modify, merge, publish, distribute, sublicense, and/or sell\n        copies of the Software, and to permit persons to whom the Software is\n        furnished to do so, subject to the following conditions:\n        \n        The above copyright notice and this permission notice shall be included in all\n        copies or substantial portions of the Software.\n        \n        THE SOFTWARE IS PROVIDED \"AS IS\", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR\n        IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,\n        FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE\n        AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER\n        LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,\n        OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE\n        SOFTWARE.\n        ",
    "summary": "Nebius Python SDK",
    "version": "0.2.53",
    "project_urls": null,
    "split_keywords": [],
    "urls": [
        {
            "comment_text": null,
            "digests": {
                "blake2b_256": "fd377c2e11838bb26fcc0018a26bba7bef0ce647866924ac431f2fd7772ee117",
                "md5": "98a7de9a2dff078115bff047013824ab",
                "sha256": "61fa88b4bf125508f941cd111b86a5a4c242e856e488293c5390e44701c899ce"
            },
            "downloads": -1,
            "filename": "nebius-0.2.53-py3-none-any.whl",
            "has_sig": false,
            "md5_digest": "98a7de9a2dff078115bff047013824ab",
            "packagetype": "bdist_wheel",
            "python_version": "py3",
            "requires_python": ">=3.10",
            "size": 917461,
            "upload_time": "2025-08-19T12:33:21",
            "upload_time_iso_8601": "2025-08-19T12:33:21.586705Z",
            "url": "https://files.pythonhosted.org/packages/fd/37/7c2e11838bb26fcc0018a26bba7bef0ce647866924ac431f2fd7772ee117/nebius-0.2.53-py3-none-any.whl",
            "yanked": false,
            "yanked_reason": null
        },
        {
            "comment_text": null,
            "digests": {
                "blake2b_256": "36ddcd7291e670badf39c3a0544d4224c26b4cab28dea3d9d547e1b726cc82d6",
                "md5": "2e37280322dc2223d00a384fcddb3df7",
                "sha256": "724e881fc999b1457305b2d02e8cf1bf07ecdff866da8604d401b6ca960144e0"
            },
            "downloads": -1,
            "filename": "nebius-0.2.53.tar.gz",
            "has_sig": false,
            "md5_digest": "2e37280322dc2223d00a384fcddb3df7",
            "packagetype": "sdist",
            "python_version": "source",
            "requires_python": ">=3.10",
            "size": 622540,
            "upload_time": "2025-08-19T12:33:23",
            "upload_time_iso_8601": "2025-08-19T12:33:23.461210Z",
            "url": "https://files.pythonhosted.org/packages/36/dd/cd7291e670badf39c3a0544d4224c26b4cab28dea3d9d547e1b726cc82d6/nebius-0.2.53.tar.gz",
            "yanked": false,
            "yanked_reason": null
        }
    ],
    "upload_time": "2025-08-19 12:33:23",
    "github": false,
    "gitlab": false,
    "bitbucket": false,
    "codeberg": false,
    "lcname": "nebius"
}