### end imports ###
def begin_transaction() -> Union[_ProxyTransaction, ContextManager[None]]:
- r"""Return a context manager that will
+ """Return a context manager that will
enclose an operation within a "transaction",
as defined by the environment's offline
and transactional DDL settings.
on_version_apply: Optional[Callable[..., None]] = None,
**kw: Any,
) -> None:
- r"""Configure a :class:`.MigrationContext` within this
+ """Configure a :class:`.MigrationContext` within this
:class:`.EnvironmentContext` which will provide database
connectivity and other configuration to a series of
migration scripts.
def execute(
sql: Union[ClauseElement, str], execution_options: Optional[dict] = None
) -> None:
- r"""Execute the given SQL using the current change context.
+ """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
"""
def get_bind() -> Connection:
- r"""Return the current 'bind'.
+ """Return the current 'bind'.
In "online" mode, this is the
:class:`sqlalchemy.engine.Connection` currently being used
"""
def get_context() -> MigrationContext:
- r"""Return the current :class:`.MigrationContext` object.
+ """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]:
- r"""Return the hex identifier of the 'head' script revision.
+ """Return the hex identifier of the 'head' script revision.
If the script directory has multiple heads, this
method raises a :class:`.CommandError`;
"""
def get_head_revisions() -> Union[str, Tuple[str, ...], None]:
- r"""Return the hex identifier of the 'heads' script revision(s).
+ """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.
"""
def get_revision_argument() -> Union[str, Tuple[str, ...], None]:
- r"""Get the 'destination' revision argument.
+ """Get the 'destination' revision argument.
This is typically the argument passed to the
``upgrade`` or ``downgrade`` command.
"""
def get_starting_revision_argument() -> Union[str, Tuple[str, ...], None]:
- r"""Return the 'starting revision' argument,
+ """Return the 'starting revision' argument,
if the revision was passed using ``start:end``.
This is only meaningful in "offline" mode.
"""
def get_tag_argument() -> Optional[str]:
- r"""Return the value passed for the ``--tag`` argument, if any.
+ """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
def get_x_argument(
as_dictionary: bool = ...,
) -> Union[List[str], Dict[str, str]]:
- r"""Return the value(s) passed for the ``-x`` argument, if any.
+ """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
"""
def is_offline_mode() -> bool:
- r"""Return True if the current migrations environment
+ """Return True if the current migrations environment
is running in "offline mode".
This is ``True`` or ``False`` depending
"""
def is_transactional_ddl():
- r"""Return True if the context is configured to expect a
+ """Return True if the context is configured to expect a
transactional DDL capable backend.
This defaults to the type of database in use, and
"""
def run_migrations(**kw: Any) -> None:
- r"""Run migrations as determined by the current command line
+ """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).
script: ScriptDirectory
def static_output(text: str) -> None:
- r"""Emit text directly to the "offline" SQL stream.
+ """Emit text directly to the "offline" SQL stream.
Typically this is for emitting comments that
start with --. The statement is not treated
# ### 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 Callable
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
def add_column(
table_name: str, column: Column, schema: Optional[str] = None
) -> Optional[Table]:
- r"""Issue an "add column" instruction using the current
+ """Issue an "add column" instruction using the current
migration context.
e.g.::
existing_nullable: Optional[bool] = None,
existing_comment: Optional[str] = None,
schema: Optional[str] = None,
- **kw: Any
+ **kw: Any,
) -> Optional[Table]:
r"""Issue an "alter column" instruction using the
current migration context.
reflect_kwargs: Mapping[str, Any] = immutabledict({}),
naming_convention: Optional[Dict[str, str]] = None,
) -> Iterator[BatchOperations]:
- r"""Invoke a series of per-table migrations in batch.
+ """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
rows: List[dict],
multiinsert: bool = True,
) -> None:
- r"""Issue a "bulk insert" operation using the current
+ """Issue a "bulk insert" operation using the current
migration context.
This provides a means of representing an INSERT of multiple rows
table_name: str,
condition: Union[str, BinaryExpression],
schema: Optional[str] = None,
- **kw: Any
+ **kw: Any,
) -> Optional[Table]:
- r"""Issue a "create check constraint" instruction using the
+ """Issue a "create check constraint" instruction using the
current migration context.
e.g.::
def create_exclude_constraint(
constraint_name: str, table_name: str, *elements: Any, **kw: Any
) -> Optional[Table]:
- r"""Issue an alter to create an EXCLUDE constraint using the
+ """Issue an alter to create an EXCLUDE constraint using the
current migration context.
.. note:: This method is Postgresql specific, and additionally
match: Optional[str] = None,
source_schema: Optional[str] = None,
referent_schema: Optional[str] = None,
- **dialect_kw: Any
+ **dialect_kw: Any,
) -> Optional[Table]:
- r"""Issue a "create foreign key" instruction using the
+ """Issue a "create foreign key" instruction using the
current migration context.
e.g.::
columns: Sequence[Union[str, TextClause, Function[Any]]],
schema: Optional[str] = None,
unique: bool = False,
- **kw: Any
+ **kw: Any,
) -> Optional[Table]:
r"""Issue a "create index" instruction using the current
migration context.
columns: List[str],
schema: Optional[str] = None,
) -> Optional[Table]:
- r"""Issue a "create primary key" instruction using the current
+ """Issue a "create primary key" instruction using the current
migration context.
e.g.::
existing_comment: None = None,
schema: Optional[str] = None,
) -> Optional[Table]:
- r"""Emit a COMMENT ON operation to set the comment for a table.
+ """Emit a COMMENT ON operation to set the comment for a table.
.. versionadded:: 1.0.6
table_name: str,
columns: Sequence[str],
schema: Optional[str] = None,
- **kw: Any
+ **kw: Any,
) -> Any:
- r"""Issue a "create unique constraint" instruction using the
+ """Issue a "create unique constraint" instruction using the
current migration context.
e.g.::
def drop_column(
table_name: str, column_name: str, schema: Optional[str] = None, **kw: Any
) -> Optional[Table]:
- r"""Issue a "drop column" instruction using the current
+ """Issue a "drop column" instruction using the current
migration context.
e.g.::
index_name: str,
table_name: Optional[str] = None,
schema: Optional[str] = None,
- **kw: Any
+ **kw: Any,
) -> Optional[Table]:
r"""Issue a "drop index" instruction using the current
migration context.
existing_comment: Optional[str] = None,
schema: Optional[str] = None,
) -> Optional[Table]:
- r"""Issue a "drop table comment" operation to
+ """Issue a "drop table comment" operation to
remove an existing comment set on a table.
.. versionadded:: 1.0.6
from sqlalchemy import String
from alembic import op
- account = table('account',
- column('name', String)
- )
+ 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')})
- )
+ 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
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"})
+ account.update()
+ .where(account.c.name == "account 1")
+ .values({"name": "account 2"})
)
Additionally, when passing the statement as a plain string, it is first
"""
def f(name: str) -> conv:
- r"""Indicate a string name that has already had a naming convention
+ """Indicate a string name that has already had a naming convention
applied to it.
This feature combines with the SQLAlchemy ``naming_convention`` feature
"""
def get_bind() -> Connection:
- r"""Return the current 'bind'.
+ """Return the current 'bind'.
Under normal circumstances, this is the
:class:`~sqlalchemy.engine.Connection` currently being used
"""
def get_context() -> MigrationContext:
- r"""Return the :class:`.MigrationContext` object that's
+ """Return the :class:`.MigrationContext` object that's
currently in use.
"""
def implementation_for(op_cls: Any) -> Callable[..., Any]:
- r"""Register an implementation for a given :class:`.MigrateOperation`.
+ """Register an implementation for a given :class:`.MigrateOperation`.
This is part of the operation extensibility API.
"""
def invoke(operation: MigrateOperation) -> Any:
- r"""Given a :class:`.MigrateOperation`, invoke it in terms of
+ """Given a :class:`.MigrateOperation`, invoke it in terms of
this :class:`.Operations` instance.
"""
def register_operation(
name: str, sourcename: Optional[str] = None
) -> Callable[..., Any]:
- r"""Register a new operation for this class.
+ """Register a new operation for this class.
This method is normally used to add new operations
to the :class:`.Operations` class, and possibly the
def rename_table(
old_table_name: str, new_table_name: str, schema: Optional[str] = None
) -> Optional[Table]:
- r"""Emit an ALTER TABLE to rename a table.
+ """Emit an ALTER TABLE to rename a table.
:param old_table_name: old name.
:param new_table_name: new name.
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
