tessl/pypi-sqlalchemy

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

Overview

Eval results

Files

Database Dialects

Name: tessl/pypi-sqlalchemy
Author: tessl

Database-specific implementations for PostgreSQL, MySQL, SQLite, Oracle, SQL Server with specialized types, functions, and features for each database. Dialects provide database-specific SQL generation and type handling.

Capabilities

PostgreSQL Dialect

Advanced PostgreSQL-specific types, operators, and features.

# Import PostgreSQL-specific types
from sqlalchemy.dialects.postgresql import ARRAY, HSTORE, JSON, JSONB, UUID
from sqlalchemy.dialects.postgresql import INET, CIDR, MACADDR
from sqlalchemy.dialects.postgresql import INT4RANGE, INT8RANGE, NUMRANGE, DATERANGE
from sqlalchemy.dialects.postgresql import TSVECTOR, TSQUERY
from sqlalchemy.dialects.postgresql import ExcludeConstraint

class ARRAY:
    """PostgreSQL array type with indexing and operators."""
    
    def __init__(self, item_type, as_tuple=False, dimensions=None, zero_indexes=False):
        """
        Create PostgreSQL array type.
        
        Parameters:
        - item_type: element type for array
        - as_tuple: bool, return tuples instead of lists
        - dimensions: int, number of array dimensions
        - zero_indexes: bool, use zero-based indexing
        """

class HSTORE:
    """PostgreSQL key-value store type."""
    
    def __init__(self, text_type=None):
        """
        Create HSTORE type.
        
        Parameters:
        - text_type: underlying text type for keys/values
        """

class JSON:
    """PostgreSQL JSON type with operators."""
    
    def __init__(self, none_as_null=False, astext_type=None):
        """
        Create JSON type.
        
        Parameters:
        - none_as_null: bool, store None as SQL NULL
        - astext_type: type for ->> text extraction
        """

class JSONB:
    """PostgreSQL binary JSON type with indexing."""
    
    def __init__(self, none_as_null=False, astext_type=None):
        """
        Create JSONB type.
        
        Parameters:
        - none_as_null: bool, store None as SQL NULL  
        - astext_type: type for ->> text extraction
        """

class UUID:
    """PostgreSQL UUID type."""
    
    def __init__(self, as_uuid=True):
        """
        Create UUID type.
        
        Parameters:
        - as_uuid: bool, return UUID objects vs strings
        """

# Network address types
class INET:
    """PostgreSQL IP address type."""

class CIDR:
    """PostgreSQL network address type."""

class MACADDR:
    """PostgreSQL MAC address type."""

# Range types
class INT4RANGE:
    """PostgreSQL integer range type."""
    
    def __init__(self, bounds="[)"):
        """
        Create integer range type.
        
        Parameters:
        - bounds: str, default bounds ("[)", "(]", "[]", "()")
        """

class INT8RANGE:
    """PostgreSQL big integer range type."""

class NUMRANGE:
    """PostgreSQL numeric range type."""

class DATERANGE:
    """PostgreSQL date range type."""

class TSRANGE:
    """PostgreSQL timestamp range type."""

class TSTZRANGE:
    """PostgreSQL timestamp with timezone range type."""

# Full-text search types
class TSVECTOR:
    """PostgreSQL text search vector type."""

class TSQUERY:
    """PostgreSQL text search query type."""

# PostgreSQL functions
def to_tsvector(config, text):
    """Convert text to search vector."""

def to_tsquery(config, query):
    """Convert text to search query."""

def plainto_tsquery(config, query):
    """Convert plain text to search query."""

# PostgreSQL constraints
class ExcludeConstraint:
    """PostgreSQL EXCLUDE constraint."""
    
    def __init__(self, *elements, **kwargs):
        """
        Create exclude constraint.
        
        Parameters:
        - elements: (column, operator) tuples
        - name: str, constraint name
        - using: str, index method (default 'gist')
        """

MySQL Dialect

MySQL-specific types, functions, and SQL features.

# Import MySQL-specific types
from sqlalchemy.dialects.mysql import MEDIUMINT, LONGTEXT, MEDIUMTEXT, TINYTEXT
from sqlalchemy.dialects.mysql import SET, ENUM, YEAR, BIT
from sqlalchemy.dialects.mysql import Insert

class MEDIUMINT:
    """MySQL medium integer type (3 bytes)."""
    
    def __init__(self, display_width=None, unsigned=False, zerofill=False):
        """
        Create MEDIUMINT type.
        
        Parameters:
        - display_width: int, display width
        - unsigned: bool, unsigned integer
        - zerofill: bool, zero-fill display
        """

class LONGTEXT:
    """MySQL long text type (4GB)."""

class MEDIUMTEXT:
    """MySQL medium text type (16MB)."""

class TINYTEXT:
    """MySQL tiny text type (255 bytes)."""

class SET:
    """MySQL SET type for multiple choice values."""
    
    def __init__(self, *values, **kwargs):
        """
        Create SET type.
        
        Parameters:
        - values: allowed set values
        - charset: str, character set
        - collation: str, collation
        """

class ENUM:
    """MySQL ENUM type."""
    
    def __init__(self, *enums, **kwargs):
        """
        Create ENUM type.
        
        Parameters:
        - enums: enumeration values
        - charset: str, character set
        - collation: str, collation
        """

class YEAR:
    """MySQL YEAR type."""
    
    def __init__(self, display_width=None):
        """
        Create YEAR type.
        
        Parameters:
        - display_width: int, display width (2 or 4)
        """

class BIT:
    """MySQL BIT type."""
    
    def __init__(self, length=None):
        """
        Create BIT type.
        
        Parameters:
        - length: int, bit length (1-64)
        """

class Insert:
    """MySQL-specific INSERT with ON DUPLICATE KEY UPDATE."""
    
    def on_duplicate_key_update(self, **kwargs):
        """
        Add ON DUPLICATE KEY UPDATE clause.
        
        Parameters:
        - kwargs: column=value pairs for updates
        
        Returns:
        Insert: Modified INSERT with ON DUPLICATE KEY UPDATE
        """

# MySQL functions
def match(*columns, against=None, in_boolean_mode=False, in_natural_language_mode=False):
    """
    MySQL MATCH() AGAINST() full-text search.
    
    Parameters:
    - columns: columns to search
    - against: search expression
    - in_boolean_mode: bool, use boolean mode
    - in_natural_language_mode: bool, use natural language mode
    
    Returns:
    Function: MATCH function call
    """

SQLite Dialect

SQLite-specific features and limitations handling.

# Import SQLite-specific types
from sqlalchemy.dialects.sqlite import JSON, Insert

class JSON:
    """SQLite JSON type (SQLite 3.38+)."""
    
    def __init__(self, none_as_null=False):
        """
        Create SQLite JSON type.
        
        Parameters:
        - none_as_null: bool, store None as SQL NULL
        """

class Insert:
    """SQLite-specific INSERT with ON CONFLICT handling."""
    
    def on_conflict_do_update(self, index_elements=None, set_=None, **kwargs):
        """
        Add ON CONFLICT DO UPDATE clause.
        
        Parameters:
        - index_elements: conflicting columns/expressions
        - set_: dictionary of column updates
        
        Returns:
        Insert: Modified INSERT with ON CONFLICT DO UPDATE
        """
    
    def on_conflict_do_nothing(self, index_elements=None):
        """
        Add ON CONFLICT DO NOTHING clause.
        
        Parameters:
        - index_elements: conflicting columns/expressions
        
        Returns:
        Insert: Modified INSERT with ON CONFLICT DO NOTHING
        """

Oracle Dialect

Oracle-specific types, functions, and SQL constructs.

# Import Oracle-specific types
from sqlalchemy.dialects.oracle import NUMBER, LONG, RAW, NCLOB, BLOB, BFILE
from sqlalchemy.dialects.oracle import ROWNUM

class NUMBER:
    """Oracle NUMBER type with precision and scale."""
    
    def __init__(self, precision=None, scale=None, asdecimal=None):
        """
        Create Oracle NUMBER type.
        
        Parameters:
        - precision: int, total digits
        - scale: int, digits after decimal
        - asdecimal: bool, return Decimal objects
        """

class LONG:
    """Oracle LONG type for large text."""

class RAW:
    """Oracle RAW type for binary data."""
    
    def __init__(self, length=None):
        """
        Create RAW type.
        
        Parameters:
        - length: int, maximum byte length
        """

class NCLOB:
    """Oracle NCLOB type for large Unicode text."""

class BLOB:
    """Oracle BLOB type for binary large objects."""

class BFILE:
    """Oracle BFILE type for external file references."""

class ROWNUM:
    """Oracle ROWNUM pseudocolumn."""

SQL Server Dialect

Microsoft SQL Server specific types and features.

# Import SQL Server-specific types
from sqlalchemy.dialects.mssql import UNIQUEIDENTIFIER, IMAGE, MONEY, SMALLMONEY
from sqlalchemy.dialects.mssql import SQL_VARIANT, ROWVERSION, try_cast

class UNIQUEIDENTIFIER:
    """SQL Server GUID/UUID type."""

class IMAGE:
    """SQL Server IMAGE type (deprecated, use VARBINARY(MAX))."""

class MONEY:
    """SQL Server MONEY type for currency."""

class SMALLMONEY:
    """SQL Server SMALLMONEY type for smaller currency values."""

class SQL_VARIANT:
    """SQL Server sql_variant type for mixed data types."""

class ROWVERSION:
    """SQL Server rowversion/timestamp type for concurrency."""

def try_cast(expression, type_):
    """
    SQL Server TRY_CAST function.
    
    Parameters:
    - expression: expression to cast
    - type_: target type
    
    Returns:
    Function: TRY_CAST function call
    """

Dialect-Specific Features

Database-specific SQL generation and optimization features.

# PostgreSQL-specific query features
from sqlalchemy.dialects.postgresql import aggregate_order_by, array_agg

def aggregate_order_by(expr, *order_by):
    """
    PostgreSQL aggregate ORDER BY clause.
    
    Parameters:
    - expr: aggregate expression
    - order_by: ordering expressions
    
    Returns:
    AggregateOrderBy: Ordered aggregate expression
    """

def array_agg(*args, **kwargs):
    """
    PostgreSQL ARRAY_AGG function.
    
    Parameters:
    - args: expressions to aggregate
    - order_by: ordering for aggregation
    
    Returns:
    Function: ARRAY_AGG function call
    """

# MySQL-specific features
from sqlalchemy.dialects.mysql import match

# SQL Server-specific features  
from sqlalchemy.dialects.mssql import try_cast

Usage Examples

PostgreSQL Arrays and JSON

from sqlalchemy.dialects.postgresql import ARRAY, JSONB, UUID
from sqlalchemy import Table, Column, Integer, String

# Array usage
tags_table = Table('articles', metadata,
    Column('id', UUID, primary_key=True),
    Column('title', String(200)),
    Column('tags', ARRAY(String(50))),
    Column('ratings', ARRAY(Integer, dimensions=2)),  # 2D array
    Column('metadata', JSONB)
)

# Query arrays
from sqlalchemy import select, func

stmt = select(tags_table).where(
    tags_table.c.tags.contains(['python', 'sqlalchemy'])
)

# JSON operations
stmt = select(tags_table).where(
    tags_table.c.metadata['author']['name'].astext == 'John Doe'
)

MySQL ON DUPLICATE KEY UPDATE

from sqlalchemy.dialects.mysql import Insert

stmt = Insert(users).values(
    id=1,
    name='John Doe',
    email='john@example.com',
    login_count=1
)

stmt = stmt.on_duplicate_key_update(
    name=stmt.inserted.name,
    email=stmt.inserted.email,
    login_count=users.c.login_count + 1
)

with engine.connect() as conn:
    conn.execute(stmt)

SQLite ON CONFLICT Handling

from sqlalchemy.dialects.sqlite import Insert

stmt = Insert(users).values(
    username='johndoe',
    email='john@example.com'
)

# Ignore conflicts
stmt = stmt.on_conflict_do_nothing()

# Update on conflict
stmt = stmt.on_conflict_do_update(
    index_elements=['username'],
    set_={'email': stmt.excluded.email}
)

PostgreSQL Full-Text Search

from sqlalchemy.dialects.postgresql import TSVECTOR, TSQUERY
from sqlalchemy.dialects.postgresql import to_tsvector, to_tsquery

documents = Table('documents', metadata,
    Column('id', Integer, primary_key=True),
    Column('title', String(200)),
    Column('content', Text),
    Column('search_vector', TSVECTOR)
)

# Create GIN index for full-text search
from sqlalchemy import Index
search_idx = Index('idx_search_vector', documents.c.search_vector, 
                   postgresql_using='gin')

# Full-text search query
stmt = select(documents).where(
    documents.c.search_vector.match('python & sqlalchemy')
)

# Update search vector
stmt = documents.update().values(
    search_vector=to_tsvector('english', 
                              func.coalesce(documents.c.title, '') + ' ' + 
                              func.coalesce(documents.c.content, ''))
)

Oracle-Specific Features

from sqlalchemy.dialects.oracle import NUMBER, ROWNUM

accounts = Table('accounts', metadata,
    Column('id', NUMBER(10), primary_key=True),
    Column('balance', NUMBER(15, 2)),
    Column('account_number', String(20))
)

# Use ROWNUM for pagination (Oracle < 12c)
stmt = select(accounts).where(ROWNUM <= 10)

# Oracle 12c+ pagination
stmt = select(accounts).order_by(accounts.c.id).offset(10).limit(10)

SQL Server Features

from sqlalchemy.dialects.mssql import UNIQUEIDENTIFIER, try_cast

orders = Table('orders', metadata,
    Column('id', UNIQUEIDENTIFIER, primary_key=True),
    Column('amount_text', String(50)),
    Column('customer_id', Integer)
)

# Use TRY_CAST for safe type conversion
stmt = select(
    orders.c.id,
    try_cast(orders.c.amount_text, Numeric(10, 2)).label('amount')
).where(
    try_cast(orders.c.amount_text, Numeric(10, 2)) > 100
)

Install with Tessl CLI