Alembic

import alembic
from alembic import context, op

from alembic.config import Config
from alembic import command

from alembic.runtime.environment import EnvironmentContext
from alembic.runtime.migration import MigrationContext

from alembic.config import Config
from alembic import command

# Initialize new migration environment
config = Config("alembic.ini")
command.init(config, "migrations")

# Create new migration revision
command.revision(config, message="Create user table", autogenerate=True)

# Run migration
command.upgrade(config, "head")

"""Create user table

Revision ID: 001
Create Date: 2023-01-01 10:00:00.000000

"""
from alembic import op
import sqlalchemy as sa

# revision identifiers
revision = '001'
down_revision = None

def upgrade():
    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)
    )

def downgrade():
    op.drop_table('user')

# Key command functions
def init(config, directory, template='generic', package=False): ...
def revision(config, message=None, autogenerate=False, sql=False, head='head', splice=False, branch_label=None, version_path=None, rev_id=None, depends_on=None, process_revision_directives=None): ...
def upgrade(config, revision, sql=False, tag=None): ...
def downgrade(config, revision, sql=False, tag=None): ...
def current(config, verbose=False): ...
def history(config, rev_range=None, verbose=False, indicate_current=False): ...
def check(config): ...  # v1.9.0+
def ensure_version(config, sql=False): ...  # v1.7.6+
def merge(config, revisions, message=None, branch_label=None, rev_id=None): ...
def stamp(config, revision, sql=False, tag=None, purge=False): ...

# Core operation functions
def create_table(table_name, *columns, if_not_exists=None, **kw): ...
def drop_table(table_name, if_exists=None, **kw): ...  # v1.16.0+
def add_column(table_name, column, if_not_exists=None, **kw): ...  # v1.16.0+
def drop_column(table_name, column_name, if_exists=None, **kw): ...  # v1.16.0+
def alter_column(table_name, column_name, **kw): ...
def create_index(index_name, table_name, columns, if_not_exists=None, **kw): ...  # v1.12.0+
def drop_index(index_name, table_name=None, if_exists=None, **kw): ...  # v1.12.0+
def create_foreign_key(constraint_name, source_table, referent_table, local_cols, remote_cols, **kw): ...
def drop_constraint(constraint_name, table_name, type_=None, if_exists=None): ...  # v1.16.0+
def batch_alter_table(table_name, schema=None, **kw): ...  # Context manager
def run_async(async_function, *args, **kw_args): ...  # v1.11+
def f(name): ...  # Naming convention support

# Core context functions
def configure(connection=None, url=None, dialect_name=None, **kw): ...
def get_bind(): ...
def get_context(): ...
def execute(sql, execution_options=None): ...
def is_offline_mode(): ...
def is_transactional_ddl(): ...
def run_migrations(): ...
def begin_transaction(): ...  # Context manager
def get_head_revision(): ...
def get_head_revisions(): ...
def get_revision_argument(): ...
def get_starting_revision_argument(): ...
def get_tag_argument(): ...
def get_x_argument(as_dictionary=False): ...
def static_output(text): ...  # Offline mode output

class Config:
    def __init__(self, file_=None, toml_file=None, ini_section='alembic', output_buffer=None, stdout=sys.stdout, cmd_opts=None, config_args=None, attributes=None): ...
    def get_main_option(self, name, default=None): ...
    def set_main_option(self, name, value): ...
    def get_section_option(self, section, name, default=None): ...
    def set_section_option(self, section, name, value): ...
    def get_section(self, name, default=None): ...
    def get_alembic_option(self, name, default=None): ...  # v1.16.0+
    def get_alembic_boolean_option(self, name): ...  # v1.16.0+
    def get_version_locations_list(self): ...
    def get_hooks_list(self): ...
    def print_stdout(self, text, *arg): ...

def compare_metadata(context, metadata): ...
def produce_migrations(context, metadata): ...
def render_python_code(up_ops, down_ops, migration_context, **kwargs): ...
def render_op_text(autogen_context, op): ...

# Customization support
class AutogenContext: ...
class RevisionContext: ...
class Rewriter: ...  # Operation rewriting

# Registries for extensibility
comparators: Registry  # Custom comparison functions
renderers: Registry    # Custom rendering functions

class ScriptDirectory:
    def __init__(self, dir, file_template=None, truncate_slug_length=40, version_locations=None, sourceless=False, output_encoding='utf-8', timezone=None, hooks=None, recursive_version_locations=False, messaging_opts=None): ...
    def from_config(cls, config): ...  # Class method
    def generate_revision(self, revid, message, **kw): ...
    def get_revision(self, id_): ...
    def get_current_head(self): ...
    def get_heads(self): ...
    def get_base(self): ...
    def walk_revisions(self, head='heads', base='base'): ...
    def iterate_revisions(self, upper, lower): ...
    def run_env(self): ...

class MigrationContext:
    def configure(cls, connection=None, url=None, **opts): ...  # Class method
    def get_current_revision(self): ...
    def get_current_heads(self): ...
    def execute(self, sql, execution_options=None): ...
    def stamp(self, script_directory, revision): ...
    def run_migrations(self): ...

class EnvironmentContext:
    def configure(self, connection=None, target_metadata=None, **kw): ...
    def run_migrations(self, **kw): ...
    # Properties
    config: Config
    script: ScriptDirectory

# alembic/env.py template structure
from alembic import context
from sqlalchemy import engine_from_config

config = context.config
target_metadata = None

def run_migrations_offline():
    context.configure(url=url, target_metadata=target_metadata)
    with context.begin_transaction():
        context.run_migrations()

def run_migrations_online():
    connectable = engine_from_config(config.get_section(config.config_ini_section))
    with connectable.connect() as connection:
        context.configure(connection=connection, target_metadata=target_metadata)
        with context.begin_transaction():
            context.run_migrations()

def upgrade():
    with op.batch_alter_table("user") as batch_op:
        batch_op.add_column(sa.Column('created_at', sa.DateTime))
        batch_op.alter_column('name', nullable=True)

# Type imports
from typing import Union, List, Optional, Set, Any, Dict, Tuple, Callable, Iterator, ContextManager
from sqlalchemy import Connection, Dialect, Table, Column
from sqlalchemy.sql.schema import Constraint, Index
from argparse import Namespace

# Type aliases
_RevIdType = Union[str, List[str], Tuple[str, ...]]
ProcessRevisionDirectiveFn = Callable[..., Any]

# Configuration types
class Config:
    cmd_opts: Optional[Namespace]
    config_args: Dict[str, Any]
    config_file_name: Optional[str]
    toml_file_name: Optional[str]  # v1.16.0+
    attributes: Dict[str, Any]
    
# Context types  
class MigrationContext:
    bind: Connection
    dialect: Dialect
    
class EnvironmentContext:
    config: Config
    script: ScriptDirectory

# Script types
class Script:
    revision: str
    down_revision: Optional[str]
    branch_labels: Optional[Set[str]]
    depends_on: Optional[Set[str]]
    path: str

class ScriptDirectory:
    dir: str
    versions: str
    revision_map: RevisionMap

# Operation types
class Operations:
    migration_context: MigrationContext
    
class BatchOperations(Operations):
    table: Table

# Messaging configuration (v1.15.3+)
class MessagingOptions:
    file_template: str
    truncate_slug_length: int
    timezone: Optional[str]
    
# Post-write hook configuration
class PostWriteHookConfig:
    type: str
    entrypoint: str
    options: Dict[str, Any]

# Exception types
class CommandError(Exception): ...
class AutogenerateDiffsDetected(Exception): ...
class RevisionError(Exception): ...