tessl/pypi-ydb

Officially supported Python client for YDB distributed SQL database

Overview

Eval results

Files

Authentication and Credentials

Name: tessl/pypi-ydb
Author: tessl

Complete authentication system supporting anonymous access, static tokens, OAuth2 token exchange, JWT authentication, service account credentials, and Yandex Cloud IAM integration.

Capabilities

Base Credentials Classes

Abstract base classes that define the credential interface.

class AbstractCredentials:
    """Abstract base class for auth metadata providers."""

class Credentials:
    def __init__(self, tracer=None):
        """
        Base credentials class.
        
        Args:
            tracer: Optional tracer for debugging
        """

    def auth_metadata(self):
        """
        Get authentication metadata.
        
        Returns:
            Iterable of (header, value) tuples for gRPC metadata
        """

    def get_auth_token(self) -> str:
        """
        Extract auth token from metadata.
        
        Returns:
            str: Authentication token or empty string
        """

Anonymous Credentials

No authentication - for local development and testing.

class AnonymousCredentials(Credentials):
    """Credentials for anonymous access (no authentication)."""
    
    def __init__(self):
        """Create anonymous credentials."""

Static Token Credentials

Fixed token authentication for simple authentication scenarios.

class StaticCredentials(Credentials):
    def __init__(self, token: str):
        """
        Static token credentials.
        
        Args:
            token (str): Authentication token
        """

OAuth2 Token Exchange Credentials

OAuth2 token exchange authentication for secure token-based access.

class OAuth2TokenExchangeCredentials(Credentials):
    def __init__(
        self,
        token_endpoint: str,
        resource: str = None,
        audience: str = None,
        scope: str = None,
        requested_token_type: str = None,
        subject_token: str = None,
        subject_token_type: str = None,
        actor_token: str = None,
        actor_token_type: str = None,
        **kwargs
    ):
        """
        OAuth2 token exchange credentials.
        
        Args:
            token_endpoint (str): OAuth2 token endpoint URL
            resource (str, optional): Target resource
            audience (str, optional): Token audience
            scope (str, optional): Requested scope
            requested_token_type (str, optional): Type of requested token
            subject_token (str, optional): Subject token for exchange
            subject_token_type (str, optional): Subject token type
            actor_token (str, optional): Actor token
            actor_token_type (str, optional): Actor token type
        """

JWT Credentials

JSON Web Token authentication for service-to-service communication.

class JWTCredentials(Credentials):
    def __init__(
        self,
        jwt_token: str,
        refresh_token: str = None,
        token_endpoint: str = None
    ):
        """
        JWT token credentials.
        
        Args:
            jwt_token (str): JWT token
            refresh_token (str, optional): Refresh token for renewal
            token_endpoint (str, optional): Token refresh endpoint
        """

Service Account Credentials

Direct service account authentication using key files or metadata.

class DirectServiceAccountCredentials(Credentials):
    def __init__(
        self,
        service_account_id: str,
        key_id: str,
        private_key: str,
        token_endpoint: str = None
    ):
        """
        Direct service account credentials.
        
        Args:
            service_account_id (str): Service account identifier
            key_id (str): Private key identifier
            private_key (str): RSA private key in PEM format
            token_endpoint (str, optional): Token endpoint URL
        """

Refreshable Credentials

Auto-refreshing credentials wrapper for long-running applications.

class RefreshableCredentials(Credentials):
    def __init__(
        self,
        credentials_factory: callable,
        refresh_threshold: float = 300
    ):
        """
        Auto-refreshing credentials wrapper.
        
        Args:
            credentials_factory (callable): Function returning fresh credentials
            refresh_threshold (float): Refresh before expiry (seconds)
        """

Credential Factory Functions

Convenience functions for creating credentials from various sources.

def default_credentials(credentials: Credentials = None, tracer=None) -> Credentials:
    """
    Get default credentials or return anonymous if none provided.
    
    Args:
        credentials (Credentials, optional): Explicit credentials
        tracer: Optional tracer
        
    Returns:
        Credentials: Default or provided credentials
    """

def credentials_from_env_variables(tracer=None) -> Credentials:
    """
    Create credentials from environment variables.
    
    Environment variables checked:
    - YDB_SERVICE_ACCOUNT_KEY_FILE_CREDENTIALS: Path to service account key file
    - YDB_ANONYMOUS_CREDENTIALS: "1" for anonymous access
    - YDB_METADATA_CREDENTIALS: "1" for metadata service
    - YDB_ACCESS_TOKEN_CREDENTIALS: Direct access token
    
    Args:
        tracer: Optional tracer
        
    Returns:
        Credentials: Credentials based on environment variables
    """

def construct_credentials_from_string(credentials_string: str) -> Credentials:
    """
    Parse credentials from connection string format.
    
    Args:
        credentials_string (str): Credentials specification
        
    Returns:
        Credentials: Parsed credentials object
    """

Yandex Cloud IAM Integration

import ydb.iam

class ServiceAccountCredentials(Credentials):
    @classmethod
    def from_file(cls, key_file: str, **kwargs) -> 'ServiceAccountCredentials':
        """
        Create credentials from service account key file.
        
        Args:
            key_file (str): Path to JSON key file
            
        Returns:
            ServiceAccountCredentials: Configured credentials
        """

class MetadataCredentials(Credentials):
    def __init__(self, metadata_url: str = None):
        """
        Credentials from Yandex Cloud metadata service.
        
        Args:
            metadata_url (str, optional): Custom metadata service URL
        """

OAuth2 Token Exchange Module

import ydb.oauth2_token_exchange

class TokenExchangeCredentials(Credentials):
    def __init__(
        self,
        token_endpoint: str,
        client_id: str,
        client_secret: str,
        **kwargs
    ):
        """
        Token exchange credentials with client authentication.
        
        Args:
            token_endpoint (str): OAuth2 token endpoint
            client_id (str): OAuth2 client ID
            client_secret (str): OAuth2 client secret
        """

Usage Examples

Anonymous Access (Development)

import ydb

# No authentication - for local development
credentials = ydb.AnonymousCredentials()

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

Static Token Authentication

import ydb

# Using static token
credentials = ydb.StaticCredentials("your-access-token")

driver = ydb.Driver(
    endpoint="grpcs://ydb.example.com:2135",
    database="/production/db",
    credentials=credentials
)

Environment-Based Credentials

import os
import ydb

# Set environment variable
os.environ["YDB_ACCESS_TOKEN_CREDENTIALS"] = "your-token"

# Automatically detect credentials from environment
credentials = ydb.credentials_from_env_variables()

driver = ydb.Driver(
    endpoint="grpcs://ydb.example.com:2135",
    database="/production/db",
    credentials=credentials
)

Yandex Cloud Service Account

import ydb
import ydb.iam

# From service account key file
credentials = ydb.iam.ServiceAccountCredentials.from_file(
    "service-account-key.json"
)

driver = ydb.Driver(
    endpoint="grpcs://ydb.serverless.yandexcloud.net:2135",
    database="/ru-central1/b1g..../etn....",
    credentials=credentials
)

Metadata Service Credentials

import ydb
import ydb.iam

# From Yandex Cloud metadata service (inside YC VM)
credentials = ydb.iam.MetadataCredentials()

driver = ydb.Driver(
    endpoint="grpcs://ydb.serverless.yandexcloud.net:2135",
    database="/ru-central1/b1g..../etn....",
    credentials=credentials
)

OAuth2 Token Exchange

import ydb

# OAuth2 token exchange
credentials = ydb.OAuth2TokenExchangeCredentials(
    token_endpoint="https://oauth.example.com/token",
    resource="https://ydb.example.com",
    audience="ydb-api",
    subject_token="initial-token",
    subject_token_type="urn:ietf:params:oauth:token-type:access_token"
)

driver = ydb.Driver(
    endpoint="grpcs://ydb.example.com:2135",
    database="/production/db",
    credentials=credentials
)

Auto-Refreshing Credentials

import ydb

def create_fresh_credentials():
    # Fetch new token from your auth system
    token = fetch_token_from_auth_system()
    return ydb.StaticCredentials(token)

# Auto-refresh 5 minutes before expiry
credentials = ydb.RefreshableCredentials(
    credentials_factory=create_fresh_credentials,
    refresh_threshold=300
)

driver = ydb.Driver(
    endpoint="grpcs://ydb.example.com:2135",
    database="/production/db",
    credentials=credentials
)

Types

# Exception types for credential operations
class CredentialsTimeout(Exception):
    """Raised when credential operation times out."""

# Type aliases
Token = str
Endpoint = str
KeyFile = str

Install with Tessl CLI