tessl/pypi-snowflake-sqlalchemy

Snowflake SQLAlchemy Dialect providing comprehensive database connectivity and ORM support for Snowflake cloud data warehouse

—

Pending

Overview

Eval results

Files

Connection and Configuration

Name: tessl/pypi-snowflake-sqlalchemy
Author: tessl

Connection management, URL building utilities, and configuration options for optimizing Snowflake database connectivity within SQLAlchemy applications.

Capabilities

Connection URL Building

Utilities for constructing proper Snowflake connection strings and URLs.

from snowflake.sqlalchemy import URL
from snowflake.sqlalchemy.util import parse_url_boolean, parse_url_integer

def URL(**db_parameters):
    """
    Compose SQLAlchemy connection string for Snowflake.
    
    Args:
        **db_parameters: Database connection parameters
    
    Returns:
        str: SQLAlchemy connection URL
    """

def parse_url_boolean(value: str) -> bool:
    """
    Parse URL boolean values from connection strings.
    
    Args:
        value: String value to parse
    
    Returns:
        bool: Parsed boolean value
    """

def parse_url_integer(value: str) -> int:
    """
    Parse URL integer values from connection strings.
    
    Args:
        value: String value to parse
    
    Returns:
        int: Parsed integer value
    """

Connection Management

Internal utilities for managing connections and application settings.

from snowflake.sqlalchemy.util import (
    _update_connection_application_name,
    _set_connection_interpolate_empty_sequences
)

def _update_connection_application_name(**conn_kwargs):
    """
    Update connection application name for telemetry.
    
    Args:
        **conn_kwargs: Connection keyword arguments
    
    Returns:
        dict: Updated connection arguments
    """

def _set_connection_interpolate_empty_sequences(connection, flag: bool):
    """
    Set connection interpolation flag for empty sequences.
    
    Args:
        connection: Database connection
        flag: Whether to interpolate empty sequences
    """

Exception Classes

Comprehensive exception hierarchy for connection and configuration errors.

from sqlalchemy.exc import ArgumentError
from snowflake.sqlalchemy.exc import (
    NoPrimaryKeyError,
    UnsupportedPrimaryKeysAndForeignKeysError,
    RequiredParametersNotProvidedError,
    UnexpectedTableOptionKeyError,
    OptionKeyNotProvidedError,
    UnexpectedOptionParameterTypeError,
    CustomOptionsAreOnlySupportedOnSnowflakeTables,
    UnexpectedOptionTypeError,
    InvalidTableParameterTypeError,
    MultipleErrors,
    StructuredTypeNotSupportedInTableColumnsError
)

class NoPrimaryKeyError(ArgumentError):
    """Raised when primary key is required but missing."""
    
    def __init__(self, target: str):
        """
        Initialize with target table name.
        
        Args:
            target: Target table name
        """

class UnsupportedPrimaryKeysAndForeignKeysError(ArgumentError):
    """Raised when primary keys or foreign keys are not supported."""
    
    def __init__(self, target: str):
        """
        Initialize with target table name.
        
        Args:
            target: Target table name
        """

class RequiredParametersNotProvidedError(ArgumentError):
    """Raised when required parameters are not provided."""
    
    def __init__(self, target: str, parameters: List[str]):
        """
        Initialize with target and missing parameters.
        
        Args:
            target: Target object name
            parameters: List of missing parameter names
        """

class UnexpectedTableOptionKeyError(ArgumentError):
    """Raised when an unexpected table option key is used."""
    
    def __init__(self, expected: str, actual: str):
        """
        Initialize with expected and actual option keys.
        
        Args:
            expected: Expected option key
            actual: Actual option key received
        """

class OptionKeyNotProvidedError(ArgumentError):
    """Raised when an option key is not provided."""
    
    def __init__(self, target: str):
        """
        Initialize with target name.
        
        Args:
            target: Target object name
        """

class UnexpectedOptionParameterTypeError(ArgumentError):
    """Raised when an option parameter has wrong type."""
    
    def __init__(self, parameter_name: str, target: str, types: List[str]):
        """
        Initialize with parameter details.
        
        Args:
            parameter_name: Parameter name
            target: Target object name
            types: List of expected types
        """

class CustomOptionsAreOnlySupportedOnSnowflakeTables(ArgumentError):
    """Raised when custom options are used on non-Snowflake tables."""

class UnexpectedOptionTypeError(ArgumentError):
    """Raised when an unexpected option type is encountered."""
    
    def __init__(self, options: List[str]):
        """
        Initialize with unsupported options.
        
        Args:
            options: List of unsupported option names
        """

class InvalidTableParameterTypeError(ArgumentError):
    """Raised when table parameter has invalid type."""
    
    def __init__(self, name: str, input_type: str, expected_types: List[str]):
        """
        Initialize with parameter type details.
        
        Args:
            name: Parameter name
            input_type: Actual input type
            expected_types: List of expected types
        """

class MultipleErrors(ArgumentError):
    """Container for multiple errors."""
    
    def __init__(self, errors: List[Exception]):
        """
        Initialize with multiple errors.
        
        Args:
            errors: List of exceptions
        """

class StructuredTypeNotSupportedInTableColumnsError(ArgumentError):
    """Raised when structured types are used inappropriately."""
    
    def __init__(self, table_type: str, table_name: str, column_name: str):
        """
        Initialize with table and column details.
        
        Args:
            table_type: Type of table
            table_name: Name of table
            column_name: Name of column with structured type
        """

Version Information

Package version access and metadata.

from snowflake.sqlalchemy.version import VERSION
import snowflake.sqlalchemy

VERSION: str = "1.7.6"
__version__ = snowflake.sqlalchemy.__version__

Usage Examples

Basic Connection

from sqlalchemy import create_engine
from snowflake.sqlalchemy import URL

# Direct connection string
engine = create_engine(
    'snowflake://user:password@account_identifier/database/schema'
)

# Using URL builder
connection_params = {
    'user': 'myuser',
    'password': 'mypassword',
    'account': 'myaccount',
    'database': 'mydatabase',
    'schema': 'myschema',
    'warehouse': 'mywarehouse',
    'role': 'myrole'
}

engine = create_engine(URL(**connection_params))

Advanced Connection Configuration

from sqlalchemy import create_engine

# Connection with additional parameters
engine = create_engine(
    'snowflake://user:password@account.region.cloud/db/schema',
    connect_args={
        'warehouse': 'COMPUTE_WH',
        'role': 'ANALYST_ROLE',
        'client_session_keep_alive': True,
        'autocommit': False,
        'numpy': True
    }
)

Connection with Authentication Options

# Key-pair authentication
engine = create_engine(
    'snowflake://user@account/database/schema',
    connect_args={
        'private_key_path': '/path/to/private_key.p8',
        'private_key_passphrase': 'key_passphrase'
    }
)

# SSO authentication
engine = create_engine(
    'snowflake://user@account/database/schema',
    connect_args={
        'authenticator': 'externalbrowser'
    }
)

# OAuth authentication
engine = create_engine(
    'snowflake://user@account/database/schema',
    connect_args={
        'authenticator': 'oauth',
        'token': 'oauth_access_token'
    }
)

URL Parameter Parsing

from snowflake.sqlalchemy.util import parse_url_boolean, parse_url_integer

# Parse boolean parameters from connection string
autocommit = parse_url_boolean('true')  # True
numpy_support = parse_url_boolean('false')  # False

# Parse integer parameters
login_timeout = parse_url_integer('30')  # 30
network_timeout = parse_url_integer('300')  # 300

Connection Pooling Configuration

from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool

# Engine with connection pooling
engine = create_engine(
    'snowflake://user:password@account/db/schema',
    poolclass=QueuePool,
    pool_size=5,
    max_overflow=10,
    pool_pre_ping=True,
    pool_recycle=3600
)

Error Handling

from sqlalchemy import create_engine
from snowflake.sqlalchemy.exc import (
    RequiredParametersNotProvidedError,
    InvalidTableParameterTypeError
)

try:
    # Connection attempt
    engine = create_engine('snowflake://incomplete_url')
    
    # Table creation with validation
    from snowflake.sqlalchemy import HybridTable
    
    # This would raise NoPrimaryKeyError for HybridTable
    table = HybridTable('test', metadata, Column('id', Integer))
    
except RequiredParametersNotProvidedError as e:
    print(f"Missing required parameters: {e}")
except InvalidTableParameterTypeError as e:
    print(f"Invalid table parameter: {e}")

Application Name and Telemetry

from snowflake.sqlalchemy.util import _update_connection_application_name

# Update connection for custom application
conn_params = {
    'user': 'myuser',
    'password': 'mypassword',
    'account': 'myaccount'
}

# Add application name for telemetry
updated_params = _update_connection_application_name(
    application='MyDataApp',
    version='1.0.0',
    **conn_params
)

engine = create_engine(URL(**updated_params))

Connection Testing and Validation

from sqlalchemy import create_engine, text

def test_snowflake_connection(connection_string):
    """Test Snowflake connection and return version."""
    try:
        engine = create_engine(connection_string)
        with engine.connect() as conn:
            result = conn.execute(text('SELECT CURRENT_VERSION()')).fetchone()
            print(f"Connected to Snowflake version: {result[0]}")
            return True
    except Exception as e:
        print(f"Connection failed: {e}")
        return False
    finally:
        engine.dispose()

# Test connection
success = test_snowflake_connection(
    'snowflake://user:password@account/database/schema'
)

Version Information

import snowflake.sqlalchemy
from snowflake.sqlalchemy.version import VERSION

# Get package version
print(f"Snowflake SQLAlchemy version: {snowflake.sqlalchemy.__version__}")
print(f"Version constant: {VERSION}")

# Version-dependent feature checks
if tuple(map(int, snowflake.sqlalchemy.__version__.split('.'))) >= (1, 7, 0):
    print("Dynamic tables supported")

Install with Tessl CLI

npx tessl i tessl/pypi-snowflake-sqlalchemy

docs

tessl/pypi-snowflake-sqlalchemy

connection-config.md.css-3qkkll{font-size:var(--chakra-font-sizes-sm);font-weight:var(--chakra-font-weights-normal);color:var(--chakra-colors-gray-300);}docs/

Connection and Configuration

Capabilities

Connection URL Building

Connection Management

Exception Classes

Version Information

Usage Examples

Basic Connection

Advanced Connection Configuration

Connection with Authentication Options

URL Parameter Parsing

Connection Pooling Configuration

Error Handling

Application Name and Telemetry

Connection Testing and Validation

Version Information

connection-config.mddocs/