Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified
📁 ..	-	2026-03-27 02:38
📄 __init__.py import sys from . import context from . import op __version__ = "1.11.1"	75 B	2026-01-20 12:58
📄 __main__.py from .config import main if __name__ == "__main__": main(prog="alembic")	78 B	2026-01-20 12:58
📁 __pycache__	-	2026-03-21 15:45
📁 autogenerate	-	2026-03-21 15:45
📄 command.py from __future__ import annotations import os from typing import List from typing import Optional from typing import TYPE_CHECKING from typing import Union from . import autogenerate as autogen from . import util from .runtime.environment import EnvironmentContext from .script import ScriptDirectory if TYPE_CHECKING: from alembic.config import Config from alembic.script.base import Script from .runtime.environment import ProcessRevisionDirectiveFn def list_templates(config: Config): """List available templates. :param config: a :class:`.Config` object. """ config.print_stdout("Available templates:\n") for tempname in os.listdir(config.get_template_directory()): with open( os.path.join(config.get_template_directory(), tempname, "README") ) as readme: synopsis = next(readme).rstrip() config.print_stdout("%s - %s", tempname, synopsis) config.print_stdout("\nTemplates are used via the 'init' command, e.g.:") config.print_stdout("\n alembic init --template generic ./scripts") def init( config: Config, directory: str, template: str = "generic", package: bool = False, ) -> None: """Initialize a new scripts directory. :param config: a :class:`.Config` object. :param directory: string path of the target directory :param template: string name of the migration environment template to use. :param package: when True, write ``__init__.py`` files into the environment location as well as the versions/ location. .. versionadded:: 1.2 """ if os.access(directory, os.F_OK) and os.listdir(directory): raise util.CommandError( "Directory %s already exists and is not empty" % directory ) template_dir = os.path.join(config.get_template_directory(), template) if not os.access(template_dir, os.F_OK): raise util.CommandError("No such template %r" % template) if not os.access(directory, os.F_OK): with util.status( f"Creating directory {os.path.abspath(directory)!r}", config.messaging_opts, ): os.makedirs(directory) versions = os.path.join(directory, "versions") with util.status( f"Creating directory {os.path.abspath(versions)!r}", config.messaging_opts, ): os.makedirs(versions) script = ScriptDirectory(directory) config_file: str \| None = None for file_ in os.listdir(template_dir): file_path = os.path.join(template_dir, file_) if file_ == "alembic.ini.mako": assert config.config_file_name is not None config_file = os.path.abspath(config.config_file_name) if os.access(config_file, os.F_OK): util.msg( f"File {config_file!r} already exists, skipping", config.messaging_opts, ) else: script._generate_template( file_path, config_file, script_location=directory ) elif os.path.isfile(file_path): output_file = os.path.join(directory, file_) script._copy_file(file_path, output_file) if package: for path in [ os.path.join(os.path.abspath(directory), "__init__.py"), os.path.join(os.path.abspath(versions), "__init__.py"), ]: with util.status("Adding {path!r}", config.messaging_opts): with open(path, "w"): pass assert config_file is not None util.msg( "Please edit configuration/connection/logging " f"settings in {config_file!r} before proceeding.", *config.messaging_opts, ) def revision( config: Config, message: Optional[str] = None, autogenerate: bool = False, sql: bool = False, head: str = "head", splice: bool = False, branch_label: Optional[str] = None, version_path: Optional[str] = None, rev_id: Optional[str] = None, depends_on: Optional[str] = None, process_revision_directives: Optional[ProcessRevisionDirectiveFn] = None, ) -> Union[Optional[Script], List[Optional[Script]]]: """Create a new revision file. :param config: a :class:`.Config` object. :param message: string message to apply to the revision; this is the ``-m`` option to ``alembic revision``. :param autogenerate: whether or not to autogenerate the script from the database; this is the ``--autogenerate`` option to ``alembic revision``. :param sql: whether to dump the script out as a SQL string; when specified, the script is dumped to stdout. This is the ``--sql`` option to ``alembic revision``. :param head: head revision to build the new revision upon as a parent; this is the ``--head`` option to ``alembic revision``. :param splice: whether or not the new revision should be made into a new head of its own; is required when the given ``head`` is not itself a head. This is the ``--splice`` option to ``alembic revision``. :param branch_label: string label to apply to the branch; this is the ``--branch-label`` option to ``alembic revision``. :param version_path: string symbol identifying a specific version path from the configuration; this is the ``--version-path`` option to ``alembic revision``. :param rev_id: optional revision identifier to use instead of having one generated; this is the ``--rev-id`` option to ``alembic revision``. :param depends_on: optional list of "depends on" identifiers; this is the ``--depends-on`` option to ``alembic revision``. :param process_revision_directives: this is a callable that takes the same form as the callable described at :paramref:`.EnvironmentContext.configure.process_revision_directives`; will be applied to the structure generated by the revision process where it can be altered programmatically. Note that unlike all the other parameters, this option is only available via programmatic use of :func:`.command.revision` """ script_directory = ScriptDirectory.from_config(config) command_args = dict( message=message, autogenerate=autogenerate, sql=sql, head=head, splice=splice, branch_label=branch_label, version_path=version_path, rev_id=rev_id, depends_on=depends_on, ) revision_context = autogen.RevisionContext( config, script_directory, command_args, process_revision_directives=process_revision_directives, ) environment = util.asbool(config.get_main_option("revision_environment")) if autogenerate: environment = True if sql: raise util.CommandError( "Using --sql with --autogenerate does not make any sense" ) def retrieve_migrations(rev, context): revision_context.run_autogenerate(rev, context) return [] elif environment: def retrieve_migrations(rev, context): revision_context.run_no_autogenerate(rev, context) return [] elif sql: raise util.CommandError( "Using --sql with the revision command when " "revision_environment is not configured does not make any sense" ) if environment: with EnvironmentContext( config, script_directory, fn=retrieve_migrations, as_sql=sql, template_args=revision_context.template_args, revision_context=revision_context, ): script_directory.run_env() # the revision_context now has MigrationScript structure(s) present. # these could theoretically be further processed / rewritten here, # in addition to the hooks present within each run_migrations() call, # or at the end of env.py run_migrations_online(). scripts = [script for script in revision_context.generate_scripts()] if len(scripts) == 1: return scripts[0] else: return scripts def check( config: "Config", ) -> None: """Check if revision command with autogenerate has pending upgrade ops. :param config: a :class:`.Config` object. .. versionadded:: 1.9.0 """ script_directory = ScriptDirectory.from_config(config) command_args = dict( message=None, autogenerate=True, sql=False, head="head", splice=False, branch_label=None, version_path=None, rev_id=None, depends_on=None, ) revision_context = autogen.RevisionContext( config, script_directory, command_args, ) def retrieve_migrations(rev, context): revision_context.run_autogenerate(rev, context) return [] with EnvironmentContext( config, script_directory, fn=retrieve_migrations, as_sql=False, template_args=revision_context.template_args, revision_context=revision_context, ): script_directory.run_env() # the revision_context now has MigrationScript structure(s) present. migration_script = revision_context.generated_revisions[-1] diffs = migration_script.upgrade_ops.as_diffs() if diffs: raise util.AutogenerateDiffsDetected( f"New upgrade operations detected: {diffs}" ) else: config.print_stdout("No new upgrade operations detected.") def merge( config: Config, revisions: str, message: Optional[str] = None, branch_label: Optional[str] = None, rev_id: Optional[str] = None, ) -> Optional[Script]: """Merge two revisions together. Creates a new migration file. :param config: a :class:`.Config` instance :param message: string message to apply to the revision :param branch_label: string label name to apply to the new revision :param rev_id: hardcoded revision identifier instead of generating a new one. .. seealso:: :ref:`branches` """ script = ScriptDirectory.from_config(config) template_args = { "config": config # Let templates use config for # e.g. multiple databases } return script.generate_revision( rev_id or util.rev_id(), message, refresh=True, head=revisions, branch_labels=branch_label, template_args, # type:ignore[arg-type] ) def upgrade( config: Config, revision: str, sql: bool = False, tag: Optional[str] = None, ) -> None: """Upgrade to a later version. :param config: a :class:`.Config` instance. :param revision: string revision target or range for --sql mode :param sql: if True, use ``--sql`` mode :param tag: an arbitrary "tag" that can be intercepted by custom ``env.py`` scripts via the :meth:`.EnvironmentContext.get_tag_argument` method. """ script = ScriptDirectory.from_config(config) starting_rev = None if ":" in revision: if not sql: raise util.CommandError("Range revision not allowed") starting_rev, revision = revision.split(":", 2) def upgrade(rev, context): return script._upgrade_revs(revision, rev) with EnvironmentContext( config, script, fn=upgrade, as_sql=sql, starting_rev=starting_rev, destination_rev=revision, tag=tag, ): script.run_env() def downgrade( config: Config, revision: str, sql: bool = False, tag: Optional[str] = None, ) -> None: """Revert to a previous version. :param config: a :class:`.Config` instance. :param revision: string revision target or range for --sql mode :param sql: if True, use ``--sql`` mode :param tag: an arbitrary "tag" that can be intercepted by custom ``env.py`` scripts via the :meth:`.EnvironmentContext.get_tag_argument` method. """ script = ScriptDirectory.from_config(config) starting_rev = None if ":" in revision: if not sql: raise util.CommandError("Range revision not allowed") starting_rev, revision = revision.split(":", 2) elif sql: raise util.CommandError( "downgrade with --sql requires <fromrev>:<torev>" ) def downgrade(rev, context): return script._downgrade_revs(revision, rev) with EnvironmentContext( config, script, fn=downgrade, as_sql=sql, starting_rev=starting_rev, destination_rev=revision, tag=tag, ): script.run_env() def show(config, rev): """Show the revision(s) denoted by the given symbol. :param config: a :class:`.Config` instance. :param revision: string revision target """ script = ScriptDirectory.from_config(config) if rev == "current": def show_current(rev, context): for sc in script.get_revisions(rev): config.print_stdout(sc.log_entry) return [] with EnvironmentContext(config, script, fn=show_current): script.run_env() else: for sc in script.get_revisions(rev): config.print_stdout(sc.log_entry) def history( config: Config, rev_range: Optional[str] = None, verbose: bool = False, indicate_current: bool = False, ) -> None: """List changeset scripts in chronological order. :param config: a :class:`.Config` instance. :param rev_range: string revision range :param verbose: output in verbose mode. :param indicate_current: indicate current revision. """ base: Optional[str] head: Optional[str] script = ScriptDirectory.from_config(config) if rev_range is not None: if ":" not in rev_range: raise util.CommandError( "History range requires [start]:[end], " "[start]:, or :[end]" ) base, head = rev_range.strip().split(":") else: base = head = None environment = ( util.asbool(config.get_main_option("revision_environment")) or indicate_current ) def _display_history(config, script, base, head, currents=()): for sc in script.walk_revisions( base=base or "base", head=head or "heads" ): if indicate_current: sc._db_current_indicator = sc.revision in currents config.print_stdout( sc.cmd_format( verbose=verbose, include_branches=True, include_doc=True, include_parents=True, ) ) def _display_history_w_current(config, script, base, head): def _display_current_history(rev, context): if head == "current": _display_history(config, script, base, rev, rev) elif base == "current": _display_history(config, script, rev, head, rev) else: _display_history(config, script, base, head, rev) return [] with EnvironmentContext(config, script, fn=_display_current_history): script.run_env() if base == "current" or head == "current" or environment: _display_history_w_current(config, script, base, head) else: _display_history(config, script, base, head) def heads(config, verbose=False, resolve_dependencies=False): """Show current available heads in the script directory. :param config: a :class:`.Config` instance. :param verbose: output in verbose mode. :param resolve_dependencies: treat dependency version as down revisions. """ script = ScriptDirectory.from_config(config) if resolve_dependencies: heads = script.get_revisions("heads") else: heads = script.get_revisions(script.get_heads()) for rev in heads: config.print_stdout( rev.cmd_format( verbose, include_branches=True, tree_indicators=False ) ) def branches(config, verbose=False): """Show current branch points. :param config: a :class:`.Config` instance. :param verbose: output in verbose mode. """ script = ScriptDirectory.from_config(config) for sc in script.walk_revisions(): if sc.is_branch_point: config.print_stdout( "%s\n%s\n", sc.cmd_format(verbose, include_branches=True), "\n".join( "%s -> %s" % ( " " len(str(sc.revision)), rev_obj.cmd_format( False, include_branches=True, include_doc=verbose ), ) for rev_obj in ( script.get_revision(rev) for rev in sc.nextrev ) ), ) def current(config: Config, verbose: bool = False) -> None: """Display the current revision for a database. :param config: a :class:`.Config` instance. :param verbose: output in verbose mode. """ script = ScriptDirectory.from_config(config) def display_version(rev, context): if verbose: config.print_stdout( "Current revision(s) for %s:", util.obfuscate_url_pw(context.connection.engine.url), ) for rev in script.get_all_current(rev): config.print_stdout(rev.cmd_format(verbose)) return [] with EnvironmentContext( config, script, fn=display_version, dont_mutate=True ): script.run_env() def stamp( config: Config, revision: str, sql: bool = False, tag: Optional[str] = None, purge: bool = False, ) -> None: """'stamp' the revision table with the given revision; don't run any migrations. :param config: a :class:`.Config` instance. :param revision: target revision or list of revisions. May be a list to indicate stamping of multiple branch heads. .. note:: this parameter is called "revisions" in the command line interface. .. versionchanged:: 1.2 The revision may be a single revision or list of revisions when stamping multiple branch heads. :param sql: use ``--sql`` mode :param tag: an arbitrary "tag" that can be intercepted by custom ``env.py`` scripts via the :class:`.EnvironmentContext.get_tag_argument` method. :param purge: delete all entries in the version table before stamping. .. versionadded:: 1.2 """ script = ScriptDirectory.from_config(config) if sql: destination_revs = [] starting_rev = None for _revision in util.to_list(revision): if ":" in _revision: srev, _revision = _revision.split(":", 2) if starting_rev != srev: if starting_rev is None: starting_rev = srev else: raise util.CommandError( "Stamp operation with --sql only supports a " "single starting revision at a time" ) destination_revs.append(_revision) else: destination_revs = util.to_list(revision) def do_stamp(rev, context): return script._stamp_revs(util.to_tuple(destination_revs), rev) with EnvironmentContext( config, script, fn=do_stamp, as_sql=sql, starting_rev=starting_rev if sql else None, destination_rev=util.to_tuple(destination_revs), tag=tag, purge=purge, ): script.run_env() def edit(config: Config, rev: str) -> None: """Edit revision script(s) using $EDITOR. :param config: a :class:`.Config` instance. :param rev: target revision. """ script = ScriptDirectory.from_config(config) if rev == "current": def edit_current(rev, context): if not rev: raise util.CommandError("No current revisions") for sc in script.get_revisions(rev): util.open_in_editor(sc.path) return [] with EnvironmentContext(config, script, fn=edit_current): script.run_env() else: revs = script.get_revisions(rev) if not revs: raise util.CommandError( "No revision files indicated by symbol '%s'" % rev ) for sc in revs: assert sc util.open_in_editor(sc.path) def ensure_version(config: Config, sql: bool = False) -> None: """Create the alembic version table if it doesn't exist already . :param config: a :class:`.Config` instance. :param sql: use ``--sql`` mode .. versionadded:: 1.7.6 """ script = ScriptDirectory.from_config(config) def do_ensure_version(rev, context): context._ensure_version_table() return [] with EnvironmentContext( config, script, fn=do_ensure_version, as_sql=sql, ): script.run_env()	20.84 KB	2026-01-20 12:58
📄 config.py from __future__ import annotations from argparse import ArgumentParser from argparse import Namespace from configparser import ConfigParser import inspect import os import sys from typing import Any from typing import cast from typing import Dict from typing import Mapping from typing import Optional from typing import overload from typing import TextIO from typing import Union from typing_extensions import TypedDict from . import __version__ from . import command from . import util from .util import compat class Config: r"""Represent an Alembic configuration. Within an ``env.py`` script, this is available via the :attr:`.EnvironmentContext.config` attribute, which in turn is available at ``alembic.context``:: from alembic import context some_param = context.config.get_main_option("my option") When invoking Alembic programatically, a new :class:`.Config` can be created by passing the name of an .ini file to the constructor:: from alembic.config import Config alembic_cfg = Config("/path/to/yourapp/alembic.ini") With a :class:`.Config` object, you can then run Alembic commands programmatically using the directives in :mod:`alembic.command`. The :class:`.Config` object can also be constructed without a filename. Values can be set programmatically, and new sections will be created as needed:: from alembic.config import Config alembic_cfg = Config() alembic_cfg.set_main_option("script_location", "myapp:migrations") alembic_cfg.set_main_option("sqlalchemy.url", "postgresql://foo/bar") alembic_cfg.set_section_option("mysection", "foo", "bar") .. warning:: When using programmatic configuration, make sure the ``env.py`` file in use is compatible with the target configuration; including that the call to Python ``logging.fileConfig()`` is omitted if the programmatic configuration doesn't actually include logging directives. For passing non-string values to environments, such as connections and engines, use the :attr:`.Config.attributes` dictionary:: with engine.begin() as connection: alembic_cfg.attributes['connection'] = connection command.upgrade(alembic_cfg, "head") :param file\_: name of the .ini file to open. :param ini_section: name of the main Alembic section within the .ini file :param output_buffer: optional file-like input buffer which will be passed to the :class:`.MigrationContext` - used to redirect the output of "offline generation" when using Alembic programmatically. :param stdout: buffer where the "print" output of commands will be sent. Defaults to ``sys.stdout``. :param config_args: A dictionary of keys and values that will be used for substitution in the alembic config file. The dictionary as given is copied to a new one, stored locally as the attribute ``.config_args``. When the :attr:`.Config.file_config` attribute is first invoked, the replacement variable ``here`` will be added to this dictionary before the dictionary is passed to ``ConfigParser()`` to parse the .ini file. :param attributes: optional dictionary of arbitrary Python keys/values, which will be populated into the :attr:`.Config.attributes` dictionary. .. seealso:: :ref:`connection_sharing` """ def __init__( self, file_: Union[str, os.PathLike[str], None] = None, ini_section: str = "alembic", output_buffer: Optional[TextIO] = None, stdout: TextIO = sys.stdout, cmd_opts: Optional[Namespace] = None, config_args: Mapping[str, Any] = util.immutabledict(), attributes: Optional[dict] = None, ) -> None: """Construct a new :class:`.Config`""" self.config_file_name = file_ self.config_ini_section = ini_section self.output_buffer = output_buffer self.stdout = stdout self.cmd_opts = cmd_opts self.config_args = dict(config_args) if attributes: self.attributes.update(attributes) cmd_opts: Optional[Namespace] = None """The command-line options passed to the ``alembic`` script. Within an ``env.py`` script this can be accessed via the :attr:`.EnvironmentContext.config` attribute. .. seealso:: :meth:`.EnvironmentContext.get_x_argument` """ config_file_name: Union[str, os.PathLike[str], None] = None """Filesystem path to the .ini file in use.""" config_ini_section: str = None # type:ignore[assignment] """Name of the config file section to read basic configuration from. Defaults to ``alembic``, that is the ``[alembic]`` section of the .ini file. This value is modified using the ``-n/--name`` option to the Alembic runner. """ @util.memoized_property def attributes(self): """A Python dictionary for storage of additional state. This is a utility dictionary which can include not just strings but engines, connections, schema objects, or anything else. Use this to pass objects into an env.py script, such as passing a :class:`sqlalchemy.engine.base.Connection` when calling commands from :mod:`alembic.command` programmatically. .. seealso:: :ref:`connection_sharing` :paramref:`.Config.attributes` """ return {} def print_stdout(self, text: str, arg) -> None: """Render a message to standard out. When :meth:`.Config.print_stdout` is called with additional args those arguments will formatted against the provided text, otherwise we simply output the provided text verbatim. This is a no-op when the``quiet`` messaging option is enabled. e.g.:: >>> config.print_stdout('Some text %s', 'arg') Some Text arg """ if arg: output = str(text) % arg else: output = str(text) util.write_outstream(self.stdout, output, "\n", self.messaging_opts) @util.memoized_property def file_config(self): """Return the underlying ``ConfigParser`` object. Direct access to the .ini file is available here, though the :meth:`.Config.get_section` and :meth:`.Config.get_main_option` methods provide a possibly simpler interface. """ if self.config_file_name: here = os.path.abspath(os.path.dirname(self.config_file_name)) else: here = "" self.config_args["here"] = here file_config = ConfigParser(self.config_args) if self.config_file_name: file_config.read([self.config_file_name]) else: file_config.add_section(self.config_ini_section) return file_config def get_template_directory(self) -> str: """Return the directory where Alembic setup templates are found. This method is used by the alembic ``init`` and ``list_templates`` commands. """ import alembic package_dir = os.path.abspath(os.path.dirname(alembic.__file__)) return os.path.join(package_dir, "templates") @overload def get_section( self, name: str, default: None = ... ) -> Optional[Dict[str, str]]: ... # "default" here could also be a TypeVar # _MT = TypeVar("_MT", bound=Mapping[str, str]), # however mypy wasn't handling that correctly (pyright was) @overload def get_section( self, name: str, default: Dict[str, str] ) -> Dict[str, str]: ... @overload def get_section( self, name: str, default: Mapping[str, str] ) -> Union[Dict[str, str], Mapping[str, str]]: ... def get_section( self, name: str, default: Optional[Mapping[str, str]] = None ) -> Optional[Mapping[str, str]]: """Return all the configuration options from a given .ini file section as a dictionary. If the given section does not exist, the value of ``default`` is returned, which is expected to be a dictionary or other mapping. """ if not self.file_config.has_section(name): return default return dict(self.file_config.items(name)) def set_main_option(self, name: str, value: str) -> None: """Set an option programmatically within the 'main' section. This overrides whatever was in the .ini file. :param name: name of the value :param value: the value. Note that this value is passed to ``ConfigParser.set``, which supports variable interpolation using pyformat (e.g. ``%(some_value)s``). A raw percent sign not part of an interpolation symbol must therefore be escaped, e.g. ``%%``. The given value may refer to another value already in the file using the interpolation format. """ self.set_section_option(self.config_ini_section, name, value) def remove_main_option(self, name: str) -> None: self.file_config.remove_option(self.config_ini_section, name) def set_section_option(self, section: str, name: str, value: str) -> None: """Set an option programmatically within the given section. The section is created if it doesn't exist already. The value here will override whatever was in the .ini file. :param section: name of the section :param name: name of the value :param value: the value. Note that this value is passed to ``ConfigParser.set``, which supports variable interpolation using pyformat (e.g. ``%(some_value)s``). A raw percent sign not part of an interpolation symbol must therefore be escaped, e.g. ``%%``. The given value may refer to another value already in the file using the interpolation format. """ if not self.file_config.has_section(section): self.file_config.add_section(section) self.file_config.set(section, name, value) def get_section_option( self, section: str, name: str, default: Optional[str] = None ) -> Optional[str]: """Return an option from the given section of the .ini file.""" if not self.file_config.has_section(section): raise util.CommandError( "No config file %r found, or file has no " "'[%s]' section" % (self.config_file_name, section) ) if self.file_config.has_option(section, name): return self.file_config.get(section, name) else: return default @overload def get_main_option(self, name: str, default: str) -> str: ... @overload def get_main_option( self, name: str, default: Optional[str] = None ) -> Optional[str]: ... def get_main_option(self, name, default=None): """Return an option from the 'main' section of the .ini file. This defaults to being a key from the ``[alembic]`` section, unless the ``-n/--name`` flag were used to indicate a different section. """ return self.get_section_option(self.config_ini_section, name, default) @util.memoized_property def messaging_opts(self) -> MessagingOptions: """The messaging options.""" return cast( MessagingOptions, util.immutabledict( {"quiet": getattr(self.cmd_opts, "quiet", False)} ), ) class MessagingOptions(TypedDict, total=False): quiet: bool class CommandLine: def __init__(self, prog: Optional[str] = None) -> None: self._generate_args(prog) def _generate_args(self, prog: Optional[str]) -> None: def add_options(fn, parser, positional, kwargs): kwargs_opts = { "template": ( "-t", "--template", dict( default="generic", type=str, help="Setup template for use with 'init'", ), ), "message": ( "-m", "--message", dict( type=str, help="Message string to use with 'revision'" ), ), "sql": ( "--sql", dict( action="store_true", help="Don't emit SQL to database - dump to " "standard output/file instead. See docs on " "offline mode.", ), ), "tag": ( "--tag", dict( type=str, help="Arbitrary 'tag' name - can be used by " "custom env.py scripts.", ), ), "head": ( "--head", dict( type=str, help="Specify head revision or <branchname>@head " "to base new revision on.", ), ), "splice": ( "--splice", dict( action="store_true", help="Allow a non-head revision as the " "'head' to splice onto", ), ), "depends_on": ( "--depends-on", dict( action="append", help="Specify one or more revision identifiers " "which this revision should depend on.", ), ), "rev_id": ( "--rev-id", dict( type=str, help="Specify a hardcoded revision id instead of " "generating one", ), ), "version_path": ( "--version-path", dict( type=str, help="Specify specific path from config for " "version file", ), ), "branch_label": ( "--branch-label", dict( type=str, help="Specify a branch label to apply to the " "new revision", ), ), "verbose": ( "-v", "--verbose", dict(action="store_true", help="Use more verbose output"), ), "resolve_dependencies": ( "--resolve-dependencies", dict( action="store_true", help="Treat dependency versions as down revisions", ), ), "autogenerate": ( "--autogenerate", dict( action="store_true", help="Populate revision script with candidate " "migration operations, based on comparison " "of database to model.", ), ), "rev_range": ( "-r", "--rev-range", dict( action="store", help="Specify a revision range; " "format is [start]:[end]", ), ), "indicate_current": ( "-i", "--indicate-current", dict( action="store_true", help="Indicate the current revision", ), ), "purge": ( "--purge", dict( action="store_true", help="Unconditionally erase the version table " "before stamping", ), ), "package": ( "--package", dict( action="store_true", help="Write empty __init__.py files to the " "environment and version locations", ), ), } positional_help = { "directory": "location of scripts directory", "revision": "revision identifier", "revisions": "one or more revisions, or 'heads' for all heads", } for arg in kwargs: if arg in kwargs_opts: args = kwargs_opts[arg] args, kw = args[0:-1], args[-1] parser.add_argument(args, *kw) for arg in positional: if ( arg == "revisions" or fn in positional_translations and positional_translations[fn][arg] == "revisions" ): subparser.add_argument( "revisions", nargs="+", help=positional_help.get("revisions"), ) else: subparser.add_argument(arg, help=positional_help.get(arg)) parser = ArgumentParser(prog=prog) parser.add_argument( "--version", action="version", version="%%(prog)s %s" % __version__ ) parser.add_argument( "-c", "--config", type=str, default=os.environ.get("ALEMBIC_CONFIG", "alembic.ini"), help="Alternate config file; defaults to value of " 'ALEMBIC_CONFIG environment variable, or "alembic.ini"', ) parser.add_argument( "-n", "--name", type=str, default="alembic", help="Name of section in .ini file to " "use for Alembic config", ) parser.add_argument( "-x", action="append", help="Additional arguments consumed by " "custom env.py scripts, e.g. -x " "setting1=somesetting -x setting2=somesetting", ) parser.add_argument( "--raiseerr", action="store_true", help="Raise a full stack trace on error", ) parser.add_argument( "-q", "--quiet", action="store_true", help="Do not log to std output.", ) subparsers = parser.add_subparsers() positional_translations = {command.stamp: {"revision": "revisions"}} for fn in [getattr(command, n) for n in dir(command)]: if ( inspect.isfunction(fn) and fn.__name__[0] != "_" and fn.__module__ == "alembic.command" ): spec = compat.inspect_getfullargspec(fn) if spec[3] is not None: positional = spec[0][1 : -len(spec[3])] kwarg = spec[0][-len(spec[3]) :] else: positional = spec[0][1:] kwarg = [] if fn in positional_translations: positional = [ positional_translations[fn].get(name, name) for name in positional ] # parse first line(s) of helptext without a line break help_ = fn.__doc__ if help_: help_text = [] for line in help_.split("\n"): if not line.strip(): break else: help_text.append(line.strip()) else: help_text = [] subparser = subparsers.add_parser( fn.__name__, help=" ".join(help_text) ) add_options(fn, subparser, positional, kwarg) subparser.set_defaults(cmd=(fn, positional, kwarg)) self.parser = parser def run_cmd(self, config: Config, options: Namespace) -> None: fn, positional, kwarg = options.cmd try: fn( config, [getattr(options, k, None) for k in positional], {k: getattr(options, k, None) for k in kwarg}, ) except util.CommandError as e: if options.raiseerr: raise else: util.err(str(e), config.messaging_opts) def main(self, argv=None): options = self.parser.parse_args(argv) if not hasattr(options, "cmd"): # see http://bugs.python.org/issue9253, argparse # behavior changed incompatibly in py3.3 self.parser.error("too few arguments") else: cfg = Config( file_=options.config, ini_section=options.name, cmd_opts=options, ) self.run_cmd(cfg, options) def main(argv=None, prog=None, **kwargs): """The console runner function for Alembic.""" CommandLine(prog=prog).main(argv=argv) if __name__ == "__main__": main()	21.4 KB	2026-01-20 12:58
📄 context.py from .runtime.environment import EnvironmentContext # create proxy functions for # each method on the EnvironmentContext class. EnvironmentContext.create_module_class_proxy(globals(), locals())	195 B	2026-01-20 12:58
📄 context.pyi # ### this file stubs are generated by tools/write_pyi.py - do not edit ### # ### imports are manually managed from __future__ import annotations from typing import Any from typing import Callable from typing import Collection from typing import ContextManager from typing import Dict from typing import List from typing import Literal from typing import Mapping from typing import MutableMapping from typing import Optional from typing import overload from typing import TextIO from typing import Tuple from typing import TYPE_CHECKING from typing import Union if TYPE_CHECKING: from sqlalchemy.engine.base import Connection from sqlalchemy.engine.url import URL from sqlalchemy.sql.elements import ClauseElement from sqlalchemy.sql.schema import Column from sqlalchemy.sql.schema import FetchedValue from sqlalchemy.sql.schema import MetaData from sqlalchemy.sql.schema import SchemaItem from .autogenerate.api import AutogenContext from .config import Config from .operations.ops import MigrateOperation from .runtime.migration import _ProxyTransaction from .runtime.migration import MigrationContext from .runtime.migration import MigrationInfo from .script import ScriptDirectory ### end imports ### def begin_transaction() -> Union[_ProxyTransaction, ContextManager[None]]: """Return a context manager that will enclose an operation within a "transaction", as defined by the environment's offline and transactional DDL settings. e.g.:: with context.begin_transaction(): context.run_migrations() :meth:`.begin_transaction` is intended to "do the right thing" regardless of calling context: * If :meth:`.is_transactional_ddl` is ``False``, returns a "do nothing" context manager which otherwise produces no transactional state or directives. * If :meth:`.is_offline_mode` is ``True``, returns a context manager that will invoke the :meth:`.DefaultImpl.emit_begin` and :meth:`.DefaultImpl.emit_commit` methods, which will produce the string directives ``BEGIN`` and ``COMMIT`` on the output stream, as rendered by the target backend (e.g. SQL Server would emit ``BEGIN TRANSACTION``). * Otherwise, calls :meth:`sqlalchemy.engine.Connection.begin` on the current online connection, which returns a :class:`sqlalchemy.engine.Transaction` object. This object demarcates a real transaction and is itself a context manager, which will roll back if an exception is raised. Note that a custom ``env.py`` script which has more specific transactional needs can of course manipulate the :class:`~sqlalchemy.engine.Connection` directly to produce transactional state in "online" mode. """ config: Config def configure( connection: Optional[Connection] = None, url: Union[str, URL, None] = None, dialect_name: Optional[str] = None, dialect_opts: Optional[Dict[str, Any]] = None, transactional_ddl: Optional[bool] = None, transaction_per_migration: bool = False, output_buffer: Optional[TextIO] = None, starting_rev: Optional[str] = None, tag: Optional[str] = None, template_args: Optional[Dict[str, Any]] = None, render_as_batch: bool = False, target_metadata: Optional[MetaData] = None, include_name: Optional[ Callable[ [ Optional[str], Literal[ "schema", "table", "column", "index", "unique_constraint", "foreign_key_constraint", ], MutableMapping[ Literal[ "schema_name", "table_name", "schema_qualified_table_name", ], Optional[str], ], ], bool, ] ] = None, include_object: Optional[ Callable[ [ SchemaItem, Optional[str], Literal[ "schema", "table", "column", "index", "unique_constraint", "foreign_key_constraint", ], bool, Optional[SchemaItem], ], bool, ] ] = None, include_schemas: bool = False, process_revision_directives: Optional[ Callable[ [MigrationContext, Tuple[str, str], List[MigrateOperation]], None ] ] = None, compare_type: bool = False, compare_server_default: Union[ bool, Callable[ [ MigrationContext, Column[Any], Column[Any], Optional[str], Optional[FetchedValue], Optional[str], ], Optional[bool], ], ] = False, render_item: Optional[ Callable[[str, Any, AutogenContext], Union[str, Literal[False]]] ] = None, literal_binds: bool = False, upgrade_token: str = "upgrades", downgrade_token: str = "downgrades", alembic_module_prefix: str = "op.", sqlalchemy_module_prefix: str = "sa.", user_module_prefix: Optional[str] = None, on_version_apply: Optional[ Callable[ [ MigrationContext, MigrationInfo, Collection[Any], Mapping[str, Any], ], None, ] ] = None, *kw: Any, ) -> None: """Configure a :class:`.MigrationContext` within this :class:`.EnvironmentContext` which will provide database connectivity and other configuration to a series of migration scripts. Many methods on :class:`.EnvironmentContext` require that this method has been called in order to function, as they ultimately need to have database access or at least access to the dialect in use. Those which do are documented as such. The important thing needed by :meth:`.configure` is a means to determine what kind of database dialect is in use. An actual connection to that database is needed only if the :class:`.MigrationContext` is to be used in "online" mode. If the :meth:`.is_offline_mode` function returns ``True``, then no connection is needed here. Otherwise, the ``connection`` parameter should be present as an instance of :class:`sqlalchemy.engine.Connection`. This function is typically called from the ``env.py`` script within a migration environment. It can be called multiple times for an invocation. The most recent :class:`~sqlalchemy.engine.Connection` for which it was called is the one that will be operated upon by the next call to :meth:`.run_migrations`. General parameters: :param connection: a :class:`~sqlalchemy.engine.Connection` to use for SQL execution in "online" mode. When present, is also used to determine the type of dialect in use. :param url: a string database url, or a :class:`sqlalchemy.engine.url.URL` object. The type of dialect to be used will be derived from this if ``connection`` is not passed. :param dialect_name: string name of a dialect, such as "postgresql", "mssql", etc. The type of dialect to be used will be derived from this if ``connection`` and ``url`` are not passed. :param dialect_opts: dictionary of options to be passed to dialect constructor. .. versionadded:: 1.0.12 :param transactional_ddl: Force the usage of "transactional" DDL on or off; this otherwise defaults to whether or not the dialect in use supports it. :param transaction_per_migration: if True, nest each migration script in a transaction rather than the full series of migrations to run. :param output_buffer: a file-like object that will be used for textual output when the ``--sql`` option is used to generate SQL scripts. Defaults to ``sys.stdout`` if not passed here and also not present on the :class:`.Config` object. The value here overrides that of the :class:`.Config` object. :param output_encoding: when using ``--sql`` to generate SQL scripts, apply this encoding to the string output. :param literal_binds: when using ``--sql`` to generate SQL scripts, pass through the ``literal_binds`` flag to the compiler so that any literal values that would ordinarily be bound parameters are converted to plain strings. .. warning:: Dialects can typically only handle simple datatypes like strings and numbers for auto-literal generation. Datatypes like dates, intervals, and others may still require manual formatting, typically using :meth:`.Operations.inline_literal`. .. note:: the ``literal_binds`` flag is ignored on SQLAlchemy versions prior to 0.8 where this feature is not supported. .. seealso:: :meth:`.Operations.inline_literal` :param starting_rev: Override the "starting revision" argument when using ``--sql`` mode. :param tag: a string tag for usage by custom ``env.py`` scripts. Set via the ``--tag`` option, can be overridden here. :param template_args: dictionary of template arguments which will be added to the template argument environment when running the "revision" command. Note that the script environment is only run within the "revision" command if the --autogenerate option is used, or if the option "revision_environment=true" is present in the alembic.ini file. :param version_table: The name of the Alembic version table. The default is ``'alembic_version'``. :param version_table_schema: Optional schema to place version table within. :param version_table_pk: boolean, whether the Alembic version table should use a primary key constraint for the "value" column; this only takes effect when the table is first created. Defaults to True; setting to False should not be necessary and is here for backwards compatibility reasons. :param on_version_apply: a callable or collection of callables to be run for each migration step. The callables will be run in the order they are given, once for each migration step, after the respective operation has been applied but before its transaction is finalized. Each callable accepts no positional arguments and the following keyword arguments: ``ctx``: the :class:`.MigrationContext` running the migration, * ``step``: a :class:`.MigrationInfo` representing the step currently being applied, * ``heads``: a collection of version strings representing the current heads, * ``run_args``: the ``*kwargs`` passed to :meth:`.run_migrations`. Parameters specific to the autogenerate feature, when ``alembic revision`` is run with the ``--autogenerate`` feature: :param target_metadata: a :class:`sqlalchemy.schema.MetaData` object, or a sequence of :class:`~sqlalchemy.schema.MetaData` objects, that will be consulted during autogeneration. The tables present in each :class:`~sqlalchemy.schema.MetaData` will be compared against what is locally available on the target :class:`~sqlalchemy.engine.Connection` to produce candidate upgrade/downgrade operations. :param compare_type: Indicates type comparison behavior during an autogenerate operation. Defaults to ``False`` which disables type comparison. Set to ``True`` to turn on default type comparison, which has varied accuracy depending on backend. See :ref:`compare_types` for an example as well as information on other type comparison options. .. seealso:: :ref:`compare_types` :paramref:`.EnvironmentContext.configure.compare_server_default` :param compare_server_default: Indicates server default comparison behavior during an autogenerate operation. Defaults to ``False`` which disables server default comparison. Set to ``True`` to turn on server default comparison, which has varied accuracy depending on backend. To customize server default comparison behavior, a callable may be specified which can filter server default comparisons during an autogenerate operation. defaults during an autogenerate operation. The format of this callable is:: def my_compare_server_default(context, inspected_column, metadata_column, inspected_default, metadata_default, rendered_metadata_default): # return True if the defaults are different, # False if not, or None to allow the default implementation # to compare these defaults return None context.configure( # ... compare_server_default = my_compare_server_default ) ``inspected_column`` is a dictionary structure as returned by :meth:`sqlalchemy.engine.reflection.Inspector.get_columns`, whereas ``metadata_column`` is a :class:`sqlalchemy.schema.Column` from the local model environment. A return value of ``None`` indicates to allow default server default comparison to proceed. Note that some backends such as Postgresql actually execute the two defaults on the database side to compare for equivalence. .. seealso:: :paramref:`.EnvironmentContext.configure.compare_type` :param include_name: A callable function which is given the chance to return ``True`` or ``False`` for any database reflected object based on its name, including database schema names when the :paramref:`.EnvironmentContext.configure.include_schemas` flag is set to ``True``. The function accepts the following positional arguments: ``name``: the name of the object, such as schema name or table name. Will be ``None`` when indicating the default schema name of the database connection. * ``type``: a string describing the type of object; currently ``"schema"``, ``"table"``, ``"column"``, ``"index"``, ``"unique_constraint"``, or ``"foreign_key_constraint"`` * ``parent_names``: a dictionary of "parent" object names, that are relative to the name being given. Keys in this dictionary may include: ``"schema_name"``, ``"table_name"`` or ``"schema_qualified_table_name"``. E.g.:: def include_name(name, type_, parent_names): if type_ == "schema": return name in ["schema_one", "schema_two"] else: return True context.configure( # ... include_schemas = True, include_name = include_name ) .. versionadded:: 1.5 .. seealso:: :ref:`autogenerate_include_hooks` :paramref:`.EnvironmentContext.configure.include_object` :paramref:`.EnvironmentContext.configure.include_schemas` :param include_object: A callable function which is given the chance to return ``True`` or ``False`` for any object, indicating if the given object should be considered in the autogenerate sweep. The function accepts the following positional arguments: * ``object``: a :class:`~sqlalchemy.schema.SchemaItem` object such as a :class:`~sqlalchemy.schema.Table`, :class:`~sqlalchemy.schema.Column`, :class:`~sqlalchemy.schema.Index` :class:`~sqlalchemy.schema.UniqueConstraint`, or :class:`~sqlalchemy.schema.ForeignKeyConstraint` object * ``name``: the name of the object. This is typically available via ``object.name``. * ``type``: a string describing the type of object; currently ``"table"``, ``"column"``, ``"index"``, ``"unique_constraint"``, or ``"foreign_key_constraint"`` * ``reflected``: ``True`` if the given object was produced based on table reflection, ``False`` if it's from a local :class:`.MetaData` object. * ``compare_to``: the object being compared against, if available, else ``None``. E.g.:: def include_object(object, name, type_, reflected, compare_to): if (type_ == "column" and not reflected and object.info.get("skip_autogenerate", False)): return False else: return True context.configure( # ... include_object = include_object ) For the use case of omitting specific schemas from a target database when :paramref:`.EnvironmentContext.configure.include_schemas` is set to ``True``, the :attr:`~sqlalchemy.schema.Table.schema` attribute can be checked for each :class:`~sqlalchemy.schema.Table` object passed to the hook, however it is much more efficient to filter on schemas before reflection of objects takes place using the :paramref:`.EnvironmentContext.configure.include_name` hook. .. seealso:: :ref:`autogenerate_include_hooks` :paramref:`.EnvironmentContext.configure.include_name` :paramref:`.EnvironmentContext.configure.include_schemas` :param render_as_batch: if True, commands which alter elements within a table will be placed under a ``with batch_alter_table():`` directive, so that batch migrations will take place. .. seealso:: :ref:`batch_migrations` :param include_schemas: If True, autogenerate will scan across all schemas located by the SQLAlchemy :meth:`~sqlalchemy.engine.reflection.Inspector.get_schema_names` method, and include all differences in tables found across all those schemas. When using this option, you may want to also use the :paramref:`.EnvironmentContext.configure.include_name` parameter to specify a callable which can filter the tables/schemas that get included. .. seealso:: :ref:`autogenerate_include_hooks` :paramref:`.EnvironmentContext.configure.include_name` :paramref:`.EnvironmentContext.configure.include_object` :param render_item: Callable that can be used to override how any schema item, i.e. column, constraint, type, etc., is rendered for autogenerate. The callable receives a string describing the type of object, the object, and the autogen context. If it returns False, the default rendering method will be used. If it returns None, the item will not be rendered in the context of a Table construct, that is, can be used to skip columns or constraints within op.create_table():: def my_render_column(type_, col, autogen_context): if type_ == "column" and isinstance(col, MySpecialCol): return repr(col) else: return False context.configure( # ... render_item = my_render_column ) Available values for the type string include: ``"column"``, ``"primary_key"``, ``"foreign_key"``, ``"unique"``, ``"check"``, ``"type"``, ``"server_default"``. .. seealso:: :ref:`autogen_render_types` :param upgrade_token: When autogenerate completes, the text of the candidate upgrade operations will be present in this template variable when ``script.py.mako`` is rendered. Defaults to ``upgrades``. :param downgrade_token: When autogenerate completes, the text of the candidate downgrade operations will be present in this template variable when ``script.py.mako`` is rendered. Defaults to ``downgrades``. :param alembic_module_prefix: When autogenerate refers to Alembic :mod:`alembic.operations` constructs, this prefix will be used (i.e. ``op.create_table``) Defaults to "``op.``". Can be ``None`` to indicate no prefix. :param sqlalchemy_module_prefix: When autogenerate refers to SQLAlchemy :class:`~sqlalchemy.schema.Column` or type classes, this prefix will be used (i.e. ``sa.Column("somename", sa.Integer)``) Defaults to "``sa.``". Can be ``None`` to indicate no prefix. Note that when dialect-specific types are rendered, autogenerate will render them using the dialect module name, i.e. ``mssql.BIT()``, ``postgresql.UUID()``. :param user_module_prefix: When autogenerate refers to a SQLAlchemy type (e.g. :class:`.TypeEngine`) where the module name is not under the ``sqlalchemy`` namespace, this prefix will be used within autogenerate. If left at its default of ``None``, the ``__module__`` attribute of the type is used to render the import module. It's a good practice to set this and to have all custom types be available from a fixed module space, in order to future-proof migration files against reorganizations in modules. .. seealso:: :ref:`autogen_module_prefix` :param process_revision_directives: a callable function that will be passed a structure representing the end result of an autogenerate or plain "revision" operation, which can be manipulated to affect how the ``alembic revision`` command ultimately outputs new revision scripts. The structure of the callable is:: def process_revision_directives(context, revision, directives): pass The ``directives`` parameter is a Python list containing a single :class:`.MigrationScript` directive, which represents the revision file to be generated. This list as well as its contents may be freely modified to produce any set of commands. The section :ref:`customizing_revision` shows an example of doing this. The ``context`` parameter is the :class:`.MigrationContext` in use, and ``revision`` is a tuple of revision identifiers representing the current revision of the database. The callable is invoked at all times when the ``--autogenerate`` option is passed to ``alembic revision``. If ``--autogenerate`` is not passed, the callable is invoked only if the ``revision_environment`` variable is set to True in the Alembic configuration, in which case the given ``directives`` collection will contain empty :class:`.UpgradeOps` and :class:`.DowngradeOps` collections for ``.upgrade_ops`` and ``.downgrade_ops``. The ``--autogenerate`` option itself can be inferred by inspecting ``context.config.cmd_opts.autogenerate``. The callable function may optionally be an instance of a :class:`.Rewriter` object. This is a helper object that assists in the production of autogenerate-stream rewriter functions. .. seealso:: :ref:`customizing_revision` :ref:`autogen_rewriter` :paramref:`.command.revision.process_revision_directives` Parameters specific to individual backends: :param mssql_batch_separator: The "batch separator" which will be placed between each statement when generating offline SQL Server migrations. Defaults to ``GO``. Note this is in addition to the customary semicolon ``;`` at the end of each statement; SQL Server considers the "batch separator" to denote the end of an individual statement execution, and cannot group certain dependent operations in one step. :param oracle_batch_separator: The "batch separator" which will be placed between each statement when generating offline Oracle migrations. Defaults to ``/``. Oracle doesn't add a semicolon between statements like most other backends. """ def execute( sql: Union[ClauseElement, str], execution_options: Optional[dict] = None ) -> None: """Execute the given SQL using the current change context. The behavior of :meth:`.execute` is the same as that of :meth:`.Operations.execute`. Please see that function's documentation for full detail including caveats and limitations. This function requires that a :class:`.MigrationContext` has first been made available via :meth:`.configure`. """ def get_bind() -> Connection: """Return the current 'bind'. In "online" mode, this is the :class:`sqlalchemy.engine.Connection` currently being used to emit SQL to the database. This function requires that a :class:`.MigrationContext` has first been made available via :meth:`.configure`. """ def get_context() -> MigrationContext: """Return the current :class:`.MigrationContext` object. If :meth:`.EnvironmentContext.configure` has not been called yet, raises an exception. """ def get_head_revision() -> Union[str, Tuple[str, ...], None]: """Return the hex identifier of the 'head' script revision. If the script directory has multiple heads, this method raises a :class:`.CommandError`; :meth:`.EnvironmentContext.get_head_revisions` should be preferred. This function does not require that the :class:`.MigrationContext` has been configured. .. seealso:: :meth:`.EnvironmentContext.get_head_revisions` """ def get_head_revisions() -> Union[str, Tuple[str, ...], None]: """Return the hex identifier of the 'heads' script revision(s). This returns a tuple containing the version number of all heads in the script directory. This function does not require that the :class:`.MigrationContext` has been configured. """ def get_revision_argument() -> Union[str, Tuple[str, ...], None]: """Get the 'destination' revision argument. This is typically the argument passed to the ``upgrade`` or ``downgrade`` command. If it was specified as ``head``, the actual version number is returned; if specified as ``base``, ``None`` is returned. This function does not require that the :class:`.MigrationContext` has been configured. """ def get_starting_revision_argument() -> Union[str, Tuple[str, ...], None]: """Return the 'starting revision' argument, if the revision was passed using ``start:end``. This is only meaningful in "offline" mode. Returns ``None`` if no value is available or was configured. This function does not require that the :class:`.MigrationContext` has been configured. """ def get_tag_argument() -> Optional[str]: """Return the value passed for the ``--tag`` argument, if any. The ``--tag`` argument is not used directly by Alembic, but is available for custom ``env.py`` configurations that wish to use it; particularly for offline generation scripts that wish to generate tagged filenames. This function does not require that the :class:`.MigrationContext` has been configured. .. seealso:: :meth:`.EnvironmentContext.get_x_argument` - a newer and more open ended system of extending ``env.py`` scripts via the command line. """ @overload def get_x_argument(as_dictionary: Literal[False]) -> List[str]: ... @overload def get_x_argument(as_dictionary: Literal[True]) -> Dict[str, str]: ... @overload def get_x_argument( as_dictionary: bool = ..., ) -> Union[List[str], Dict[str, str]]: """Return the value(s) passed for the ``-x`` argument, if any. The ``-x`` argument is an open ended flag that allows any user-defined value or values to be passed on the command line, then available here for consumption by a custom ``env.py`` script. The return value is a list, returned directly from the ``argparse`` structure. If ``as_dictionary=True`` is passed, the ``x`` arguments are parsed using ``key=value`` format into a dictionary that is then returned. For example, to support passing a database URL on the command line, the standard ``env.py`` script can be modified like this:: cmd_line_url = context.get_x_argument( as_dictionary=True).get('dbname') if cmd_line_url: engine = create_engine(cmd_line_url) else: engine = engine_from_config( config.get_section(config.config_ini_section), prefix='sqlalchemy.', poolclass=pool.NullPool) This then takes effect by running the ``alembic`` script as:: alembic -x dbname=postgresql://user:pass@host/dbname upgrade head This function does not require that the :class:`.MigrationContext` has been configured. .. seealso:: :meth:`.EnvironmentContext.get_tag_argument` :attr:`.Config.cmd_opts` """ def is_offline_mode() -> bool: """Return True if the current migrations environment is running in "offline mode". This is ``True`` or ``False`` depending on the ``--sql`` flag passed. This function does not require that the :class:`.MigrationContext` has been configured. """ def is_transactional_ddl(): """Return True if the context is configured to expect a transactional DDL capable backend. This defaults to the type of database in use, and can be overridden by the ``transactional_ddl`` argument to :meth:`.configure` This function requires that a :class:`.MigrationContext` has first been made available via :meth:`.configure`. """ def run_migrations(kw: Any) -> None: """Run migrations as determined by the current command line configuration as well as versioning information present (or not) in the current database connection (if one is present). The function accepts optional ``kw`` arguments. If these are passed, they are sent directly to the ``upgrade()`` and ``downgrade()`` functions within each target revision file. By modifying the ``script.py.mako`` file so that the ``upgrade()`` and ``downgrade()`` functions accept arguments, parameters can be passed here so that contextual information, usually information to identify a particular database in use, can be passed from a custom ``env.py`` script to the migration functions. This function requires that a :class:`.MigrationContext` has first been made available via :meth:`.configure`. """ script: ScriptDirectory def static_output(text: str) -> None: """Emit text directly to the "offline" SQL stream. Typically this is for emitting comments that start with --. The statement is not treated as a SQL execution, no ; or batch separator is added, etc. """	30.12 KB	2026-01-20 12:58
📁 ddl	-	2026-03-21 15:45
📄 environment.py from .runtime.environment import * # noqa	43 B	2026-01-20 12:58
📄 migration.py from .runtime.migration import * # noqa	41 B	2026-01-20 12:58
📄 op.py from .operations.base import Operations # create proxy functions for # each method on the Operations class. Operations.create_module_class_proxy(globals(), locals())	167 B	2026-01-20 12:58
📄 op.pyi # ### this file stubs are generated by tools/write_pyi.py - do not edit ### # ### imports are manually managed from __future__ import annotations from contextlib import contextmanager from typing import Any from typing import Awaitable from typing import Callable from typing import Dict from typing import Iterator from typing import List from typing import Literal from typing import Mapping from typing import Optional from typing import Sequence from typing import Tuple from typing import Type from typing import TYPE_CHECKING from typing import TypeVar from typing import Union from sqlalchemy.sql.expression import TableClause from sqlalchemy.sql.expression import Update if TYPE_CHECKING: from sqlalchemy.engine import Connection from sqlalchemy.sql.elements import BinaryExpression from sqlalchemy.sql.elements import conv from sqlalchemy.sql.elements import TextClause from sqlalchemy.sql.functions import Function from sqlalchemy.sql.schema import Column from sqlalchemy.sql.schema import Computed from sqlalchemy.sql.schema import Identity from sqlalchemy.sql.schema import SchemaItem from sqlalchemy.sql.schema import Table from sqlalchemy.sql.type_api import TypeEngine from sqlalchemy.util import immutabledict from .operations.ops import BatchOperations from .operations.ops import MigrateOperation from .runtime.migration import MigrationContext from .util.sqla_compat import _literal_bindparam _T = TypeVar("_T") ### end imports ### def add_column( table_name: str, column: Column[Any], , schema: Optional[str] = None ) -> None: """Issue an "add column" instruction using the current migration context. e.g.:: from alembic import op from sqlalchemy import Column, String op.add_column("organization", Column("name", String())) The :meth:`.Operations.add_column` method typically corresponds to the SQL command "ALTER TABLE... ADD COLUMN". Within the scope of this command, the column's name, datatype, nullability, and optional server-generated defaults may be indicated. .. note:: With the exception of NOT NULL constraints or single-column FOREIGN KEY constraints, other kinds of constraints such as PRIMARY KEY, UNIQUE or CHECK constraints cannot* be generated using this method; for these constraints, refer to operations such as :meth:`.Operations.create_primary_key` and :meth:`.Operations.create_check_constraint`. In particular, the following :class:`~sqlalchemy.schema.Column` parameters are ignored: * :paramref:`~sqlalchemy.schema.Column.primary_key` - SQL databases typically do not support an ALTER operation that can add individual columns one at a time to an existing primary key constraint, therefore it's less ambiguous to use the :meth:`.Operations.create_primary_key` method, which assumes no existing primary key constraint is present. * :paramref:`~sqlalchemy.schema.Column.unique` - use the :meth:`.Operations.create_unique_constraint` method * :paramref:`~sqlalchemy.schema.Column.index` - use the :meth:`.Operations.create_index` method The provided :class:`~sqlalchemy.schema.Column` object may include a :class:`~sqlalchemy.schema.ForeignKey` constraint directive, referencing a remote table name. For this specific type of constraint, Alembic will automatically emit a second ALTER statement in order to add the single-column FOREIGN KEY constraint separately:: from alembic import op from sqlalchemy import Column, INTEGER, ForeignKey op.add_column( "organization", Column("account_id", INTEGER, ForeignKey("accounts.id")), ) The column argument passed to :meth:`.Operations.add_column` is a :class:`~sqlalchemy.schema.Column` construct, used in the same way it's used in SQLAlchemy. In particular, values or functions to be indicated as producing the column's default value on the database side are specified using the ``server_default`` parameter, and not ``default`` which only specifies Python-side defaults:: from alembic import op from sqlalchemy import Column, TIMESTAMP, func # specify "DEFAULT NOW" along with the column add op.add_column( "account", Column("timestamp", TIMESTAMP, server_default=func.now()), ) :param table_name: String name of the parent table. :param column: a :class:`sqlalchemy.schema.Column` object representing the new column. :param schema: Optional schema name to operate within. To control quoting of the schema outside of the default behavior, use the SQLAlchemy construct :class:`~sqlalchemy.sql.elements.quoted_name`. """ def alter_column( table_name: str, column_name: str, , nullable: Optional[bool] = None, comment: Union[str, Literal[False], None] = False, server_default: Any = False, new_column_name: Optional[str] = None, type_: Union[TypeEngine, Type[TypeEngine], None] = None, existing_type: Union[TypeEngine, Type[TypeEngine], None] = None, existing_server_default: Union[ str, bool, Identity, Computed, None ] = False, existing_nullable: Optional[bool] = None, existing_comment: Optional[str] = None, schema: Optional[str] = None, kw: Any, ) -> None: r"""Issue an "alter column" instruction using the current migration context. Generally, only that aspect of the column which is being changed, i.e. name, type, nullability, default, needs to be specified. Multiple changes can also be specified at once and the backend should "do the right thing", emitting each change either separately or together as the backend allows. MySQL has special requirements here, since MySQL cannot ALTER a column without a full specification. When producing MySQL-compatible migration files, it is recommended that the ``existing_type``, ``existing_server_default``, and ``existing_nullable`` parameters be present, if not being altered. Type changes which are against the SQLAlchemy "schema" types :class:`~sqlalchemy.types.Boolean` and :class:`~sqlalchemy.types.Enum` may also add or drop constraints which accompany those types on backends that don't support them natively. The ``existing_type`` argument is used in this case to identify and remove a previous constraint that was bound to the type object. :param table_name: string name of the target table. :param column_name: string name of the target column, as it exists before the operation begins. :param nullable: Optional; specify ``True`` or ``False`` to alter the column's nullability. :param server_default: Optional; specify a string SQL expression, :func:`~sqlalchemy.sql.expression.text`, or :class:`~sqlalchemy.schema.DefaultClause` to indicate an alteration to the column's default value. Set to ``None`` to have the default removed. :param comment: optional string text of a new comment to add to the column. .. versionadded:: 1.0.6 :param new_column_name: Optional; specify a string name here to indicate the new name within a column rename operation. :param type\_: Optional; a :class:`~sqlalchemy.types.TypeEngine` type object to specify a change to the column's type. For SQLAlchemy types that also indicate a constraint (i.e. :class:`~sqlalchemy.types.Boolean`, :class:`~sqlalchemy.types.Enum`), the constraint is also generated. :param autoincrement: set the ``AUTO_INCREMENT`` flag of the column; currently understood by the MySQL dialect. :param existing_type: Optional; a :class:`~sqlalchemy.types.TypeEngine` type object to specify the previous type. This is required for all MySQL column alter operations that don't otherwise specify a new type, as well as for when nullability is being changed on a SQL Server column. It is also used if the type is a so-called SQLlchemy "schema" type which may define a constraint (i.e. :class:`~sqlalchemy.types.Boolean`, :class:`~sqlalchemy.types.Enum`), so that the constraint can be dropped. :param existing_server_default: Optional; The existing default value of the column. Required on MySQL if an existing default is not being changed; else MySQL removes the default. :param existing_nullable: Optional; the existing nullability of the column. Required on MySQL if the existing nullability is not being changed; else MySQL sets this to NULL. :param existing_autoincrement: Optional; the existing autoincrement of the column. Used for MySQL's system of altering a column that specifies ``AUTO_INCREMENT``. :param existing_comment: string text of the existing comment on the column to be maintained. Required on MySQL if the existing comment on the column is not being changed. .. versionadded:: 1.0.6 :param schema: Optional schema name to operate within. To control quoting of the schema outside of the default behavior, use the SQLAlchemy construct :class:`~sqlalchemy.sql.elements.quoted_name`. :param postgresql_using: String argument which will indicate a SQL expression to render within the Postgresql-specific USING clause within ALTER COLUMN. This string is taken directly as raw SQL which must explicitly include any necessary quoting or escaping of tokens within the expression. """ @contextmanager def batch_alter_table( table_name: str, schema: Optional[str] = None, recreate: Literal["auto", "always", "never"] = "auto", partial_reordering: Optional[tuple] = None, copy_from: Optional[Table] = None, table_args: Tuple[Any, ...] = (), table_kwargs: Mapping[str, Any] = immutabledict({}), reflect_args: Tuple[Any, ...] = (), reflect_kwargs: Mapping[str, Any] = immutabledict({}), naming_convention: Optional[Dict[str, str]] = None, ) -> Iterator[BatchOperations]: """Invoke a series of per-table migrations in batch. Batch mode allows a series of operations specific to a table to be syntactically grouped together, and allows for alternate modes of table migration, in particular the "recreate" style of migration required by SQLite. "recreate" style is as follows: 1. A new table is created with the new specification, based on the migration directives within the batch, using a temporary name. 2. the data copied from the existing table to the new table. 3. the existing table is dropped. 4. the new table is renamed to the existing table name. The directive by default will only use "recreate" style on the SQLite backend, and only if directives are present which require this form, e.g. anything other than ``add_column()``. The batch operation on other backends will proceed using standard ALTER TABLE operations. The method is used as a context manager, which returns an instance of :class:`.BatchOperations`; this object is the same as :class:`.Operations` except that table names and schema names are omitted. E.g.:: with op.batch_alter_table("some_table") as batch_op: batch_op.add_column(Column("foo", Integer)) batch_op.drop_column("bar") The operations within the context manager are invoked at once when the context is ended. When run against SQLite, if the migrations include operations not supported by SQLite's ALTER TABLE, the entire table will be copied to a new one with the new specification, moving all data across as well. The copy operation by default uses reflection to retrieve the current structure of the table, and therefore :meth:`.batch_alter_table` in this mode requires that the migration is run in "online" mode. The ``copy_from`` parameter may be passed which refers to an existing :class:`.Table` object, which will bypass this reflection step. .. note:: The table copy operation will currently not copy CHECK constraints, and may not copy UNIQUE constraints that are unnamed, as is possible on SQLite. See the section :ref:`sqlite_batch_constraints` for workarounds. :param table_name: name of table :param schema: optional schema name. :param recreate: under what circumstances the table should be recreated. At its default of ``"auto"``, the SQLite dialect will recreate the table if any operations other than ``add_column()``, ``create_index()``, or ``drop_index()`` are present. Other options include ``"always"`` and ``"never"``. :param copy_from: optional :class:`~sqlalchemy.schema.Table` object that will act as the structure of the table being copied. If omitted, table reflection is used to retrieve the structure of the table. .. seealso:: :ref:`batch_offline_mode` :paramref:`~.Operations.batch_alter_table.reflect_args` :paramref:`~.Operations.batch_alter_table.reflect_kwargs` :param reflect_args: a sequence of additional positional arguments that will be applied to the table structure being reflected / copied; this may be used to pass column and constraint overrides to the table that will be reflected, in lieu of passing the whole :class:`~sqlalchemy.schema.Table` using :paramref:`~.Operations.batch_alter_table.copy_from`. :param reflect_kwargs: a dictionary of additional keyword arguments that will be applied to the table structure being copied; this may be used to pass additional table and reflection options to the table that will be reflected, in lieu of passing the whole :class:`~sqlalchemy.schema.Table` using :paramref:`~.Operations.batch_alter_table.copy_from`. :param table_args: a sequence of additional positional arguments that will be applied to the new :class:`~sqlalchemy.schema.Table` when created, in addition to those copied from the source table. This may be used to provide additional constraints such as CHECK constraints that may not be reflected. :param table_kwargs: a dictionary of additional keyword arguments that will be applied to the new :class:`~sqlalchemy.schema.Table` when created, in addition to those copied from the source table. This may be used to provide for additional table options that may not be reflected. :param naming_convention: a naming convention dictionary of the form described at :ref:`autogen_naming_conventions` which will be applied to the :class:`~sqlalchemy.schema.MetaData` during the reflection process. This is typically required if one wants to drop SQLite constraints, as these constraints will not have names when reflected on this backend. Requires SQLAlchemy 0.9.4* or greater. .. seealso:: :ref:`dropping_sqlite_foreign_keys` :param partial_reordering: a list of tuples, each suggesting a desired ordering of two or more columns in the newly created table. Requires that :paramref:`.batch_alter_table.recreate` is set to ``"always"``. Examples, given a table with columns "a", "b", "c", and "d": Specify the order of all columns:: with op.batch_alter_table( "some_table", recreate="always", partial_reordering=[("c", "d", "a", "b")], ) as batch_op: pass Ensure "d" appears before "c", and "b", appears before "a":: with op.batch_alter_table( "some_table", recreate="always", partial_reordering=[("d", "c"), ("b", "a")], ) as batch_op: pass The ordering of columns not included in the partial_reordering set is undefined. Therefore it is best to specify the complete ordering of all columns for best results. .. versionadded:: 1.4.0 .. note:: batch mode requires SQLAlchemy 0.8 or above. .. seealso:: :ref:`batch_migrations` """ def bulk_insert( table: Union[Table, TableClause], rows: List[dict], , multiinsert: bool = True, ) -> None: """Issue a "bulk insert" operation using the current migration context. This provides a means of representing an INSERT of multiple rows which works equally well in the context of executing on a live connection as well as that of generating a SQL script. In the case of a SQL script, the values are rendered inline into the statement. e.g.:: from alembic import op from datetime import date from sqlalchemy.sql import table, column from sqlalchemy import String, Integer, Date # Create an ad-hoc table to use for the insert statement. accounts_table = table( "account", column("id", Integer), column("name", String), column("create_date", Date), ) op.bulk_insert( accounts_table, [ { "id": 1, "name": "John Smith", "create_date": date(2010, 10, 5), }, { "id": 2, "name": "Ed Williams", "create_date": date(2007, 5, 27), }, { "id": 3, "name": "Wendy Jones", "create_date": date(2008, 8, 15), }, ], ) When using --sql mode, some datatypes may not render inline automatically, such as dates and other special types. When this issue is present, :meth:`.Operations.inline_literal` may be used:: op.bulk_insert( accounts_table, [ { "id": 1, "name": "John Smith", "create_date": op.inline_literal("2010-10-05"), }, { "id": 2, "name": "Ed Williams", "create_date": op.inline_literal("2007-05-27"), }, { "id": 3, "name": "Wendy Jones", "create_date": op.inline_literal("2008-08-15"), }, ], multiinsert=False, ) When using :meth:`.Operations.inline_literal` in conjunction with :meth:`.Operations.bulk_insert`, in order for the statement to work in "online" (e.g. non --sql) mode, the :paramref:`~.Operations.bulk_insert.multiinsert` flag should be set to ``False``, which will have the effect of individual INSERT statements being emitted to the database, each with a distinct VALUES clause, so that the "inline" values can still be rendered, rather than attempting to pass the values as bound parameters. :param table: a table object which represents the target of the INSERT. :param rows: a list of dictionaries indicating rows. :param multiinsert: when at its default of True and --sql mode is not enabled, the INSERT statement will be executed using "executemany()" style, where all elements in the list of dictionaries are passed as bound parameters in a single list. Setting this to False results in individual INSERT statements being emitted per parameter set, and is needed in those cases where non-literal values are present in the parameter sets. """ def create_check_constraint( constraint_name: Optional[str], table_name: str, condition: Union[str, BinaryExpression, TextClause], , schema: Optional[str] = None, *kw: Any, ) -> None: """Issue a "create check constraint" instruction using the current migration context. e.g.:: from alembic import op from sqlalchemy.sql import column, func op.create_check_constraint( "ck_user_name_len", "user", func.len(column("name")) > 5, ) CHECK constraints are usually against a SQL expression, so ad-hoc table metadata is usually needed. The function will convert the given arguments into a :class:`sqlalchemy.schema.CheckConstraint` bound to an anonymous table in order to emit the CREATE statement. :param name: Name of the check constraint. The name is necessary so that an ALTER statement can be emitted. For setups that use an automated naming scheme such as that described at :ref:`sqla:constraint_naming_conventions`, ``name`` here can be ``None``, as the event listener will apply the name to the constraint object when it is associated with the table. :param table_name: String name of the source table. :param condition: SQL expression that's the condition of the constraint. Can be a string or SQLAlchemy expression language structure. :param deferrable: optional bool. If set, emit DEFERRABLE or NOT DEFERRABLE when issuing DDL for this constraint. :param initially: optional string. If set, emit INITIALLY <value> when issuing DDL for this constraint. :param schema: Optional schema name to operate within. To control quoting of the schema outside of the default behavior, use the SQLAlchemy construct :class:`~sqlalchemy.sql.elements.quoted_name`. """ def create_exclude_constraint( constraint_name: str, table_name: str, elements: Any, *kw: Any ) -> Optional[Table]: """Issue an alter to create an EXCLUDE constraint using the current migration context. .. note:: This method is Postgresql specific, and additionally requires at least SQLAlchemy 1.0. e.g.:: from alembic import op op.create_exclude_constraint( "user_excl", "user", ("period", "&&"), ("group", "="), where=("group != 'some group'"), ) Note that the expressions work the same way as that of the ``ExcludeConstraint`` object itself; if plain strings are passed, quoting rules must be applied manually. :param name: Name of the constraint. :param table_name: String name of the source table. :param elements: exclude conditions. :param where: SQL expression or SQL string with optional WHERE clause. :param deferrable: optional bool. If set, emit DEFERRABLE or NOT DEFERRABLE when issuing DDL for this constraint. :param initially: optional string. If set, emit INITIALLY <value> when issuing DDL for this constraint. :param schema: Optional schema name to operate within. """ def create_foreign_key( constraint_name: Optional[str], source_table: str, referent_table: str, local_cols: List[str], remote_cols: List[str], , onupdate: Optional[str] = None, ondelete: Optional[str] = None, deferrable: Optional[bool] = None, initially: Optional[str] = None, match: Optional[str] = None, source_schema: Optional[str] = None, referent_schema: Optional[str] = None, *dialect_kw: Any, ) -> None: """Issue a "create foreign key" instruction using the current migration context. e.g.:: from alembic import op op.create_foreign_key( "fk_user_address", "address", "user", ["user_id"], ["id"], ) This internally generates a :class:`~sqlalchemy.schema.Table` object containing the necessary columns, then generates a new :class:`~sqlalchemy.schema.ForeignKeyConstraint` object which it then associates with the :class:`~sqlalchemy.schema.Table`. Any event listeners associated with this action will be fired off normally. The :class:`~sqlalchemy.schema.AddConstraint` construct is ultimately used to generate the ALTER statement. :param constraint_name: Name of the foreign key constraint. The name is necessary so that an ALTER statement can be emitted. For setups that use an automated naming scheme such as that described at :ref:`sqla:constraint_naming_conventions`, ``name`` here can be ``None``, as the event listener will apply the name to the constraint object when it is associated with the table. :param source_table: String name of the source table. :param referent_table: String name of the destination table. :param local_cols: a list of string column names in the source table. :param remote_cols: a list of string column names in the remote table. :param onupdate: Optional string. If set, emit ON UPDATE <value> when issuing DDL for this constraint. Typical values include CASCADE, DELETE and RESTRICT. :param ondelete: Optional string. If set, emit ON DELETE <value> when issuing DDL for this constraint. Typical values include CASCADE, DELETE and RESTRICT. :param deferrable: optional bool. If set, emit DEFERRABLE or NOT DEFERRABLE when issuing DDL for this constraint. :param source_schema: Optional schema name of the source table. :param referent_schema: Optional schema name of the destination table. """ def create_index( index_name: Optional[str], table_name: str, columns: Sequence[Union[str, TextClause, Function[Any]]], , schema: Optional[str] = None, unique: bool = False, kw: Any, ) -> None: r"""Issue a "create index" instruction using the current migration context. e.g.:: from alembic import op op.create_index("ik_test", "t1", ["foo", "bar"]) Functional indexes can be produced by using the :func:`sqlalchemy.sql.expression.text` construct:: from alembic import op from sqlalchemy import text op.create_index("ik_test", "t1", [text("lower(foo)")]) :param index_name: name of the index. :param table_name: name of the owning table. :param columns: a list consisting of string column names and/or :func:`~sqlalchemy.sql.expression.text` constructs. :param schema: Optional schema name to operate within. To control quoting of the schema outside of the default behavior, use the SQLAlchemy construct :class:`~sqlalchemy.sql.elements.quoted_name`. :param unique: If True, create a unique index. :param quote: Force quoting of this column's name on or off, corresponding to ``True`` or ``False``. When left at its default of ``None``, the column identifier will be quoted according to whether the name is case sensitive (identifiers with at least one upper case character are treated as case sensitive), or if it's a reserved word. This flag is only needed to force quoting of a reserved word which is not known by the SQLAlchemy dialect. :param \kw: Additional keyword arguments not mentioned above are dialect specific, and passed in the form ``<dialectname>_<argname>``. See the documentation regarding an individual dialect at :ref:`dialect_toplevel` for detail on documented arguments. """ def create_primary_key( constraint_name: Optional[str], table_name: str, columns: List[str], , schema: Optional[str] = None, ) -> None: """Issue a "create primary key" instruction using the current migration context. e.g.:: from alembic import op op.create_primary_key("pk_my_table", "my_table", ["id", "version"]) This internally generates a :class:`~sqlalchemy.schema.Table` object containing the necessary columns, then generates a new :class:`~sqlalchemy.schema.PrimaryKeyConstraint` object which it then associates with the :class:`~sqlalchemy.schema.Table`. Any event listeners associated with this action will be fired off normally. The :class:`~sqlalchemy.schema.AddConstraint` construct is ultimately used to generate the ALTER statement. :param constraint_name: Name of the primary key constraint. The name is necessary so that an ALTER statement can be emitted. For setups that use an automated naming scheme such as that described at :ref:`sqla:constraint_naming_conventions` ``name`` here can be ``None``, as the event listener will apply the name to the constraint object when it is associated with the table. :param table_name: String name of the target table. :param columns: a list of string column names to be applied to the primary key constraint. :param schema: Optional schema name to operate within. To control quoting of the schema outside of the default behavior, use the SQLAlchemy construct :class:`~sqlalchemy.sql.elements.quoted_name`. """ def create_table(table_name: str, columns: SchemaItem, *kw: Any) -> Table: r"""Issue a "create table" instruction using the current migration context. This directive receives an argument list similar to that of the traditional :class:`sqlalchemy.schema.Table` construct, but without the metadata:: from sqlalchemy import INTEGER, VARCHAR, NVARCHAR, Column from alembic import op op.create_table( "account", Column("id", INTEGER, primary_key=True), Column("name", VARCHAR(50), nullable=False), Column("description", NVARCHAR(200)), Column("timestamp", TIMESTAMP, server_default=func.now()), ) Note that :meth:`.create_table` accepts :class:`~sqlalchemy.schema.Column` constructs directly from the SQLAlchemy library. In particular, default values to be created on the database side are specified using the ``server_default`` parameter, and not ``default`` which only specifies Python-side defaults:: from alembic import op from sqlalchemy import Column, TIMESTAMP, func # specify "DEFAULT NOW" along with the "timestamp" column op.create_table( "account", Column("id", INTEGER, primary_key=True), Column("timestamp", TIMESTAMP, server_default=func.now()), ) The function also returns a newly created :class:`~sqlalchemy.schema.Table` object, corresponding to the table specification given, which is suitable for immediate SQL operations, in particular :meth:`.Operations.bulk_insert`:: from sqlalchemy import INTEGER, VARCHAR, NVARCHAR, Column from alembic import op account_table = op.create_table( "account", Column("id", INTEGER, primary_key=True), Column("name", VARCHAR(50), nullable=False), Column("description", NVARCHAR(200)), Column("timestamp", TIMESTAMP, server_default=func.now()), ) op.bulk_insert( account_table, [ {"name": "A1", "description": "account 1"}, {"name": "A2", "description": "account 2"}, ], ) :param table_name: Name of the table :param \columns: collection of :class:`~sqlalchemy.schema.Column` objects within the table, as well as optional :class:`~sqlalchemy.schema.Constraint` objects and :class:`~.sqlalchemy.schema.Index` objects. :param schema: Optional schema name to operate within. To control quoting of the schema outside of the default behavior, use the SQLAlchemy construct :class:`~sqlalchemy.sql.elements.quoted_name`. :param \*kw: Other keyword arguments are passed to the underlying :class:`sqlalchemy.schema.Table` object created for the command. :return: the :class:`~sqlalchemy.schema.Table` object corresponding to the parameters given. """ def create_table_comment( table_name: str, comment: Optional[str], , existing_comment: Optional[str] = None, schema: Optional[str] = None, ) -> None: """Emit a COMMENT ON operation to set the comment for a table. .. versionadded:: 1.0.6 :param table_name: string name of the target table. :param comment: string value of the comment being registered against the specified table. :param existing_comment: String value of a comment already registered on the specified table, used within autogenerate so that the operation is reversible, but not required for direct use. .. seealso:: :meth:`.Operations.drop_table_comment` :paramref:`.Operations.alter_column.comment` """ def create_unique_constraint( constraint_name: Optional[str], table_name: str, columns: Sequence[str], , schema: Optional[str] = None, kw: Any, ) -> Any: """Issue a "create unique constraint" instruction using the current migration context. e.g.:: from alembic import op op.create_unique_constraint("uq_user_name", "user", ["name"]) This internally generates a :class:`~sqlalchemy.schema.Table` object containing the necessary columns, then generates a new :class:`~sqlalchemy.schema.UniqueConstraint` object which it then associates with the :class:`~sqlalchemy.schema.Table`. Any event listeners associated with this action will be fired off normally. The :class:`~sqlalchemy.schema.AddConstraint` construct is ultimately used to generate the ALTER statement. :param name: Name of the unique constraint. The name is necessary so that an ALTER statement can be emitted. For setups that use an automated naming scheme such as that described at :ref:`sqla:constraint_naming_conventions`, ``name`` here can be ``None``, as the event listener will apply the name to the constraint object when it is associated with the table. :param table_name: String name of the source table. :param columns: a list of string column names in the source table. :param deferrable: optional bool. If set, emit DEFERRABLE or NOT DEFERRABLE when issuing DDL for this constraint. :param initially: optional string. If set, emit INITIALLY <value> when issuing DDL for this constraint. :param schema: Optional schema name to operate within. To control quoting of the schema outside of the default behavior, use the SQLAlchemy construct :class:`~sqlalchemy.sql.elements.quoted_name`. """ def drop_column( table_name: str, column_name: str, , schema: Optional[str] = None, *kw: Any, ) -> None: """Issue a "drop column" instruction using the current migration context. e.g.:: drop_column("organization", "account_id") :param table_name: name of table :param column_name: name of column :param schema: Optional schema name to operate within. To control quoting of the schema outside of the default behavior, use the SQLAlchemy construct :class:`~sqlalchemy.sql.elements.quoted_name`. :param mssql_drop_check: Optional boolean. When ``True``, on Microsoft SQL Server only, first drop the CHECK constraint on the column using a SQL-script-compatible block that selects into a @variable from sys.check_constraints, then exec's a separate DROP CONSTRAINT for that constraint. :param mssql_drop_default: Optional boolean. When ``True``, on Microsoft SQL Server only, first drop the DEFAULT constraint on the column using a SQL-script-compatible block that selects into a @variable from sys.default_constraints, then exec's a separate DROP CONSTRAINT for that default. :param mssql_drop_foreign_key: Optional boolean. When ``True``, on Microsoft SQL Server only, first drop a single FOREIGN KEY constraint on the column using a SQL-script-compatible block that selects into a @variable from sys.foreign_keys/sys.foreign_key_columns, then exec's a separate DROP CONSTRAINT for that default. Only works if the column has exactly one FK constraint which refers to it, at the moment. """ def drop_constraint( constraint_name: str, table_name: str, type_: Optional[str] = None, , schema: Optional[str] = None, ) -> None: r"""Drop a constraint of the given name, typically via DROP CONSTRAINT. :param constraint_name: name of the constraint. :param table_name: table name. :param type\_: optional, required on MySQL. can be 'foreignkey', 'primary', 'unique', or 'check'. :param schema: Optional schema name to operate within. To control quoting of the schema outside of the default behavior, use the SQLAlchemy construct :class:`~sqlalchemy.sql.elements.quoted_name`. """ def drop_index( index_name: str, table_name: Optional[str] = None, , schema: Optional[str] = None, kw: Any, ) -> None: r"""Issue a "drop index" instruction using the current migration context. e.g.:: drop_index("accounts") :param index_name: name of the index. :param table_name: name of the owning table. Some backends such as Microsoft SQL Server require this. :param schema: Optional schema name to operate within. To control quoting of the schema outside of the default behavior, use the SQLAlchemy construct :class:`~sqlalchemy.sql.elements.quoted_name`. :param \kw: Additional keyword arguments not mentioned above are dialect specific, and passed in the form ``<dialectname>_<argname>``. See the documentation regarding an individual dialect at :ref:`dialect_toplevel` for detail on documented arguments. """ def drop_table( table_name: str, , schema: Optional[str] = None, kw: Any ) -> None: r"""Issue a "drop table" instruction using the current migration context. e.g.:: drop_table("accounts") :param table_name: Name of the table :param schema: Optional schema name to operate within. To control quoting of the schema outside of the default behavior, use the SQLAlchemy construct :class:`~sqlalchemy.sql.elements.quoted_name`. :param \kw: Other keyword arguments are passed to the underlying :class:`sqlalchemy.schema.Table` object created for the command. """ def drop_table_comment( table_name: str, , existing_comment: Optional[str] = None, schema: Optional[str] = None, ) -> None: """Issue a "drop table comment" operation to remove an existing comment set on a table. .. versionadded:: 1.0.6 :param table_name: string name of the target table. :param existing_comment: An optional string value of a comment already registered on the specified table. .. seealso:: :meth:`.Operations.create_table_comment` :paramref:`.Operations.alter_column.comment` """ def execute( sqltext: Union[str, TextClause, Update], , execution_options: Optional[dict[str, Any]] = None, ) -> None: r"""Execute the given SQL using the current migration context. The given SQL can be a plain string, e.g.:: op.execute("INSERT INTO table (foo) VALUES ('some value')") Or it can be any kind of Core SQL Expression construct, such as below where we use an update construct:: from sqlalchemy.sql import table, column from sqlalchemy import String from alembic import op account = table("account", column("name", String)) op.execute( account.update() .where(account.c.name == op.inline_literal("account 1")) .values({"name": op.inline_literal("account 2")}) ) Above, we made use of the SQLAlchemy :func:`sqlalchemy.sql.expression.table` and :func:`sqlalchemy.sql.expression.column` constructs to make a brief, ad-hoc table construct just for our UPDATE statement. A full :class:`~sqlalchemy.schema.Table` construct of course works perfectly fine as well, though note it's a recommended practice to at least ensure the definition of a table is self-contained within the migration script, rather than imported from a module that may break compatibility with older migrations. In a SQL script context, the statement is emitted directly to the output stream. There is no return result, however, as this function is oriented towards generating a change script that can run in "offline" mode. Additionally, parameterized statements are discouraged here, as they will not work in offline mode. Above, we use :meth:`.inline_literal` where parameters are to be used. For full interaction with a connected database where parameters can also be used normally, use the "bind" available from the context:: from alembic import op connection = op.get_bind() connection.execute( account.update() .where(account.c.name == "account 1") .values({"name": "account 2"}) ) Additionally, when passing the statement as a plain string, it is first coerceed into a :func:`sqlalchemy.sql.expression.text` construct before being passed along. In the less likely case that the literal SQL string contains a colon, it must be escaped with a backslash, as:: op.execute(r"INSERT INTO table (foo) VALUES ('\:colon_value')") :param sqltext: Any legal SQLAlchemy expression, including: * a string * a :func:`sqlalchemy.sql.expression.text` construct. * a :func:`sqlalchemy.sql.expression.insert` construct. * a :func:`sqlalchemy.sql.expression.update`, :func:`sqlalchemy.sql.expression.insert`, or :func:`sqlalchemy.sql.expression.delete` construct. * Any "executable" described in SQLAlchemy Core documentation, noting that no result set is returned. .. note:: when passing a plain string, the statement is coerced into a :func:`sqlalchemy.sql.expression.text` construct. This construct considers symbols with colons, e.g. ``:foo`` to be bound parameters. To avoid this, ensure that colon symbols are escaped, e.g. ``\:foo``. :param execution_options: Optional dictionary of execution options, will be passed to :meth:`sqlalchemy.engine.Connection.execution_options`. """ def f(name: str) -> conv: """Indicate a string name that has already had a naming convention applied to it. This feature combines with the SQLAlchemy ``naming_convention`` feature to disambiguate constraint names that have already had naming conventions applied to them, versus those that have not. This is necessary in the case that the ``"%(constraint_name)s"`` token is used within a naming convention, so that it can be identified that this particular name should remain fixed. If the :meth:`.Operations.f` is used on a constraint, the naming convention will not take effect:: op.add_column("t", "x", Boolean(name=op.f("ck_bool_t_x"))) Above, the CHECK constraint generated will have the name ``ck_bool_t_x`` regardless of whether or not a naming convention is in use. Alternatively, if a naming convention is in use, and 'f' is not used, names will be converted along conventions. If the ``target_metadata`` contains the naming convention ``{"ck": "ck_bool_%(table_name)s_%(constraint_name)s"}``, then the output of the following: op.add_column("t", "x", Boolean(name="x")) will be:: CONSTRAINT ck_bool_t_x CHECK (x in (1, 0))) The function is rendered in the output of autogenerate when a particular constraint name is already converted. """ def get_bind() -> Connection: """Return the current 'bind'. Under normal circumstances, this is the :class:`~sqlalchemy.engine.Connection` currently being used to emit SQL to the database. In a SQL script context, this value is ``None``. [TODO: verify this] """ def get_context() -> MigrationContext: """Return the :class:`.MigrationContext` object that's currently in use. """ def implementation_for(op_cls: Any) -> Callable[..., Any]: """Register an implementation for a given :class:`.MigrateOperation`. This is part of the operation extensibility API. .. seealso:: :ref:`operation_plugins` - example of use """ def inline_literal( value: Union[str, int], type_: Optional[TypeEngine] = None ) -> _literal_bindparam: r"""Produce an 'inline literal' expression, suitable for using in an INSERT, UPDATE, or DELETE statement. When using Alembic in "offline" mode, CRUD operations aren't compatible with SQLAlchemy's default behavior surrounding literal values, which is that they are converted into bound values and passed separately into the ``execute()`` method of the DBAPI cursor. An offline SQL script needs to have these rendered inline. While it should always be noted that inline literal values are an enormous security hole in an application that handles untrusted input, a schema migration is not run in this context, so literals are safe to render inline, with the caveat that advanced types like dates may not be supported directly by SQLAlchemy. See :meth:`.Operations.execute` for an example usage of :meth:`.Operations.inline_literal`. The environment can also be configured to attempt to render "literal" values inline automatically, for those simple types that are supported by the dialect; see :paramref:`.EnvironmentContext.configure.literal_binds` for this more recently added feature. :param value: The value to render. Strings, integers, and simple numerics should be supported. Other types like boolean, dates, etc. may or may not be supported yet by various backends. :param type\_: optional - a :class:`sqlalchemy.types.TypeEngine` subclass stating the type of this value. In SQLAlchemy expressions, this is usually derived automatically from the Python type of the value itself, as well as based on the context in which the value is used. .. seealso:: :paramref:`.EnvironmentContext.configure.literal_binds` """ def invoke(operation: MigrateOperation) -> Any: """Given a :class:`.MigrateOperation`, invoke it in terms of this :class:`.Operations` instance. """ def register_operation( name: str, sourcename: Optional[str] = None ) -> Callable[..., Any]: """Register a new operation for this class. This method is normally used to add new operations to the :class:`.Operations` class, and possibly the :class:`.BatchOperations` class as well. All Alembic migration operations are implemented via this system, however the system is also available as a public API to facilitate adding custom operations. .. seealso:: :ref:`operation_plugins` """ def rename_table( old_table_name: str, new_table_name: str, , schema: Optional[str] = None ) -> None: """Emit an ALTER TABLE to rename a table. :param old_table_name: old name. :param new_table_name: new name. :param schema: Optional schema name to operate within. To control quoting of the schema outside of the default behavior, use the SQLAlchemy construct :class:`~sqlalchemy.sql.elements.quoted_name`. """ def run_async( async_function: Callable[..., Awaitable[_T]], args: Any, **kw_args: Any ) -> _T: """Invoke the given asynchronous callable, passing an asynchronous :class:`~sqlalchemy.ext.asyncio.AsyncConnection` as the first argument. This method allows calling async functions from within the synchronous ``upgrade()`` or ``downgrade()`` alembic migration method. The async connection passed to the callable shares the same transaction as the connection running in the migration context. Any additional arg or kw_arg passed to this function are passed to the provided async function. .. versionadded: 1.11 .. note:: This method can be called only when alembic is called using an async dialect. """	47.38 KB	2026-01-20 12:58
📁 operations	-	2026-03-21 15:45
📄 py.typed	0 B	2026-01-20 12:58
📁 runtime	-	2026-03-21 15:45
📁 script	-	2026-03-21 15:45
📁 templates	-	2026-01-20 12:58
📁 testing	-	2026-03-21 15:45
📁 util	-	2026-03-21 15:45

Edit File:

Terminal