tessl/pypi-sqlalchemy

Python SQL toolkit and Object Relational Mapper providing full power and flexibility of SQL

Overview

Eval results

Files

Core Engine and Connectivity

Name: tessl/pypi-sqlalchemy
Author: tessl

Database engine creation, connection management, URL handling, connection pooling, and transaction management. The Core Engine provides the foundation for all SQLAlchemy database operations.

Capabilities

Engine Creation

Create database engines for connection management and SQL execution.

def create_engine(url, **kwargs):
    """
    Create a database engine.
    
    Parameters:
    - url: str or URL, database connection URL
    - echo: bool, log all SQL statements (default False)
    - pool_size: int, connection pool size (default 5)
    - max_overflow: int, maximum pool overflow (default 10)
    - pool_timeout: int, connection timeout in seconds (default 30)
    - pool_recycle: int, connection recycle time (default -1)
    - isolation_level: str, transaction isolation level
    
    Returns:
    Engine: Database engine instance
    """

def create_mock_engine(url, executor, **kwargs):
    """
    Create a mock engine for testing SQL generation.
    
    Parameters:
    - url: str or URL, database URL for dialect selection
    - executor: callable, function to handle SQL execution
    
    Returns:
    Engine: Mock engine that captures SQL without database connection
    """

def engine_from_config(configuration, prefix='sqlalchemy.', **kwargs):
    """
    Create engine from configuration dictionary.
    
    Parameters:
    - configuration: dict, configuration parameters
    - prefix: str, prefix for configuration keys
    
    Returns:
    Engine: Configured database engine
    """

URL Construction and Parsing

Handle database connection URLs with parsing and construction utilities.

def make_url(name_or_url):
    """
    Create URL object from string or existing URL.
    
    Parameters:
    - name_or_url: str or URL, connection specification
    
    Returns:
    URL: Parsed URL object
    """

class URL:
    """Database connection URL representation."""
    
    def __init__(self, drivername, username=None, password=None, 
                 host=None, port=None, database=None, query=None):
        """
        Create URL object.
        
        Parameters:
        - drivername: str, database driver name
        - username: str, database username
        - password: str, database password  
        - host: str, database host
        - port: int, database port
        - database: str, database name
        - query: dict, additional query parameters
        """
    
    def render_as_string(self, hide_password=True):
        """
        Render URL as connection string.
        
        Parameters:
        - hide_password: bool, mask password in output
        
        Returns:
        str: Connection string representation
        """

Engine Operations

Core engine functionality for connection management and SQL execution.

class Engine:
    """Database engine providing connection management and execution."""
    
    def connect(self):
        """
        Create new database connection.
        
        Returns:
        Connection: New database connection
        """
    
    def execute(self, statement, parameters=None):
        """
        Execute SQL statement with automatic connection management.
        
        Parameters:
        - statement: str or executable, SQL statement
        - parameters: dict or sequence, bound parameters
        
        Returns:
        Result: Query results
        """
    
    def begin(self):
        """
        Begin transaction with automatic connection management.
        
        Returns:
        Transaction: Transaction context manager
        """
    
    def dispose(self):
        """Close all connections and dispose of connection pool."""
    
    @property
    def dialect(self):
        """Database dialect for this engine."""
    
    @property
    def pool(self):
        """Connection pool for this engine."""

Connection Management

Direct connection handling with transaction support.

class Connection:
    """Database connection with transaction support."""
    
    def execute(self, statement, parameters=None):
        """
        Execute SQL statement on this connection.
        
        Parameters:
        - statement: str or executable, SQL statement
        - parameters: dict or sequence, bound parameters
        
        Returns:
        Result: Query results
        """
    
    def begin(self):
        """
        Begin transaction on this connection.
        
        Returns:
        Transaction: Transaction object
        """
    
    def commit(self):
        """Commit current transaction."""
    
    def rollback(self):
        """Rollback current transaction."""
    
    def close(self):
        """Close this connection."""
    
    def scalar(self, statement, parameters=None):
        """
        Execute statement and return scalar result.
        
        Parameters:
        - statement: str or executable, SQL statement
        - parameters: dict or sequence, bound parameters
        
        Returns:
        Any: Single scalar value
        """

Transaction Management

Transaction handling with context manager support and savepoints.

class Transaction:
    """Database transaction with rollback support."""
    
    def commit(self):
        """Commit this transaction."""
    
    def rollback(self):
        """Rollback this transaction."""
    
    def close(self):
        """Close transaction (rollback if not committed)."""

class NestedTransaction(Transaction):
    """Nested transaction using savepoints."""
    
    def rollback(self):
        """Rollback to savepoint."""

class TwoPhaseTransaction(Transaction):
    """Two-phase commit transaction."""
    
    def prepare(self):
        """Prepare transaction for two-phase commit."""

Result Handling

Query result processing with multiple access patterns.

class Result:
    """Query result with row iteration and data access."""
    
    def fetchone(self):
        """
        Fetch next row.
        
        Returns:
        Row or None: Next row or None if no more rows
        """
    
    def fetchmany(self, size=None):
        """
        Fetch multiple rows.
        
        Parameters:
        - size: int, number of rows to fetch
        
        Returns:
        List[Row]: List of rows
        """
    
    def fetchall(self):
        """
        Fetch all remaining rows.
        
        Returns:
        List[Row]: All remaining rows
        """
    
    def scalar(self):
        """
        Fetch scalar value from first column of first row.
        
        Returns:
        Any: Scalar value or None
        """
    
    def mappings(self):
        """
        Return result as mapping-like objects.
        
        Returns:
        MappingResult: Result with dict-like row access
        """

class Row:
    """Individual result row with column access."""
    
    def __getitem__(self, key):
        """Get column value by index or name."""
    
    def __getattr__(self, name):
        """Get column value by attribute access."""
    
    def _asdict(self):
        """
        Convert row to dictionary.
        
        Returns:
        dict: Row data as dictionary
        """

Connection Pooling

Connection pool management with various pooling strategies.

class Pool:
    """Base connection pool class."""
    
    def connect(self):
        """Get connection from pool."""
    
    def dispose(self):
        """Dispose all connections in pool."""

class QueuePool(Pool):
    """Thread-safe connection pool with overflow and timeout."""
    
    def __init__(self, creator, pool_size=5, max_overflow=10, 
                 timeout=30, **kwargs):
        """
        Create queue-based connection pool.
        
        Parameters:
        - creator: callable, function to create connections
        - pool_size: int, base pool size
        - max_overflow: int, maximum overflow connections
        - timeout: int, connection timeout in seconds
        """

class StaticPool(Pool):
    """Single connection pool for lightweight scenarios."""

class NullPool(Pool):
    """No pooling - create new connection for each request."""

class SingletonThreadPool(Pool):
    """One connection per thread pool."""

Database Inspection

Database schema inspection and reflection capabilities.

def inspect(subject):
    """
    Inspect database engine, connection, or mapped object.
    
    Parameters:
    - subject: Engine, Connection, or mapped class
    
    Returns:
    Inspector: Appropriate inspector object
    """

class Inspector:
    """Database schema inspector."""
    
    def get_schema_names(self):
        """
        Get list of schema names.
        
        Returns:
        List[str]: Available schema names
        """
    
    def get_table_names(self, schema=None):
        """
        Get list of table names in schema.
        
        Parameters:
        - schema: str, schema name (default schema if None)
        
        Returns:
        List[str]: Table names in schema
        """
    
    def get_columns(self, table_name, schema=None):
        """
        Get column information for table.
        
        Parameters:
        - table_name: str, table name
        - schema: str, schema name
        
        Returns:
        List[dict]: Column information dictionaries
        """
    
    def get_primary_keys(self, table_name, schema=None):
        """
        Get primary key columns for table.
        
        Parameters:
        - table_name: str, table name
        - schema: str, schema name
        
        Returns:
        List[str]: Primary key column names
        """
    
    def get_foreign_keys(self, table_name, schema=None):
        """
        Get foreign key constraints for table.
        
        Parameters:
        - table_name: str, table name
        - schema: str, schema name
        
        Returns:
        List[dict]: Foreign key constraint information
        """

Usage Examples

Basic Engine Usage

from sqlalchemy import create_engine, text

# Create engine
engine = create_engine("postgresql://user:pass@localhost/dbname")

# Execute simple query
with engine.connect() as conn:
    result = conn.execute(text("SELECT version()"))
    print(result.scalar())

# Transaction usage
with engine.begin() as conn:
    conn.execute(text("INSERT INTO users (name) VALUES ('John')"))
    # Automatically committed

Connection Pool Configuration

from sqlalchemy import create_engine

engine = create_engine(
    "postgresql://user:pass@localhost/dbname",
    pool_size=20,        # Base pool size
    max_overflow=0,      # No overflow connections
    pool_timeout=30,     # 30 second timeout
    pool_recycle=3600    # Recycle connections after 1 hour
)

Database Inspection

from sqlalchemy import create_engine, inspect

engine = create_engine("sqlite:///example.db")
inspector = inspect(engine)

# Get table information
tables = inspector.get_table_names()
for table in tables:
    columns = inspector.get_columns(table)
    print(f"Table {table}: {[col['name'] for col in columns]}")

Install with Tessl CLI