tessl/pypi-ydb

Officially supported Python client for YDB distributed SQL database

Overview

Eval results

Files

Driver and Connection Management

Name: tessl/pypi-ydb
Author: tessl

Core connection functionality for establishing and managing connections to YDB clusters, including driver configuration, endpoint discovery, credential management, and connection lifecycle.

Capabilities

Driver Creation and Configuration

The Driver is the main entry point for connecting to YDB. It manages connection pooling, endpoint discovery, and provides the foundation for all database operations.

class Driver:
    def __init__(
        self,
        endpoint: str,
        database: str,
        credentials: Credentials = None,
        compression: int = None,
        root_certificates: bytes = None,
        **kwargs
    ):
        """
        Create a YDB driver instance.
        
        Args:
            endpoint (str): YDB endpoint URL (e.g., "grpc://localhost:2136")
            database (str): Database path (e.g., "/local")
            credentials (Credentials, optional): Authentication credentials
            compression (int, optional): gRPC compression method
            root_certificates (bytes, optional): TLS root certificates
        """

    def wait(self, fail_fast: bool = True, timeout: float = None) -> bool:
        """
        Wait for driver to become ready.
        
        Args:
            fail_fast (bool): Raise exception on first error
            timeout (float, optional): Timeout in seconds
            
        Returns:
            bool: True if driver is ready, False on timeout
        """

    def stop(self, timeout: float = None):
        """
        Stop the driver and close connections.
        
        Args:
            timeout (float, optional): Shutdown timeout in seconds
        """

    @property
    def discovery_debug_details(self) -> str:
        """Get endpoint discovery debug information."""

Driver Configuration Object

Provides structured configuration for driver creation with validation and default values.

class DriverConfig:
    def __init__(
        self,
        endpoint: str,
        database: str,
        credentials: Credentials = None,
        compression: int = None,
        root_certificates: bytes = None,
        **kwargs
    ):
        """
        Driver configuration container.
        
        Args:
            endpoint (str): YDB endpoint URL
            database (str): Database path
            credentials (Credentials, optional): Authentication credentials
            compression (int, optional): gRPC compression method
            root_certificates (bytes, optional): TLS root certificates
        """

Driver Factory Functions

Convenience functions for creating drivers from various sources.

def construct_driver(config: DriverConfig) -> Driver:
    """
    Create driver from configuration object.
    
    Args:
        config (DriverConfig): Driver configuration
        
    Returns:
        Driver: Configured YDB driver
    """

def construct_driver_from_string(connection_string: str, **kwargs) -> Driver:
    """
    Create driver from connection string.
    
    Args:
        connection_string (str): YDB connection string
        **kwargs: Additional driver parameters
        
    Returns:
        Driver: Configured YDB driver
    """

RPC Compression Options

Configure gRPC compression for network efficiency.

class RPCCompression:
    """gRPC compression methods."""
    NoCompression: int  # No compression
    Deflate: int       # Deflate compression
    Gzip: int          # Gzip compression

Usage Examples

Basic Driver Setup

import ydb

# Create driver with minimal configuration
driver = ydb.Driver(
    endpoint="grpc://localhost:2136",
    database="/local",
    credentials=ydb.AnonymousCredentials()
)

# Wait for driver to be ready
if not driver.wait(timeout=10):
    raise RuntimeError("Driver failed to initialize")

# Use driver for operations...

# Clean shutdown
driver.stop()

Advanced Driver Configuration

import ydb

# Configure driver with custom settings
config = ydb.DriverConfig(
    endpoint="grpcs://ydb.example.com:2135",
    database="/production/mydb",
    credentials=ydb.StaticCredentials("your-token"),
    compression=ydb.RPCCompression.Gzip,
    root_certificates=open("ca.pem", "rb").read()
)

driver = ydb.construct_driver(config)
driver.wait(fail_fast=True, timeout=30)

# Driver is ready for use
print(f"Connected to: {driver.discovery_debug_details}")

Connection String Usage

import ydb

# Create driver from connection string
connection_string = "grpc://user:pass@localhost:2136/local?compression=gzip"
driver = ydb.construct_driver_from_string(connection_string)

driver.wait(fail_fast=True)
# Driver ready for operations

Context Manager Pattern

import ydb
from contextlib import closing

# Ensure proper cleanup
with closing(ydb.Driver(
    endpoint="grpc://localhost:2136",
    database="/local",
    credentials=ydb.AnonymousCredentials()
)) as driver:
    driver.wait(fail_fast=True, timeout=10)
    
    # Use driver for operations
    session_pool = ydb.SessionPool(driver)
    # ... perform operations ...
    session_pool.stop()

# Driver automatically stopped

Types

from typing import Optional, Union

# Type aliases used in driver operations
ConnectionString = str
Endpoint = str
DatabasePath = str
Timeout = Optional[float]

Install with Tessl CLI