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

Script Management

Name: tessl/pypi-alembic
Author: tessl

Classes and utilities for managing migration scripts, revision tracking, and dependency graphs. The script management system handles migration file generation, versioning, and execution ordering.

Core Imports

from alembic.script import ScriptDirectory, Script
from alembic.script.revision import RevisionMap

Capabilities

Script Directory Management

Main class for managing the migration script directory and its contents.

class ScriptDirectory:
    def __init__(self, dir, file_template=None, truncate_slug_length=40, version_locations=None, sourceless=False, output_encoding='utf-8', timezone=None, hook=None, env_py_location=None):
        """
        Initialize script directory manager.
        
        Args:
            dir (str): Path to script directory
            file_template (str): Template for migration file names
            truncate_slug_length (int): Maximum length for revision slugs
            version_locations (list): Additional version file locations
            sourceless (bool): Whether to support .pyc-only migrations
            output_encoding (str): Encoding for generated files
            timezone (str): Timezone for timestamps
            hook (callable): Custom hook function
            env_py_location (str): Path to env.py file
        """

    @classmethod
    def from_config(cls, config):
        """
        Create ScriptDirectory from Alembic configuration.
        
        Args:
            config (Config): Alembic configuration object
        
        Returns:
            ScriptDirectory: Configured script directory
        """

    def get_revision(self, id_):
        """
        Get a revision by ID.
        
        Args:
            id_ (str): Revision identifier
        
        Returns:
            Script: Script object for the revision
        """

    def get_revisions(self, id_):
        """
        Get multiple revisions by ID pattern.
        
        Args:
            id_ (str): Revision identifier or pattern
        
        Returns:
            tuple: Tuple of Script objects
        """

    def get_all_current(self, id_):
        """
        Get all current head revisions.
        
        Args:
            id_ (str): Starting revision identifier
        
        Returns:
            tuple: Tuple of current head revisions
        """

    def get_heads(self):
        """
        Get all head revisions.
        
        Returns:
            tuple: Tuple of head revision identifiers
        """

    def get_base(self):
        """
        Get base revision identifier.
        
        Returns:
            str: Base revision identifier
        """

    def walk_revisions(self, base='base', head='heads'):
        """
        Walk through revisions from base to head.
        
        Args:
            base (str): Starting revision
            head (str): Ending revision
        
        Yields:
            Script: Script objects in dependency order
        """

    def get_current_head(self):
        """
        Get the current single head revision.
        
        Returns:
            str: Head revision identifier
        
        Raises:
            CommandError: If multiple heads exist
        """

Usage Examples:

from alembic.script import ScriptDirectory
from alembic.config import Config

# Create from configuration
config = Config('alembic.ini')
script_dir = ScriptDirectory.from_config(config)

# Get revision information
latest_rev = script_dir.get_current_head()
all_heads = script_dir.get_heads()

# Walk through revision history
for script in script_dir.walk_revisions():
    print(f"Revision: {script.revision}, Message: {script.doc}")

# Get specific revision
rev_abc123 = script_dir.get_revision('abc123')
print(f"Down revision: {rev_abc123.down_revision}")

Revision Generation

Generate new migration script files.

class ScriptDirectory:
    def generate_revision(self, revid, message, refresh=False, head='head', splice=False, branch_labels=None, version_path=None, depends_on=None, **kw):
        """
        Generate a new revision script.
        
        Args:
            revid (str): Revision identifier
            message (str): Revision message
            refresh (bool): Refresh revision map
            head (str): Head revision to base on
            splice (bool): Allow revision to be spliced
            branch_labels (list): Branch labels for revision
            version_path (str): Path for version files
            depends_on (list): Dependencies for revision
            **kw: Additional template variables
        
        Returns:
            Script: Generated script object
        """

    def run_env(self):
        """
        Execute the env.py script.
        
        Returns:
            dict: Environment module's globals
        """

    def get_upgrade_token(self):
        """
        Get the upgrade token from env.py.
        
        Returns:
            str: Upgrade token identifier
        """

    def get_downgrade_token(self):
        """
        Get the downgrade token from env.py.
        
        Returns:
            str: Downgrade token identifier
        """

Usage Examples:

# Generate new revision
script = script_dir.generate_revision(
    revid='abc123',
    message='Create user table',
    head='head',
    branch_labels=['feature']
)

# Generate with dependencies
script = script_dir.generate_revision(
    revid='def456', 
    message='Add indexes',
    depends_on=['abc123']
)

# Generate with custom template variables
script = script_dir.generate_revision(
    revid='ghi789',
    message='Custom migration',
    author='John Doe',
    created_by='AutoGen'
)

Script Objects

Individual migration script representation.

class Script:
    def __init__(self, module, rev_id, path):
        """
        Initialize script object.
        
        Args:
            module: Python module containing the migration
            rev_id (str): Revision identifier
            path (str): File path to script
        """
    
    # Properties
    revision: str  # Revision identifier
    down_revision: Optional[str]  # Previous revision
    branch_labels: Optional[Set[str]]  # Branch labels
    depends_on: Optional[Set[str]]  # Dependencies
    path: str  # File path
    doc: str  # Revision message
    module: ModuleType  # Python module
    
    def get_path(self):
        """
        Get the file path for this script.
        
        Returns:
            str: Path to script file
        """

    def nextrev(self):
        """
        Get the next revision in the chain.
        
        Returns:
            str: Next revision identifier
        """

    def is_base(self):
        """
        Check if this is a base revision.
        
        Returns:
            bool: True if base revision
        """

    def is_head(self):
        """
        Check if this is a head revision.
        
        Returns:
            bool: True if head revision
        """

    def is_merge_point(self):
        """
        Check if this is a merge point.
        
        Returns:
            bool: True if merge point
        """

    def is_branch_point(self):
        """
        Check if this is a branch point.
        
        Returns:
            bool: True if branch point
        """

Usage Examples:

# Access script properties
script = script_dir.get_revision('abc123')
print(f"Revision: {script.revision}")
print(f"Message: {script.doc}")
print(f"Path: {script.path}")
print(f"Down revision: {script.down_revision}")
print(f"Branch labels: {script.branch_labels}")

# Check script status
if script.is_head():
    print("This is a head revision")
if script.is_merge_point():
    print("This is a merge point")

Revision Maps

Manage revision dependency graphs and navigation.

class RevisionMap:
    def __init__(self, generator):
        """
        Initialize revision map.
        
        Args:
            generator: Generator function yielding Script objects
        """

    def get_revision(self, id_):
        """
        Get revision by identifier.
        
        Args:
            id_ (str): Revision identifier
        
        Returns:
            Script: Script object
        """

    def get_revisions(self, id_):
        """
        Get multiple revisions by identifier.
        
        Args:
            id_ (str): Revision identifier pattern
        
        Returns:
            tuple: Tuple of Script objects
        """

    def iterate_revisions(self, upper, lower):
        """
        Iterate revisions between upper and lower bounds.
        
        Args:
            upper (str): Upper bound revision
            lower (str): Lower bound revision
        
        Yields:
            Script: Script objects in order
        """

    def get_heads(self):
        """
        Get all head revision identifiers.
        
        Returns:
            tuple: Head revision identifiers
        """

    def get_bases(self):
        """
        Get all base revision identifiers.
        
        Returns:
            tuple: Base revision identifiers
        """

Version Locations

Manage multiple locations for migration scripts.

# Configure multiple version locations
script_dir = ScriptDirectory(
    'migrations',
    version_locations=[
        'migrations/versions',    # Default location
        'migrations/feature',     # Feature branch migrations  
        'migrations/hotfix'       # Hotfix migrations
    ]
)

# Generate revision in specific location
script = script_dir.generate_revision(
    revid='feature_001',
    message='Feature migration',
    version_path='migrations/feature'
)

Branching and Merging

Branch Management

# Create branch revision
branch_script = script_dir.generate_revision(
    revid='branch_001',
    message='Start feature branch',
    head='abc123',  # Branch from this revision
    branch_labels=['feature_x']
)

# Create merge revision
merge_script = script_dir.generate_revision(
    revid='merge_001', 
    message='Merge feature branch',
    head=['head', 'feature_x']  # Merge multiple heads
)

Dependency Management

# Create revision with dependencies
dependent_script = script_dir.generate_revision(
    revid='dep_001',
    message='Migration with dependencies',
    depends_on=['abc123', 'def456']  # Depends on these revisions
)

# Check dependencies
for dep in dependent_script.depends_on:
    dep_script = script_dir.get_revision(dep)
    print(f"Depends on: {dep_script.doc}")

Template Customization

Custom File Templates

# Custom template with additional variables
custom_template = '''"""${message}

Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}
Author: ${author}
Branch: ${branch_name}
"""

from alembic import op
import sqlalchemy as sa
${imports if imports else ""}

# revision identifiers, used by Alembic.
revision = ${repr(up_revision)}
down_revision = ${repr(down_revision)}
branch_labels = ${repr(branch_labels)}
depends_on = ${repr(depends_on)}

def upgrade():
    ${upgrades if upgrades else "pass"}

def downgrade():
    ${downgrades if downgrades else "pass"}
'''

# Use custom template
script_dir = ScriptDirectory(
    'migrations',
    file_template=custom_template
)

Template Variables

Available variables in migration templates:

${message}: Revision message
${up_revision}: Revision identifier
${down_revision}: Previous revision identifier
${create_date}: Creation timestamp
${branch_labels}: Branch labels
${depends_on}: Dependencies
${upgrades}: Upgrade operations (autogenerate)
${downgrades}: Downgrade operations (autogenerate)
${imports}: Required imports (autogenerate)

Advanced Features

Sourceless Migrations

Run migrations from compiled .pyc files without .py sources.

# Configure sourceless mode
script_dir = ScriptDirectory(
    'migrations',
    sourceless=True  # Support .pyc-only migrations
)

Custom Hook Functions

Execute custom logic during script operations.

def custom_hook(rev, context):
    """Custom hook function."""
    print(f"Processing revision: {rev}")
    # Custom validation or processing logic

script_dir = ScriptDirectory(
    'migrations', 
    hook=custom_hook
)

Environment Integration

# Custom env.py location
script_dir = ScriptDirectory(
    'migrations',
    env_py_location='custom/env.py'
)

# Execute environment script
env_globals = script_dir.run_env()
target_metadata = env_globals.get('target_metadata')

Error Handling

Script management operations may raise:

RevisionError: Revision identifier conflicts or invalid references
MultipleHeads: Multiple head revisions when single expected
CommandError: General script management errors
FileNotFoundError: Missing script files or directories

Integration Examples

Web Application Integration

from alembic.script import ScriptDirectory
from alembic.config import Config

class MigrationManager:
    def __init__(self, config_path):
        self.config = Config(config_path)
        self.script_dir = ScriptDirectory.from_config(self.config)
    
    def get_migration_status(self):
        """Get current migration status."""
        current_heads = self.script_dir.get_heads()
        return {
            'heads': current_heads,
            'is_current': len(current_heads) == 1
        }
    
    def list_migrations(self):
        """List all migrations."""
        migrations = []
        for script in self.script_dir.walk_revisions():
            migrations.append({
                'revision': script.revision,
                'message': script.doc,
                'down_revision': script.down_revision
            })
        return migrations

Automated Migration Generation

def auto_generate_migration(app_metadata, message):
    """Generate migration automatically."""
    config = Config('alembic.ini')
    script_dir = ScriptDirectory.from_config(config)
    
    # Generate unique revision ID
    import uuid
    rev_id = str(uuid.uuid4())[:12]
    
    # Generate migration script
    script = script_dir.generate_revision(
        revid=rev_id,
        message=message,
        autogenerate=True,
        target_metadata=app_metadata
    )
    
    return script

Types

# Script management types
class Script:
    revision: str
    down_revision: Optional[str]
    branch_labels: Optional[Set[str]]
    depends_on: Optional[Set[str]]
    path: str
    doc: str
    module: ModuleType

class ScriptDirectory:
    dir: str
    file_template: str
    truncate_slug_length: int
    version_locations: Optional[List[str]]
    
class RevisionMap:
    map: Dict[str, Script]
    
# Hook function type
HookFunc = Callable[[str, Any], None]

Install with Tessl CLI