YDB Python SDK

https://github.com/ydb-platform/ydb-python-sdk
Admin
YDB Python SDK is the officially supported Python client for YDB, providing libraries and APIs for...
Tokens:287,770
Snippets:1,973
Trust Score:9.4
License:Apache-2.0
Update:1 week ago
Show doc for...
Tokens:
Raw
### Install dependencies from examples/opentelemetry directory

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/examples/opentelemetry/README.md

Equivalent installation command when the shell is already in the example directory.

```sh
pip install -e '../..[opentelemetry]' -r requirements.txt
```

--------------------------------

### Start full stack from the example folder

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/examples/opentelemetry/README.md

Alternative command to start the full stack when the current directory is examples/opentelemetry. The build context is resolved correctly because it is relative to the compose file.

```sh
cd /path/to/ydb-python-sdk/examples/opentelemetry
docker compose -f compose-e2e.yaml up --build
```

--------------------------------

### Install dependencies from repository root

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/examples/opentelemetry/README.md

Creates a virtual environment and installs the SDK in editable mode with OpenTelemetry extras, plus the example's requirements.

```sh
python3 -m venv .venv
source .venv/bin/activate   # Windows: .venv\Scripts\activate
pip install -e '.[opentelemetry]' -r examples/opentelemetry/requirements.txt
```

--------------------------------

### Install Python dependencies for YDB examples

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/examples/README.md

Install the required Python packages: iso8601 for date/time formatting, ydb for the YDB SDK, and yandexcloud for running examples on YDB Managed Services.

```bash
# Date and time formatting functions used in some examples
python3 -m pip install iso8601
# YDB Python SDK
python3 -m pip install ydb
# Yandex Cloud Python SDK (for running the examples on YDB Managed Services)
python3 -m pip install yandexcloud
```

--------------------------------

### Run example on local single node YDB instance

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/examples/README.md

Configure and run a local YDB cluster, then run the example with anonymous authentication. Sets YDB_ANONYMOUS_CREDENTIALS=1, YDB_ENDPOINT=grpc://localhost:2136, YDB_DATABASE=/local, and runs the script with the example path.

```bash
# Change the directory for the particular example, e.g. basic_example_v1
EXAMPLE=basic_example_v1
cd $EXAMPLE
# Enable the anonymous authentication mode
export YDB_ANONYMOUS_CREDENTIALS=1
# Set the database path and its endpoint
export YDB_ENDPOINT=grpc://localhost:2136
export YDB_DATABASE=/local
# Run the script
python3 __main__.py -e $YDB_ENDPOINT -d $YDB_DATABASE -p python-examples/$EXAMPLE
```

--------------------------------

### Create table with covered index

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/table.rst

Full example that creates a table with a covered index from the start using the table client.

```python
driver.table_client.create_table(
    "/local/users",
    ydb.TableDescription()
    .with_columns(
        ydb.Column("id",    ydb.PrimitiveType.Uint64),
        ydb.Column("name",  ydb.OptionalType(ydb.PrimitiveType.Utf8)),
        ydb.Column("email", ydb.OptionalType(ydb.PrimitiveType.Utf8)),
        ydb.Column("age",   ydb.OptionalType(ydb.PrimitiveType.Uint32)),
    )
    .with_primary_keys(
```

--------------------------------

### Typical local run with Docker and pip

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/opentelemetry.rst

Runs the example script on the host with YDB in Docker. Compose must be started before pip/python. After running, open http://localhost:3000 (Grafana) to explore traces via Tempo.

```sh
docker compose up -d
pip install -e '.[opentelemetry]' -r examples/opentelemetry/requirements.txt
python examples/opentelemetry/otel_example.py
```

--------------------------------

### Start full stack with Docker Compose (from repo root)

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/examples/opentelemetry/README.md

Starts YDB, OTLP collector, Tempo, Prometheus, Grafana, and the otel-example and load-generator services using compose-e2e.yaml. The first run builds the otel-example image from local SDK source.

```sh
cd /path/to/ydb-python-sdk
docker compose -f examples/opentelemetry/compose-e2e.yaml up --build
```

--------------------------------

### Run example on self-hosted YDB cluster

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/examples/README.md

Run an example on a self-hosted cluster using token authentication. Obtains a token via `ydb auth get-token`, sets YDB_ENDPOINT and YDB_DATABASE, and runs the script.

```bash
# Change the directory for the particular example, e.g. basic_example_v1
EXAMPLE=basic_example_v1
cd $EXAMPLE
# Obtain the YDB authentication token
export YDB_ACCESS_TOKEN_CREDENTIALS=`ydb -p ydb1 auth get-token -f`
# Set the database path and its endpoint
export YDB_ENDPOINT=grpcs://cluster-endpoint.domain.com:2136
export YDB_DATABASE=/Root/testdb
# Run the script
python3 __main__.py -e $YDB_ENDPOINT -d $YDB_DATABASE -p python-examples/$EXAMPLE
```

--------------------------------

### Run the OpenTelemetry example

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/examples/opentelemetry/README.md

Runs the main example script after YDB is up. Defaults to YDB grpc://localhost:2136 and OTLP http://localhost:4317.

```sh
python examples/opentelemetry/otel_example.py
```

--------------------------------

### Set up development environment with venv and install dependencies

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/AGENTS.md

Creates a Python virtual environment, activates it, and installs the package in editable mode along with tox. Always activate the virtual environment before running any project commands.

```sh
python3 -m venv .venv
source .venv/bin/activate
pip install -e . tox
```

--------------------------------

### Run example on managed YDB instance in Yandex Cloud

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/examples/README.md

Runs an example on a managed YDB instance using service account key authentication. Sets YDB_SERVICE_ACCOUNT_KEY_FILE_CREDENTIALS, YDB_ENDPOINT, and YDB_DATABASE, then runs the script.

```bash
# Change the directory for the particular example, e.g. basic_example_v1
EXAMPLE=basic_example_v1
cd $EXAMPLE
# Set the path to the service account key file
export YDB_SERVICE_ACCOUNT_KEY_FILE_CREDENTIALS=$HOME/key-ydb-sa-0.json
# Set the database path and its endpoint
export YDB_ENDPOINT=grpcs://ydb.serverless.yandexcloud.net:2135
export YDB_DATABASE=/ru-central1/b1gfvslmokutuvt2g019/etnvbffeqegu1ub2rg2o
# Run the script
python3 __main__.py -e $YDB_ENDPOINT -d $YDB_DATABASE -p python-examples/$EXAMPLE
```

--------------------------------

### Install OTLP/gRPC exporter

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/opentelemetry.rst

Installs the OTLP/gRPC exporter, which works with Jaeger, Tempo, and other tracing backends.

```sh
# OTLP/gRPC exporter (works with Jaeger, Tempo, and others)
pip install opentelemetry-exporter-otlp-proto-grpc
```

--------------------------------

### Explain query plan directly on a session

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/query.rst

Call session.explain to get the execution plan for a statement. The example shows the call inside a retry_operation_sync callback.

```python
def callee(session: ydb.QuerySession):
    plan = session.explain("SELECT * FROM users WHERE id = 1")
    print(plan)

pool.retry_operation_sync(callee)
```

--------------------------------

### Enable OpenTelemetry tracing with enable_tracing()

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/observability.rst

Enables the recommended OpenTelemetry tracing backend. Call enable_tracing() once to install the OtelTracingProvider. See the OpenTelemetry page for installation, exporters, and a full runnable example.

```python
from ydb.opentelemetry import enable_tracing

enable_tracing()
```

--------------------------------

### Create table with PartitioningSettings via with_partitioning_settings

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/table.rst

Creates a table with partitioning settings passed to `TableDescription.with_partitioning_settings`. The example sets minimum and maximum partition counts and enables load-based partitioning.

```python
driver.table_client.create_table(
    "/local/events",
    ydb.TableDescription()
    .with_columns(
        ydb.Column("id",   ydb.PrimitiveType.Uint64),
        ydb.Column("data", ydb.OptionalType(ydb.PrimitiveType.Utf8)),
    )
    .with_primary_keys("id")
    .with_partitioning_settings(
        ydb.PartitioningSettings()
        .with_min_partitions_count(4)
        .with_max_partitions_count(64)
        .with_partitioning_by_load(ydb.FeatureFlag.ENABLED)
    ),
)
```

--------------------------------

### Install YDB Python SDK from sources

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/quickstart.rst

Installs the YDB Python SDK from the root of the repository in editable mode.

```default
pip install -e .
```

--------------------------------

### Enable tracing with explicit provider

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/opentelemetry.rst

Advanced usage: builds the provider explicitly and installs it through the vendor-neutral entrypoint. This is what the convenience wrapper does.

```python
from ydb.observability import enable_tracing
from ydb.opentelemetry import OtelTracingProvider

enable_tracing(OtelTracingProvider())          # or OtelTracingProvider(my_tracer)
```

--------------------------------

### Start Docker Compose infra with telemetry profile

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/tests/slo/README.md

Starts the SLO infrastructure (YDB, Prometheus) from the ydb-slo-action repo using the telemetry profile. Run from the repo root; the infra network must be named ydb_cluster.

```bash
docker compose -f deploy/compose.yml --profile telemetry up -d --build
```

--------------------------------

### Install YDB Python SDK via PyPI

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/quickstart.rst

Installs the YDB Python SDK from PyPI using pip.

```default
pip install ydb
```

--------------------------------

### Install YDB SDK with OpenTelemetry extra

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/opentelemetry.rst

Installs the ydb package with the 'opentelemetry' extra, which pulls in opentelemetry-api.

```sh
pip install ydb[opentelemetry]
```

--------------------------------

### Start only YDB with Docker Compose

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/examples/opentelemetry/README.md

Starts the minimal docker-compose.yml from the repository root. Wait until the ydb container is healthy or port 2136 is open before continuing.

```sh
cd /path/to/ydb-python-sdk
docker compose up -d
# wait until the ydb container is healthy / port 2136 is open, then continue
```

--------------------------------

### Run full stack with Docker Compose

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/opentelemetry.rst

Starts the full stack (YDB + OTLP + Tempo + Grafana) and runs the otel-example service once. The first run builds the image from local SDK source; subsequent runs reuse the cached image. Pass --build again if you change the SDK or demo script.

```sh
docker compose -f examples/opentelemetry/compose-e2e.yaml up --build
```

--------------------------------

### Idempotent directory setup with make_directory and CREATE TABLE

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/scheme.rst

Use this snippet during application initialization to create a directory structure, tolerating existing directories by catching `ydb.SchemeError`. It also creates a table using `CREATE TABLE IF NOT EXISTS` within a `QuerySessionPool`.

```python
import ydb

def setup_schema(driver: ydb.Driver, base_path: str) -> None:
    for subdir in ("tables", "topics", "coordination"):
        path = f"{base_path}/{subdir}"
        try:
            driver.scheme_client.make_directory(path)
        except ydb.SchemeError:
            pass  # already exists

    with ydb.QuerySessionPool(driver) as pool:
        pool.execute_with_retries(
            f"CREATE TABLE IF NOT EXISTS `{base_path}/tables/events` "
            "(id Uint64, payload Utf8, PRIMARY KEY (id))"
        )
```

--------------------------------

### Enable tracing with async driver

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/opentelemetry.rst

Call enable_tracing() once at startup to enable tracing with the async driver. The example shows creating an async driver and query session pool, then executing a query.

```python
import asyncio
import ydb
from ydb.opentelemetry import enable_tracing

enable_tracing()

async def main():
    async with ydb.aio.Driver(
        endpoint="grpc://localhost:2136",
        database="/local",
    ) as driver:
        await driver.wait(timeout=5)
        async with ydb.aio.QuerySessionPool(driver) as pool:
            await pool.execute_with_retries("SELECT 1")

asyncio.run(main())
```

--------------------------------

### Describe a path with driver.scheme_client.describe_path

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/scheme.rst

Returns a SchemeEntry for any path (table, topic, directory, coordination node, etc.). The example prints the entry's name, owner, type, and approximate size.

```python
entry = driver.scheme_client.describe_path("/local/my_app/users")

print(entry.name)        # "users"
print(entry.owner)       # owner login
print(entry.type)        # SchemeEntryType.TABLE
print(entry.size_bytes)  # approximate size
```

--------------------------------

### Start YDB with chaos for topic SLO testing

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/AGENTS.md

Runs a Docker Compose playground that kills a DB node roughly every 20 seconds. Use only for changes affecting topic reader/writer reconnection logic.

```sh
docker compose -f tests/slo/playground/configs/compose.yaml up -d
```

--------------------------------

### Asynchronous YDB SDK Example: Create Table, Insert, and Query

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/query.rst

Uses ydb.aio.Driver and ydb.aio.QuerySessionPool to asynchronously execute schema DDL, insert rows in a transaction via retry_tx_async, and read results. Requires a running YDB instance at the specified endpoint and database.

```python
import asyncio
import ydb

async def main():
    async with ydb.aio.Driver(
        endpoint="grpc://localhost:2136", database="/local"
    ) as driver:
        await driver.wait(timeout=5, fail_fast=True)

        async with ydb.aio.QuerySessionPool(driver) as pool:

            # Schema
            await pool.execute_with_retries("DROP TABLE IF EXISTS users")
            await pool.execute_with_retries(
                "CREATE TABLE users (id Uint64, name Utf8, PRIMARY KEY (id))"
            )

            # Insert in a transaction
            async def insert(tx: ydb.aio.QueryTxContext):
                async with await tx.execute(
                    "INSERT INTO users (id, name) VALUES (1, 'Alice'), (2, 'Bob')",
                    commit_tx=True,
                ):
                    pass

            await pool.retry_tx_async(insert)

            # Read
            result_sets = await pool.execute_with_retries(
                "SELECT id, name FROM users"
            )
            for row in result_sets[0].rows:
                print(row["id"], row["name"])

asyncio.run(main())

```

--------------------------------

### Request Arrow-format query results with pyarrow

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/query.rst

Pass ydb.QueryResultSetFormat.ARROW to execute_with_retries to get results as Arrow IPC. Requires pyarrow installed. The example reads the schema and record batch from the result set and converts to a pandas DataFrame.

```python
import pyarrow as pa
import ydb

result_sets = pool.execute_with_retries(
    "SELECT id, name, score FROM users ORDER BY id LIMIT 1000",
    result_set_format=ydb.QueryResultSetFormat.ARROW,
)

for result_set in result_sets:
    schema = pa.ipc.read_schema(pa.py_buffer(result_set.arrow_format_meta.schema))
    batch  = pa.ipc.read_record_batch(pa.py_buffer(result_set.data), schema)
    df = batch.to_pandas()
    print(df.head())
```

--------------------------------

### Use TopicReaderSelector for per-topic settings

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/topic.rst

Use TopicReaderSelector to fine-tune per-topic settings such as start offset or timestamp. The example sets read_from to a specific datetime.

```python
reader = driver.topic_client.reader(
    topic=ydb.TopicReaderSelector(
        path="/local/my-topic",
        read_from=datetime.datetime(2024, 1, 1, tzinfo=datetime.timezone.utc),
    ),
    consumer="my-consumer",
)
```

--------------------------------

### begin(settings)

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/apireference.rst

Explicitly begins a transaction. Returns the transaction context or raises an exception if begin fails.

```APIDOC
## begin(settings)

### Description
Explicitly begins a transaction.

### Method
begin

### Signature
`async begin(settings: BaseRequestSettings | None = None) -> QueryTxContext`

### Parameters
- **settings** (BaseRequestSettings | None) - Optional - An additional request settings.

### Returns
- **QueryTxContext** - The transaction context.

### Exceptions
- Raises an exception if begin fails.
```

--------------------------------

### Usage Example: Timestamp Parameter

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/types.rst

Example of using a Timestamp typed value in a query parameter.

```APIDOC
## Usage Example: Timestamp Parameter

### Description
Example of using a Timestamp typed value in a query parameter.

### Code Example
```python
import datetime
import ydb

pool.execute_with_retries(
    "DECLARE $ts AS Timestamp; SELECT * FROM events WHERE created_at > $ts",
    parameters={
        "$ts": ydb.TypedValue(
            datetime.datetime(2024, 1, 1, tzinfo=datetime.timezone.utc),
            ydb.PrimitiveType.Timestamp,
        )
    },
)
```

```

--------------------------------

### Create a test topic with topic-create

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/AGENTS.md

Creates a test topic at /Root/testdb/slo_topic from the tests/slo/ directory. Requires the virtual environment to be activated first.

```sh
source .venv/bin/activate
python ./src topic-create grpc://localhost:2135 /Root/testdb \
    --path /Root/testdb/slo_topic --debug
```

--------------------------------

### QueryTxContext.begin(settings)

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/apireference.rst

Explicitly begins a transaction. Returns the transaction object or raises an exception if begin fails.

```APIDOC
## QueryTxContext.begin(settings)

### Description
Explicitly begins a transaction. Returns the transaction object or raises an exception if begin fails.

### Method
begin

### Signature
begin(settings: BaseRequestSettings | None = None) → QueryTxContext

### Parameters
- **settings** (BaseRequestSettings | None) - Optional - An additional request settings BaseRequestSettings.

### Returns
- **QueryTxContext** - Transaction object or exception if begin is failed.

```

--------------------------------

### ydb.PrimitiveType

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/apireference.rst

Enumerates all available primitive types that can be used in computations. Provides methods to get and set values in protocol buffers, and a property to get the protocol buffer representation.

```APIDOC
## ydb.PrimitiveType

### Description
Enumerates all available primitive types that can be used in computations.

### Members
- Int32
- Uint32
- Int64
- Uint64
- Int8
- Uint8
- Int16
- Uint16
- Bool
- Double
- Float
- String
- Utf8
- Yson
- Json
- JsonDocument
- UUID
- Date
- Date32
- Datetime
- Datetime64
- Timestamp
- Timestamp64
- TzDate
- TzDatetime
- TzTimestamp
- Interval
- Interval64
- DyNumber

### Methods

#### get_value(value_pb: Value, table_client_settings: TableClientSettings) -> Any
Extracts value from protocol buffer.

**Parameters:**
- value_pb (Value) - The protocol buffer.
- table_client_settings (TableClientSettings) - The table client settings.

**Returns:** A valid value of primitive type.

#### set_value(pb: Value, value: Any) -> None
Sets value in a protocol buffer.

**Parameters:**
- pb (Value) - The protocol buffer.
- value (Any) - A valid value to set.

**Returns:** None

### Properties

#### proto -> Type
Returns protocol buffer representation of a primitive type.

```

--------------------------------

### Get full query stats inside a transaction

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/query.rst

Pass stats_mode=ydb.QueryStatsMode.FULL to tx.execute to get full per-operator statistics. Stats reflect the last executed statement in the transaction.

```python
def callee(tx: ydb.QueryTxContext):
    with tx.execute(
        "SELECT COUNT(*) AS cnt FROM users",
        stats_mode=ydb.QueryStatsMode.FULL,
    ) as results:
        for result_set in results:
            print(result_set.rows[0]["cnt"])

    print(tx.last_query_stats)

pool.retry_tx_sync(callee)
```

--------------------------------

### Create a Topic (Synchronous and Asynchronous)

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/topic.rst

Creates a topic with specified partition counts, retention period, and consumers. The synchronous version uses driver.topic_client.create_topic directly; the asynchronous version uses await. Both are shown as separate code blocks.

```python
# Synchronous
driver.topic_client.create_topic(
    "/local/my-topic",
    min_active_partitions=1,
    max_active_partitions=10,
    retention_period=datetime.timedelta(hours=24),
    consumers=["my-consumer"],
)
```

```python
# Asynchronous
await driver.topic_client.create_topic(
    "/local/my-topic",
    min_active_partitions=1,
    max_active_partitions=10,
    retention_period=datetime.timedelta(hours=24),
    consumers=["my-consumer"],
)
```

--------------------------------

### Create Driver with connection string

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/driver.rst

Alternative way to create a driver using a connection string. The format is `<protocol>://<host>:<port>/?database=<path>`.

```python
driver = ydb.Driver(
    connection_string="grpc://localhost:2136/?database=/local",
)
```

--------------------------------

### Create topic with auto-partitioning settings

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/topic.rst

Enables auto-partitioning when creating a topic. Uses SCALE_UP strategy to split partitions at 70% utilization, with a 60-second stabilization window. The down_utilization_percent parameter only applies to SCALE_UP_AND_DOWN.

```python
import datetime
import ydb

driver.topic_client.create_topic(
    "/local/my-topic",
    consumers=["my-consumer"],
    max_active_partitions=100,
    auto_partitioning_settings=ydb.TopicAutoPartitioningSettings(
        strategy=ydb.TopicAutoPartitioningStrategy.SCALE_UP,
        up_utilization_percent=70,      # split a partition when it reaches 70% capacity
        down_utilization_percent=30,    # merge partitions when below 30% (SCALE_UP_AND_DOWN only)
        stabilization_window=datetime.timedelta(seconds=60),
    ),
)
```

--------------------------------

### Explain query plan as JSON string or dict

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/query.rst

Use pool.explain_with_retries to get the execution plan without running the query. By default returns a JSON string; pass result_format=ydb.QueryExplainResultFormat.DICT to get a parsed dict.

```python
# Returns a JSON string
plan = pool.explain_with_retries("SELECT * FROM users WHERE id = 1")
print(plan)

# Returns a parsed dict
plan_dict = pool.explain_with_retries(
    "SELECT * FROM users WHERE id = 1",
    result_format=ydb.QueryExplainResultFormat.DICT,
)
print(plan_dict["Plan"])
```

--------------------------------

### Create a synchronous topic writer with context manager

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/topic.rst

Recommended way to create a writer. The context manager ensures proper cleanup.

```python
with driver.topic_client.writer("/local/my-topic") as writer:
    writer.write("hello")
```

--------------------------------

### Implicit type mapping examples

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/types.rst

Shows how the SDK infers YDB types from Python values when passing parameters to queries. The first example passes a string 'Alice' which maps to Utf8, and the second passes a boolean True which maps to Bool.

```python
pool.execute_with_retries(
    "DECLARE $name AS Utf8; SELECT * FROM users WHERE name = $name",
    parameters={"$name": "Alice"},        # str → Utf8
)

pool.execute_with_retries(
    "DECLARE $active AS Bool; SELECT * FROM users WHERE active = $active",
    parameters={"$active": True},         # bool → Bool
)
```

--------------------------------

### Configure PartitioningSettings with min/max partitions and size/load flags

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/table.rst

Configures partitioning settings for a table, setting minimum and maximum partition counts, target partition size, and enabling both load-based and size-based partitioning. The `with_partition_size_mb` method triggers a split when a shard exceeds the target size, but only if size-based partitioning is enabled.

```python
ydb.PartitioningSettings() \
    .with_min_partitions_count(4) \
    .with_max_partitions_count(256) \
    .with_partition_size_mb(512) \
    .with_partitioning_by_load(ydb.FeatureFlag.ENABLED) \
    .with_partitioning_by_size(ydb.FeatureFlag.ENABLED)
```

--------------------------------

### Get YDB managed database endpoint and path

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/examples/README.md

Retrieves the endpoint and database path for a managed YDB instance using the yc CLI. The command `yc ydb db get --name ydb1` outputs the required data.

```bash
yc ydb db get --name ydb1
```

--------------------------------

### SchemeClient.async_make_directory

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/apireference.rst

Asynchronously creates a directory at the specified path.

```APIDOC
## async_make_directory(path, settings=None)

### Description
Asynchronously creates a directory at the specified path.

### Method
async_make_directory

### Parameters
- **path** (str) - Required - The path of the directory to create.
- **settings** (optional) - Optional - Additional settings for the operation.

### Returns
A future of the operation result.
```

--------------------------------

### Async Writer with Timeout

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/topic.rst

Example of using the async writer with a timeout to handle buffer full conditions.

```APIDOC
## Async Writer with Timeout

### Description
Demonstrates using the async writer with a timeout to avoid indefinite blocking.

### Method
N/A (SDK usage)

### Endpoint
N/A (SDK usage)

### Parameters
#### Constructor Parameters
- **max_buffer_size_bytes** (int) - Optional - Pause when this many bytes are in flight.
- **buffer_wait_timeout_sec** (float) - Optional - Timeout for `write()` when the buffer is full.

### Request Example
```python
writer = driver.topic_client.writer(
    "/local/my-topic",
    max_buffer_size_bytes=4 * 1024 * 1024,
    buffer_wait_timeout_sec=10.0,
)

try:
    await writer.write("message")
except ydb.TopicWriterBufferFullError:
    ...
```

### Response
#### Success Response
- **write()** - Suspends the coroutine until space is available or the timeout expires.

#### Error Handling
- **TopicWriterBufferFullError** - Raised when the timeout expires before space is freed.

### Notes
- To apply your own timeout without raising an error, wrap the call with `asyncio.wait_for`.
```

--------------------------------

### Describe a Topic (Basic and with Statistics)

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/topic.rst

Describes a topic and prints partition and consumer information. The first block shows a basic describe call. The second block includes partition-level statistics by setting include_stats=True.

```python
description = driver.topic_client.describe_topic("/local/my-topic")
print(description.partitions)
print(description.consumers)
```

```python
# Include partition-level statistics
description = driver.topic_client.describe_topic(
    "/local/my-topic",
    include_stats=True,
)
```

--------------------------------

### Result of SELECT COUNT(*)

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/quickstart.rst

Shows the result of the SELECT COUNT(*) query from the previous example, returning 1 row.

```pycon
>>> res[0].rows_count
1
```

--------------------------------

### Invoke time-series function with curl

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/examples/time-series-serverless/sample-commands.txt

Calls the deployed function with query parameters for start, end, interval, mean, and sigma.

```bash
curl "https://functions.yandexcloud.net/d4ek54ao5ouqq1tg5tbq?start=1615000000000&end=1615000010000&interval=1&mean=12.3&sigma=5"
```

--------------------------------

### describe_system_view(path: str, settings: BaseRequestSettings | None = None) → Any

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/apireference.rst

Returns a full description of a system view by the provided path.

```APIDOC
## describe_system_view(path: str, settings: BaseRequestSettings | None = None) → Any

### Description
Returns a full description of a system view by the provided path.

### Method
describe_system_view

### Parameters
- **path** (str) - Required - A system view path.
- **settings** (BaseRequestSettings | None) - Optional - A request settings.

### Returns
SystemViewSchemeEntry describing the system view.

### Return Type
Any
```

--------------------------------

### Fail fast on connection errors during startup

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/errors.rst

Use driver.wait with fail_fast=True to immediately raise TimeoutError if the connection cannot be established within the timeout. Catch TimeoutError and exit with a clear message.

```python
try:
    driver.wait(timeout=5, fail_fast=True)
except TimeoutError:
    raise SystemExit("Could not reach YDB — check endpoint and credentials")
```

--------------------------------

### DDL queries: CREATE TABLE and DROP TABLE

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/query.rst

Execute DDL statements using pool.execute_with_retries(). The example creates a table and then drops it.

```python
pool.execute_with_retries(
    """
    CREATE TABLE orders (
        order_id Uint64,
        user_id  Uint64,
        amount   Double,
        created  Timestamp,
        PRIMARY KEY (order_id)
    )
    """
)

pool.execute_with_retries("DROP TABLE IF EXISTS orders")
```

--------------------------------

### retry_tx_sync with explicit transaction mode

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/query.rst

Use this to specify an explicit transaction mode when calling retry_tx_sync. The example uses ydb.QuerySnapshotReadOnly() as the tx_mode.

```python
pool.retry_tx_sync(callee, tx_mode=ydb.QuerySnapshotReadOnly())
```

--------------------------------

### topic-create command

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/tests/slo/README.md

Creates a topic with a consumer in the database for the Topic SLO workload. Requires endpoint, database path, topic path, and consumer name.

```bash
python tests/slo/src/ topic-create localhost:2136 /local
--topic-path /local/slo_topic --topic-consumer slo_consumer
```

--------------------------------

### ydb.BulkUpsertColumns

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/apireference.rst

Represents a set of columns for bulk upsert operations. Provides a method to add a column and a property to get the protocol buffer representation.

```APIDOC
## ydb.BulkUpsertColumns

### Description
Represents a set of columns for bulk upsert operations.

### Methods
- **add_column(name: str, column_type: AbstractTypeBuilder | PrimitiveType) -> BulkUpsertColumns** - Adds a column with the given name and type.

### Properties
- **proto** (Type) - Returns the protocol buffer representation of the type.
```

--------------------------------

### Initialize ydb.Driver

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/quickstart.rst

Initializes a YDB driver with an endpoint and database, using credentials from environment variables. The driver waits up to 5 seconds and fails fast if connection fails.

```python
endpoint = "grpc://localhost:2136"  # your ydb endpoint
database = "/local"  # your ydb database

with ydb.Driver(
    endpoint=endpoint,
    database=database,
    credentials=ydb.credentials_from_env_variables(),
) as driver:
    driver.wait(timeout=5, fail_fast=True)
```

--------------------------------

### execute_with_retries with Arrow format

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/query.rst

Executes a query and returns results in Apache Arrow IPC format, which can be converted to pandas DataFrames. Requires pyarrow to be installed.

```APIDOC
## execute_with_retries (Arrow format)

### Description
Executes a query with retries and returns results in Apache Arrow IPC format. The results can be read into pandas DataFrames using pyarrow.

### Method
SDK method call

### Endpoint
`pool.execute_with_retries(query, result_set_format=ydb.QueryResultSetFormat.ARROW)`

### Parameters
#### Query Parameters
- **query** (string) - Required - The SQL query to execute.
- **result_set_format** (ydb.QueryResultSetFormat) - Required - Set to `ydb.QueryResultSetFormat.ARROW` to request Arrow format.
- **arrow_format_settings** (ydb.ArrowFormatSettings) - Optional - Settings for Arrow compression.

### Request Example
```python
result_sets = pool.execute_with_retries(
    "SELECT id, name, score FROM users ORDER BY id LIMIT 1000",
    result_set_format=ydb.QueryResultSetFormat.ARROW,
)
```

### Response
#### Success Response
- **result_sets** (list of ydb.ResultSet) - Each result set contains Arrow IPC data accessible via `arrow_format_meta.schema` and `data`.

#### Response Example
```python
for result_set in result_sets:
    schema = pa.ipc.read_schema(pa.py_buffer(result_set.arrow_format_meta.schema))
    batch  = pa.ipc.read_record_batch(pa.py_buffer(result_set.data), schema)
    df = batch.to_pandas()
    print(df.head())
```
```

--------------------------------

### ListType: List<Int64> parameter

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/types.rst

Creates a homogeneous ordered sequence type. The example declares a List<Int64> parameter and passes a list of integers.

```python
ydb.ListType(ydb.PrimitiveType.Int64)

pool.execute_with_retries(
    "DECLARE $ids AS List<Int64>; "
    "SELECT * FROM users WHERE id IN $ids",
    parameters={"$ids": ([1, 2, 3], ydb.ListType(ydb.PrimitiveType.Int64))},
)
```

--------------------------------

### Invoke function via CLI

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/examples/time-series-serverless/README.md

Invokes the serverless function with query parameters for start, end, interval, mean, and sigma. The function ID is required.

```shell
yc serverless function invoke <function-id> -d '{"queryStringParameters": {"start": "1615000000000", "end": "1615000010000", "interval": "1", "mean": "12.3", "sigma": "5"}}'
```

--------------------------------

### Inspect a semaphore with describe()

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/coordination.rst

Inspect a semaphore with describe() to get its limit, current count, owners, and waiters. Use this to monitor the semaphore's state.

```python
with driver.coordination_client.session("/local/my_node") as session:
    semaphore = session.semaphore("my_lock")
    info = semaphore.describe()

    print("limit:   ", info.limit)    # max count
    print("count:   ", info.count)    # currently acquired count
    print("owners:  ", info.owners)   # list of current holders
    print("waiters: ", info.waiters)  # list of processes waiting to acquire

```

--------------------------------

### Create Driver with keyword arguments

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/driver.rst

Alternative way to create a driver by passing endpoint, database, and credentials directly to ydb.Driver. Equivalent to using a DriverConfig object.

```python
driver = ydb.Driver(
    endpoint="grpc://localhost:2136",
    database="/local",
    credentials=ydb.credentials_from_env_variables(),
)
```

--------------------------------

### Async Driver usage with ydb.aio.Driver

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/driver.rst

Creates an async driver using ydb.aio.Driver with an endpoint, database, and credentials from environment variables. Waits for the driver to be ready with a 5-second timeout and fail_fast enabled before using it.

```python
import asyncio
import ydb

async def main():
    async with ydb.aio.Driver(
        endpoint="grpc://localhost:2136",
        database="/local",
        credentials=ydb.credentials_from_env_variables(),
    ) as driver:
        await driver.wait(timeout=5, fail_fast=True)
        # ... use driver

asyncio.run(main())
```

--------------------------------

### Get basic query stats with session.execute

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/query.rst

Pass stats_mode=ydb.QueryStatsMode.BASIC to session.execute to collect execution statistics. The iterator must be fully consumed for last_query_stats to be populated.

```python
def callee(session: ydb.QuerySession):
    with session.execute(
        "SELECT * FROM users",
        stats_mode=ydb.QueryStatsMode.BASIC,
    ):
        pass  # must iterate to completion for stats to be populated

    print(session.last_query_stats)

pool.retry_operation_sync(callee)
```

--------------------------------

### Compress Arrow results with ZSTD codec

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/query.rst

Pass arrow_format_settings with an ArrowCompressionCodec to compress Arrow results on the wire. The example uses ZSTD compression at level 10.

```python
settings = ydb.ArrowFormatSettings(
    compression_codec=ydb.ArrowCompressionCodec(
        ydb.ArrowCompressionCodecType.ZSTD,
        level=10,
    )
)

result_sets = pool.execute_with_retries(
    "SELECT * FROM events LIMIT 100000",
    result_set_format=ydb.QueryResultSetFormat.ARROW,
    arrow_format_settings=settings,
)
```

--------------------------------

### create_topic

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/topic.rst

Creates a topic with the specified configuration, including partition settings, retention policy, codecs, and consumers.

```APIDOC
## create_topic

### Description
Creates a topic with the specified configuration, including partition settings, retention policy, codecs, and consumers.

### Method
create_topic

### Parameters
- **path** (string) - Required - Full topic path including database prefix.
- **min_active_partitions** (int) - Optional - Minimum number of active partitions (default: 1).
- **max_active_partitions** (int) - Optional - Maximum number of partitions when auto-scaling is enabled.
- **retention_period** (datetime.timedelta) - Optional - How long messages are kept.
- **retention_storage_mb** (int) - Optional - Maximum storage size per partition in megabytes.
- **supported_codecs** (list of TopicCodec) - Optional - List of codecs the topic accepts (default: RAW and GZIP).
- **consumers** (list of str or TopicConsumer) - Optional - List of consumer names or TopicConsumer objects to create upfront.
- **partition_write_speed_bytes_per_second** (int) - Optional - Per-partition write throughput limit.

### Request Example
```python
# Synchronous
driver.topic_client.create_topic(
    "/local/my-topic",
    min_active_partitions=1,
    max_active_partitions=10,
    retention_period=datetime.timedelta(hours=24),
    consumers=["my-consumer"],
)

# Asynchronous
await driver.topic_client.create_topic(
    "/local/my-topic",
    min_active_partitions=1,
    max_active_partitions=10,
    retention_period=datetime.timedelta(hours=24),
    consumers=["my-consumer"],
)
```
```

--------------------------------

### Get synchronous topic client

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/topic.rst

Creates a synchronous YDB driver and accesses the topic_client property. Requires a running YDB endpoint and credentials from environment variables.

```python
import ydb

with ydb.Driver(
    endpoint="grpc://localhost:2136",
    database="/local",
    credentials=ydb.credentials_from_env_variables(),
) as driver:
    driver.wait(timeout=5, fail_fast=True)
    topic_client = driver.topic_client
```

--------------------------------

### DictType: Dict<Utf8, Int64> parameter

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/types.rst

Creates a map from a key type to a value type. The example shows a Dict<Utf8, Int64> parameter with a dictionary of scores.

```python
# Dict<Utf8, Int64>
ydb.DictType(ydb.PrimitiveType.Utf8, ydb.PrimitiveType.Int64)

parameters={
    "$scores": (
        {"alice": 10, "bob": 20},
        ydb.DictType(ydb.PrimitiveType.Utf8, ydb.PrimitiveType.Int64),
    )
}
```

--------------------------------

### Create an asynchronous topic writer with context manager

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/topic.rst

Async version of the writer creation. Use with async/await.

```python
async with driver.topic_client.writer("/local/my-topic") as writer:
    await writer.write("hello")
```

--------------------------------

### Create global async index with with_global_async_index

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/table.rst

Builds the index in the background so the table stays available while the index is being constructed.

```python
ydb.TableIndex("name_idx").with_global_async_index().with_index_columns("name")
```

--------------------------------

### Call function via CURL

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/examples/time-series-serverless/README.md

Calls the public function endpoint with query parameters for start, end, interval, mean, and sigma. The function must be made public first.

```shell
curl "https://functions.yandexcloud.net/<function-id>?start=1615000000000&end=1615000010000&interval=1&mean=12.3&sigma=5"
```

--------------------------------

### TableDescription.with_partitioning_settings

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/table.rst

Passes PartitioningSettings to a TableDescription to configure automatic partitioning for a table being created.

```APIDOC
## TableDescription.with_partitioning_settings

### Description
Sets the partitioning settings for a table description, controlling how the table is split into shards.

### Method
`with_partitioning_settings(settings)`

### Parameters
- **settings** (PartitioningSettings) - Required - The partitioning settings to apply.

### Returns
- `TableDescription` - The updated table description.

### Example
```python
driver.table_client.create_table(
    "/local/events",
    ydb.TableDescription()
    .with_columns(
        ydb.Column("id",   ydb.PrimitiveType.Uint64),
        ydb.Column("data", ydb.OptionalType(ydb.PrimitiveType.Utf8)),
    )
    .with_primary_keys("id")
    .with_partitioning_settings(
        ydb.PartitioningSettings()
        .with_min_partitions_count(4)
        .with_max_partitions_count(64)
        .with_partitioning_by_load(ydb.FeatureFlag.ENABLED)
    ),
)
```
```

--------------------------------

### Pre-split table at specific keys with ExplicitPartitions

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/table.rst

Creates a table with explicit pre-splitting at specific keys using `with_partition_at_keys` and `ydb.ExplicitPartitions`. Useful when the key distribution is known upfront.

```python
driver.table_client.create_table(
    "/local/events",
    ydb.TableDescription()
    .with_columns(
        ydb.Column("id",   ydb.PrimitiveType.Uint64),
        ydb.Column("data", ydb.OptionalType(ydb.PrimitiveType.Utf8)),
    )
    .with_primary_keys("id")
    .with_partition_at_keys(
        ydb.ExplicitPartitions(
            [ydb.SplitPoint(1000), ydb.SplitPoint(2000), ydb.SplitPoint(3000)]
        )
    ),
)
```

--------------------------------

### TupleType: Tuple<Double, Double> parameter

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/types.rst

Creates a positional (unnamed) record by chaining add_element. The example builds a point type with two Double elements and passes a list of coordinates.

```python
point_type = (
    ydb.TupleType()
    .add_element(ydb.PrimitiveType.Double)   # x
    .add_element(ydb.PrimitiveType.Double)   # y
)

parameters={"$point": ([1.0, 2.0], point_type)}
```

--------------------------------

### Create coordination node with default config

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/coordination.rst

Creates a coordination node with default configuration. The node must exist before sessions can attach.

```python
driver.coordination_client.create_node("/local/my_node")
```

--------------------------------

### Get asynchronous topic client

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/topic.rst

Creates an asynchronous YDB driver using ydb.aio.Driver and accesses the topic_client property. Requires asyncio and a running YDB endpoint with credentials from environment variables.

```python
import asyncio
import ydb

async def main():
    async with ydb.aio.Driver(
        endpoint="grpc://localhost:2136",
        database="/local",
        credentials=ydb.credentials_from_env_variables(),
    ) as driver:
        await driver.wait(timeout=5, fail_fast=True)
        topic_client = driver.topic_client
```

--------------------------------

### PartitioningSettings builder methods

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/table.rst

Methods to configure automatic partitioning behavior for a table, such as min/max partitions, target size, and auto-splitting by load or size.

```APIDOC
## PartitioningSettings Builder Methods

### Description
These methods configure how a table is split into shards and how that split evolves over time.

### Methods

#### with_min_partitions_count(n)
- **Parameters**: `n` (int) - Required - Minimum number of shards the table always has.
- **Returns**: `PartitioningSettings` - The updated settings object.

#### with_max_partitions_count(n)
- **Parameters**: `n` (int) - Required - Maximum number of shards the table never exceeds.
- **Returns**: `PartitioningSettings` - The updated settings object.

#### with_partition_size_mb(mb)
- **Parameters**: `mb` (int) - Required - Target shard size in megabytes; triggers a split when exceeded (requires by-size enabled).
- **Returns**: `PartitioningSettings` - The updated settings object.

#### with_partitioning_by_size(flag)
- **Parameters**: `flag` (FeatureFlag) - Required - Auto-split when a shard grows beyond the target size. Values: `ENABLED`, `DISABLED`, `UNSPECIFIED`.
- **Returns**: `PartitioningSettings` - The updated settings object.

#### with_partitioning_by_load(flag)
- **Parameters**: `flag` (FeatureFlag) - Required - Auto-split hot shards and merge idle ones based on load. Values: `ENABLED`, `DISABLED`, `UNSPECIFIED`.
- **Returns**: `PartitioningSettings` - The updated settings object.

### Example
```python
ydb.PartitioningSettings() \
    .with_min_partitions_count(4) \
    .with_max_partitions_count(256) \
    .with_partition_size_mb(512) \
    .with_partitioning_by_load(ydb.FeatureFlag.ENABLED) \
    .with_partitioning_by_size(ydb.FeatureFlag.ENABLED)
```
```

--------------------------------

### Create a directory with make_directory

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/scheme.rst

Creates a directory at the given path. Nested directories must be created one level at a time, as shown by the sequential calls.

```python
driver.scheme_client.make_directory("/local/my_app")

# Nested directories must be created one level at a time
driver.scheme_client.make_directory("/local/my_app/production")
driver.scheme_client.make_directory("/local/my_app/staging")
```

--------------------------------

### StructType: Struct<id:Uint64,name:Utf8> parameter

Source: https://github.com/ydb-platform/ydb-python-sdk/blob/main/docs/types.rst

Creates a named record type for passing rows as parameters. The example builds a struct with id and name members and uses it in an INSERT.

```python
row_type = (
    ydb.StructType()
    .add_member("id", ydb.PrimitiveType.Uint64)
    .add_member("name", ydb.PrimitiveType.Utf8)
)

pool.execute_with_retries(
    "DECLARE $row AS Struct<id:Uint64,name:Utf8>; "
    "INSERT INTO users (id, name) VALUES ($row.id, $row.name)",
    parameters={
        "$row": ({"id": 1, "name": "Alice"}, row_type)
    },
)
```
Tokens:
Raw