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
runtime.mddocs/
Runtime Components
Core runtime classes and utilities for migration execution, context management, and revision tracking. These components provide the underlying infrastructure for Alembic's migration operations.
Core Imports
from alembic.runtime.migration import MigrationContext
from alembic.runtime.environment import EnvironmentContextCapabilities
Migration Context
Core context class that manages database connections, dialect-specific behavior, and migration execution state.
class MigrationContext:
def __init__(self, dialect, connection, opts, environment_context=None):
"""
Initialize migration context.
Args:
dialect (Dialect): SQLAlchemy dialect
connection (Connection): Database connection
opts (dict): Configuration options
environment_context (EnvironmentContext): Environment context
"""
@classmethod
def configure(cls, connection=None, url=None, dialect_name=None, dialect_opts=None, opts=None, environment_context=None):
"""
Configure and create migration context.
Args:
connection (Connection): Database connection
url (str): Database URL if connection not provided
dialect_name (str): Dialect name
dialect_opts (dict): Dialect options
opts (dict): Migration options
environment_context (EnvironmentContext): Environment context
Returns:
MigrationContext: Configured migration context
"""
def get_current_revision(self):
"""
Get current revision from database.
Returns:
str: Current revision identifier
"""
def stamp(self, script_directory, revision):
"""
Stamp database with revision without running migration.
Args:
script_directory (ScriptDirectory): Script directory
revision (str): Revision to stamp
"""
def run_migrations(self, **kw):
"""
Execute migration operations.
Args:
**kw: Additional migration arguments
"""
def execute(self, sql, execution_options=None):
"""
Execute SQL statement.
Args:
sql (str|ClauseElement): SQL to execute
execution_options (dict): Execution options
Returns:
Result of SQL execution
"""
def begin_transaction(self):
"""
Begin transaction context for migration.
Returns:
Transaction context manager
"""
# Properties
bind: Connection # Database connection
dialect: Dialect # SQLAlchemy dialect
script: ScriptDirectory # Script directory
opts: Dict[str, Any] # Configuration options
environment_context: EnvironmentContext # Environment contextEnvironment Context
High-level context for managing migration environment configuration and execution.
class EnvironmentContext:
def __init__(self, config, script, **kw):
"""
Initialize environment context.
Args:
config (Config): Alembic configuration
script (ScriptDirectory): Script directory
**kw: Additional context options
"""
def configure(self, connection=None, url=None, dialect_name=None, transactional_ddl=None, output_buffer=None, starting_rev=None, tag=None, template_args=None, render_as_batch=None, target_metadata=None, include_name=None, include_object=None, include_schemas=None, process_revision_directives=None, compare_type=None, compare_server_default=None, render_item=None, literal_binds=None, upgrade_token=None, downgrade_token=None, alembic_module_prefix=None, sqlalchemy_module_prefix=None, user_module_prefix=None, **kw):
"""
Configure the environment context for migrations.
Args:
connection (Connection): Database connection
url (str): Database URL
dialect_name (str): Dialect name
transactional_ddl (bool): Use transactional DDL
output_buffer (StringIO): Output buffer for offline mode
starting_rev (str): Starting revision
tag (str): Migration tag
template_args (dict): Template variables
render_as_batch (bool): Render as batch operations
target_metadata (MetaData): Target metadata for comparison
include_name (callable): Name filtering function
include_object (callable): Object filtering function
include_schemas (bool): Include schema information
process_revision_directives (callable): Custom revision processing
compare_type (bool): Enable type comparison
compare_server_default (bool): Enable server default comparison
render_item (callable): Custom rendering function
literal_binds (bool): Use literal parameter binds
upgrade_token (str): Upgrade operation token
downgrade_token (str): Downgrade operation token
alembic_module_prefix (str): Alembic module prefix
sqlalchemy_module_prefix (str): SQLAlchemy module prefix
user_module_prefix (str): User module prefix
**kw: Additional configuration options
"""
def run_migrations(self, **kw):
"""
Execute migration operations.
Args:
**kw: Additional migration arguments
"""
def get_context(self):
"""
Get the migration context.
Returns:
MigrationContext: Current migration context
"""
def get_bind(self):
"""
Get the database connection.
Returns:
Connection: Database connection
"""
# Properties
config: Config # Alembic configuration
script: ScriptDirectory # Script directory
context_opts: Dict[str, Any] # Context optionsMigration Tracking Classes
Classes for tracking migration steps and revision information.
class MigrationInfo:
def __init__(self, revision, is_upgrade, is_stamp):
"""
Information about a migration step.
Args:
revision (str): Revision identifier
is_upgrade (bool): Whether this is an upgrade
is_stamp (bool): Whether this is a stamp operation
"""
revision: str # Revision identifier
is_upgrade: bool # Upgrade direction
is_stamp: bool # Stamp operation
class MigrationStep:
def __init__(self, up_revision, down_revision, is_upgrade, migration_fn):
"""
A single migration step.
Args:
up_revision (str): Target revision
down_revision (str): Source revision
is_upgrade (bool): Migration direction
migration_fn (callable): Migration function
"""
up_revision: str # Target revision
down_revision: str # Source revision
is_upgrade: bool # Migration direction
migration_fn: Callable # Migration function
class RevisionStep:
def __init__(self, revision_map, revision, is_upgrade):
"""
A revision step in migration path.
Args:
revision_map (RevisionMap): Revision map
revision (Script): Revision script
is_upgrade (bool): Migration direction
"""
revision: Script # Revision script
is_upgrade: bool # Migration direction
doc: str # Revision message
class StampStep:
def __init__(self, from_, to_, is_upgrade):
"""
A stamp step that marks revision without migration.
Args:
from_ (str): Source revision
to_ (str): Target revision
is_upgrade (bool): Direction
"""
from_: str # Source revision
to_: str # Target revision
is_upgrade: bool # DirectionHead Maintenance
Utility for tracking and managing revision heads.
class HeadMaintainer:
def __init__(self, context, heads):
"""
Initialize head maintainer.
Args:
context (MigrationContext): Migration context
heads (tuple): Current head revisions
"""
def update_to_step(self, step):
"""
Update heads based on migration step.
Args:
step (MigrationStep): Step being executed
"""
def heads(self):
"""
Get current heads.
Returns:
tuple: Current head revisions
"""Transaction Proxy
Internal transaction management for migration contexts.
class _ProxyTransaction:
def __init__(self, migration_context):
"""
Initialize transaction proxy.
Args:
migration_context (MigrationContext): Migration context
"""
def rollback(self):
"""Rollback the transaction."""
def commit(self):
"""Commit the transaction."""
def __enter__(self):
"""Enter transaction context."""
return self
def __exit__(self, type_, value, traceback):
"""Exit transaction context."""Usage Patterns
Creating Migration Context
from alembic.runtime.migration import MigrationContext
from sqlalchemy import create_engine
engine = create_engine('postgresql://user:pass@localhost/db')
with engine.connect() as connection:
context = MigrationContext.configure(
connection=connection,
opts={
'target_metadata': target_metadata,
'compare_type': True,
'compare_server_default': True
}
)
# Get current revision
current = context.get_current_revision()
print(f"Current revision: {current}")
# Execute SQL
context.execute("SELECT version()")Environment Context Usage
from alembic.runtime.environment import EnvironmentContext
from alembic.config import Config
from alembic.script import ScriptDirectory
config = Config('alembic.ini')
script_dir = ScriptDirectory.from_config(config)
env_context = EnvironmentContext(config, script_dir)
def run_migrations():
engine = create_engine(config.get_main_option('sqlalchemy.url'))
with engine.connect() as connection:
env_context.configure(
connection=connection,
target_metadata=target_metadata
)
with env_context.begin_transaction():
env_context.run_migrations()Migration Step Processing
from alembic.runtime.migration import MigrationStep
def process_migration_steps(steps):
"""Process a sequence of migration steps."""
for step in steps:
print(f"{'Upgrading' if step.is_upgrade else 'Downgrading'} "
f"from {step.down_revision} to {step.up_revision}")
# Execute migration function
step.migration_fn()Integration Examples
Custom Migration Context
class CustomMigrationContext(MigrationContext):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.custom_state = {}
def execute(self, sql, execution_options=None):
# Add custom logging
print(f"Executing: {sql}")
return super().execute(sql, execution_options)
def custom_operation(self, data):
"""Custom migration operation."""
self.custom_state['last_operation'] = data
return self.execute(f"-- Custom: {data}")Advanced Environment Setup
def create_advanced_environment():
"""Create environment with advanced configuration."""
config = Config('alembic.ini')
script_dir = ScriptDirectory.from_config(config)
def run_migrations_with_custom_context():
engine = create_engine(config.get_main_option('sqlalchemy.url'))
with engine.connect() as connection:
# Create custom migration context
context = CustomMigrationContext.configure(
connection=connection,
opts={
'target_metadata': target_metadata,
'process_revision_directives': custom_directive_processor,
'include_object': custom_object_filter,
'compare_type': advanced_type_comparison
}
)
with context.begin_transaction():
context.run_migrations()
return run_migrations_with_custom_contextError Handling
Runtime components may raise:
RevisionError: Revision-related errorsEnvironmentError: Environment configuration errorsMigrationError: Migration execution errors- Database-specific exceptions during execution
Types
# Runtime component types
class MigrationContext:
bind: Connection
dialect: Dialect
script: ScriptDirectory
opts: Dict[str, Any]
environment_context: EnvironmentContext
class EnvironmentContext:
config: Config
script: ScriptDirectory
context_opts: Dict[str, Any]
# Migration step types
class MigrationInfo:
revision: str
is_upgrade: bool
is_stamp: bool
class MigrationStep:
up_revision: str
down_revision: str
is_upgrade: bool
migration_fn: Callable
# Function type annotations
MigrationFunction = Callable[[], None]
DirectiveProcessor = Callable[[MigrationContext, Tuple, List], None]
ObjectFilter = Callable[[Any, str, str, bool, Any], bool]Install with Tessl CLI
npx tessl i tessl/pypi-alembic