tessl/pypi-alembic

A database migration tool for SQLAlchemy.

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Migration Operations

Name: tessl/pypi-alembic
Author: tessl

Schema modification operations available in migration scripts through the alembic.op module. These operations provide a high-level interface for database schema changes including tables, columns, constraints, indexes, and data manipulation.

Core Imports

from alembic import op
import sqlalchemy as sa

Capabilities

Table Operations

Create, drop, and modify database tables.

def create_table(table_name, *columns, if_not_exists=None, **kw):
    """
    Create a new table.
    
    Args:
        table_name (str): Name of the table to create
        *columns: SQLAlchemy Column objects
        if_not_exists (bool): Add IF NOT EXISTS clause (v1.16.0+)
        **kw: Additional table options (schema, mysql_engine, etc.)
    
    Returns:
        Table: SQLAlchemy Table object
    """

def drop_table(table_name, if_exists=None, **kw):
    """
    Drop an existing table.
    
    Args:
        table_name (str): Name of the table to drop
        if_exists (bool): Add IF EXISTS clause (v1.16.0+)
        **kw: Additional options (schema, etc.)
        
    Returns:
        None
    """

def rename_table(old_table_name, new_table_name, schema=None):
    """
    Rename a table.
    
    Args:
        old_table_name (str): Current table name
        new_table_name (str): New table name
        schema (str): Schema name if applicable
        
    Returns:
        None
    """

Usage Examples:

# Create table with columns
op.create_table('user',
    sa.Column('id', sa.Integer, primary_key=True),
    sa.Column('name', sa.String(50), nullable=False),
    sa.Column('email', sa.String(120), unique=True),
    sa.Column('created_at', sa.DateTime, default=sa.func.now())
)

# Drop table
op.drop_table('old_table')

# Rename table
op.rename_table('user', 'users')

Column Operations

Add, remove, and modify table columns.

def add_column(table_name, column, if_not_exists=None, **kw):
    """
    Add a column to an existing table.
    
    Args:
        table_name (str): Name of the table
        column (Column): SQLAlchemy Column object to add
        if_not_exists (bool): Add IF NOT EXISTS clause (v1.16.0+)
        **kw: Additional options (schema, etc.)
    """

def drop_column(table_name, column_name, **kw):
    """
    Drop a column from a table.
    
    Args:
        table_name (str): Name of the table
        column_name (str): Name of the column to drop
        **kw: Additional options (schema, etc.)
    """

def alter_column(table_name, column_name, nullable=None, comment=False, server_default=False, new_column_name=None, type_=None, **kw):
    """
    Alter properties of an existing column.
    
    Args:
        table_name (str): Name of the table
        column_name (str): Name of the column to alter
        nullable (bool): Change nullable constraint
        comment (str|False): Change column comment (False=no change, None=remove)
        server_default (str|False): Change server default (False=no change, None=remove)
        new_column_name (str): Rename the column
        type_ (TypeEngine): Change column type
        **kw: Additional options (schema, etc.)
    """

Usage Examples:

# Add column
op.add_column('user', sa.Column('phone', sa.String(20)))

# Drop column
op.drop_column('user', 'phone')

# Alter column - change type and nullable
op.alter_column('user', 'email', type_=sa.String(200), nullable=True)

# Rename column
op.alter_column('user', 'name', new_column_name='full_name')

# Change server default
op.alter_column('user', 'created_at', server_default=sa.text('CURRENT_TIMESTAMP'))

Constraint Operations

Create and drop various types of database constraints.

def create_primary_key(constraint_name, table_name, columns, schema=None):
    """
    Create a primary key constraint.
    
    Args:
        constraint_name (str): Name of the constraint
        table_name (str): Name of the table
        columns (list): List of column names
        schema (str): Schema name if applicable
        
    Returns:
        None
    """

def create_foreign_key(constraint_name, source_table, referent_table, local_cols, remote_cols, onupdate=None, ondelete=None, deferrable=None, initially=None, match=None, source_schema=None, referent_schema=None, **dialect_kw):
    """
    Create a foreign key constraint.
    
    Args:
        constraint_name (str): Name of the constraint
        source_table (str): Source table name
        referent_table (str): Referenced table name
        local_cols (list): Local column names
        remote_cols (list): Referenced column names
        onupdate (str): ON UPDATE action
        ondelete (str): ON DELETE action
        deferrable (bool): Whether constraint is deferrable
        initially (str): Initial constraint state ('DEFERRED' or 'IMMEDIATE')
        match (str): MATCH clause ('FULL', 'PARTIAL', 'SIMPLE')
        source_schema (str): Source table schema
        referent_schema (str): Referenced table schema
        **dialect_kw: Additional dialect-specific options
        
    Returns:
        None
    """

def create_check_constraint(constraint_name, table_name, condition, schema=None, **kw):
    """
    Create a check constraint.
    
    Args:
        constraint_name (str): Name of the constraint
        table_name (str): Name of the table
        condition (str|ClauseElement): Check condition
        schema (str): Schema name if applicable
        **kw: Additional options
        
    Returns:
        None
    """

def create_unique_constraint(constraint_name, table_name, columns, schema=None, **kw):
    """
    Create a unique constraint.
    
    Args:
        constraint_name (str): Name of the constraint
        table_name (str): Name of the table
        columns (list): List of column names
        schema (str): Schema name if applicable
        **kw: Additional options
        
    Returns:
        Any: Constraint object
    """

def create_exclude_constraint(constraint_name, table_name, *elements, where=None, schema=None, deferrable=None, initially=None, using=None, **kw):
    """
    Create an exclude constraint (PostgreSQL).
    
    Args:
        constraint_name (str): Name of the constraint
        table_name (str): Name of the table
        *elements: Column/expression and operator pairs
        where (str): WHERE clause for partial constraint
        schema (str): Schema name if applicable
        deferrable (bool): Whether constraint is deferrable
        initially (str): Initial constraint checking mode
        using (str): Index method to use
        **kw: Additional options
        
    Returns:
        Optional[Table]: Table object (if created)
    """

def drop_constraint(constraint_name, table_name, type_=None, if_exists=None, schema=None):
    """
    Drop a constraint.
    
    Args:
        constraint_name (str): Name of the constraint to drop
        table_name (str): Name of the table
        type_ (str): Constraint type ('foreignkey', 'primary', 'unique', 'check')
        if_exists (bool): Add IF EXISTS clause (v1.16.0+)
        schema (str): Schema name if applicable
        
    Returns:
        None
    """

Usage Examples:

# Create primary key
op.create_primary_key('pk_user', 'user', ['id'])

# Create foreign key
op.create_foreign_key('fk_post_user', 'post', 'user', ['user_id'], ['id'], ondelete='CASCADE')

# Create unique constraint
op.create_unique_constraint('uq_user_email', 'user', ['email'])

# Create check constraint
op.create_check_constraint('ck_user_age', 'user', 'age >= 0')

# Drop constraint
op.drop_constraint('fk_old_constraint', 'table_name', type_='foreignkey')

Index Operations

Create and drop database indexes for query optimization.

def create_index(index_name, table_name, columns, schema=None, unique=False, if_not_exists=None, **kw):
    """
    Create an index.
    
    Args:
        index_name (str): Name of the index
        table_name (str): Name of the table
        columns (list): List of column names or expressions
        schema (str): Schema name if applicable
        unique (bool): Whether index should be unique
        if_not_exists (bool): Add IF NOT EXISTS clause (v1.12.0+)
        **kw: Database-specific index options
        
    Returns:
        None
    """

def drop_index(index_name, table_name=None, schema=None, if_exists=None, **kw):
    """
    Drop an index.
    
    Args:
        index_name (str): Name of the index to drop
        table_name (str): Name of the table (optional, some databases require)
        schema (str): Schema name if applicable
        if_exists (bool): Add IF EXISTS clause (v1.12.0+)
        **kw: Additional options
        
    Returns:
        None
    """

Usage Examples:

# Create simple index
op.create_index('ix_user_email', 'user', ['email'])

# Create unique index
op.create_index('ix_user_username', 'user', ['username'], unique=True)

# Create composite index
op.create_index('ix_post_user_date', 'post', ['user_id', 'created_at'])

# Drop index
op.drop_index('ix_old_index', 'user')

Data Operations

Execute SQL statements and perform bulk data operations.

def execute(sqltext, execution_options=None):
    """
    Execute arbitrary SQL.
    
    Args:
        sqltext (str|ClauseElement): SQL statement to execute
        execution_options (dict): Execution options
    """

def bulk_insert(table, rows, multiinsert=True):
    """
    Perform bulk insert operation.
    
    Args:
        table (Table|str): Target table
        rows (list): List of dictionaries representing rows
        multiinsert (bool): Use multi-row INSERT statements
    """

def inline_literal(value):
    """
    Create an inline literal value for SQL generation.
    
    Args:
        value: Python value to convert to SQL literal
    
    Returns:
        Literal value for inline SQL
    """

Usage Examples:

# Execute raw SQL
op.execute("UPDATE user SET status = 'active' WHERE created_at > '2023-01-01'")

# Bulk insert data
user_table = sa.table('user',
    sa.column('name'),
    sa.column('email')
)
op.bulk_insert(user_table, [
    {'name': 'John', 'email': 'john@example.com'},
    {'name': 'Jane', 'email': 'jane@example.com'}
])

# Use inline literal
op.execute(f"INSERT INTO config (key, value) VALUES ('version', {op.inline_literal('1.0')})")

Batch Operations

Special operations for databases with limited ALTER support (particularly SQLite).

def batch_alter_table(table_name, schema=None, recreate='auto', copy_from=None, table_args=(), table_kwargs=None, reflect_args=(), reflect_kwargs=None, naming_convention=None):
    """
    Context manager for batch table alterations.
    
    Args:
        table_name (str): Name of the table to alter
        schema (str): Schema name if applicable
        recreate (str): Recreation strategy ('auto', 'always', 'never')
        copy_from (Table): Source table to copy from
        table_args: Arguments for new table
        table_kwargs: Keyword arguments for new table
        reflect_args: Arguments for table reflection
        reflect_kwargs: Keyword arguments for table reflection
        naming_convention (dict): Naming convention for constraints
    
    Returns:
        BatchOperations: Context manager for batch operations
    """

Usage Examples:

# Batch alter table for SQLite compatibility
with op.batch_alter_table('user') as batch_op:
    batch_op.add_column(sa.Column('phone', sa.String(20)))
    batch_op.alter_column('email', nullable=True)
    batch_op.create_unique_constraint('uq_user_phone', ['phone'])
    batch_op.drop_column('old_field')

Table Comments

Add and remove table-level comments.

def create_table_comment(table_name, comment, schema=None):
    """
    Create a table comment.
    
    Args:
        table_name (str): Name of the table
        comment (str): Comment text
        schema (str): Schema name if applicable
    """

def drop_table_comment(table_name, schema=None):
    """
    Drop a table comment.
    
    Args:
        table_name (str): Name of the table
        schema (str): Schema name if applicable
    """

Advanced Operations

Advanced operation management and customization.

def get_bind():
    """
    Get the current database connection.
    
    Returns:
        Connection: SQLAlchemy connection object
    """

def get_context():
    """
    Get the current migration context.
    
    Returns:
        MigrationContext: Current migration context
    """

def f(name):
    """
    Mark a name as having a naming convention applied.
    
    Args:
        name (str): Name to mark for naming convention processing
        
    Returns:
        conv: Naming convention placeholder
    """

def run_async(async_function, *args, **kw_args):
    """
    Run an async function in migration context (v1.11+).
    
    Args:
        async_function: Async function to execute
        *args: Positional arguments
        **kw_args: Keyword arguments
        
    Returns:
        Any: Function result
    """

def invoke(operation):
    """
    Invoke a migration operation programmatically.
    
    Args:
        operation: Operation object to invoke
        
    Returns:
        Any: Operation result
    """

def implementation_for(op_cls):
    """
    Register an implementation for a custom operation class.
    
    Args:
        op_cls: Operation class to register implementation for
        
    Returns:
        Callable: Decorator function
    """

def register_operation(name, sourcename=None):
    """
    Register a new custom operation.
    
    Args:
        name (str): Name of the operation
        sourcename (str): Source module name
        
    Returns:
        Callable: Decorator function
    """

def invoke(operation):
    """
    Invoke a migration operation.
    
    Args:
        operation (MigrateOperation): Operation to invoke
    """

def f(name):
    """
    Create a naming convention reference.
    
    Args:
        name (str): Template name
    
    Returns:
        str: Formatted name according to naming convention
    """

def implementation_for(op_cls):
    """
    Register operation implementation for custom operations.
    
    Args:
        op_cls: Operation class to register implementation for
    
    Returns:
        Decorator function for implementation registration
    """

def register_operation(name, operation_class):
    """
    Register a custom operation.
    
    Args:
        name (str): Operation name
        operation_class: Custom operation class
    """

def run_async(coro):
    """
    Run async operation within migration context.
    
    Args:
        coro: Coroutine to execute
    
    Returns:
        Result of coroutine execution
    """

Operation Context

All operations are executed within a migration context that provides:

Database connection management
Transaction handling
SQL dialect-specific behavior
Offline SQL generation mode
Custom operation implementations

Database-Specific Considerations

PostgreSQL

Supports most operations natively
Concurrent index creation: postgresql_concurrently=True
Array and JSONB column types supported

MySQL

Limited foreign key support in some versions
Engine-specific options: mysql_engine='InnoDB'
Charset and collation options available

SQLite

Limited ALTER TABLE support
Use batch_alter_table() for complex changes
Foreign key constraints require special handling

SQL Server

Schema-qualified names supported
Identity column considerations
Specific index options available

Error Handling

Operations may raise:

OperationalError: Database-level errors
ProgrammingError: SQL syntax or logic errors
IntegrityError: Constraint violations
CommandError: Alembic-specific operation errors

Types

# Operation result types
class Table:
    name: str
    schema: Optional[str]
    columns: List[Column]

class BatchOperations:
    def add_column(self, column): ...
    def drop_column(self, column_name): ...
    def alter_column(self, column_name, **kw): ...
    def create_index(self, index_name, columns, **kw): ...
    def create_unique_constraint(self, constraint_name, columns): ...
    def create_foreign_key(self, constraint_name, referent_table, local_cols, remote_cols): ...
    def drop_constraint(self, constraint_name): ...

Install with Tessl CLI