Da

✏️ Editing: 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 overload
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

if TYPE_CHECKING:
    from sqlalchemy.engine import Connection
    from sqlalchemy.sql import Executable
    from sqlalchemy.sql.elements import ColumnElement
    from sqlalchemy.sql.elements import conv
    from sqlalchemy.sql.elements import TextClause
    from sqlalchemy.sql.expression import TableClause
    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.base import BatchOperations
    from .operations.ops import AddColumnOp
    from .operations.ops import AddConstraintOp
    from .operations.ops import AlterColumnOp
    from .operations.ops import AlterTableOp
    from .operations.ops import BulkInsertOp
    from .operations.ops import CreateIndexOp
    from .operations.ops import CreateTableCommentOp
    from .operations.ops import CreateTableOp
    from .operations.ops import DropColumnOp
    from .operations.ops import DropConstraintOp
    from .operations.ops import DropIndexOp
    from .operations.ops import DropTableCommentOp
    from .operations.ops import DropTableOp
    from .operations.ops import ExecuteSQLOp
    from .operations.ops import MigrateOperation
    from .runtime.migration import MigrationContext
    from .util.sqla_compat import _literal_bindparam

_T = TypeVar("_T")
_C = TypeVar("_C", bound=Callable[..., Any])

### 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[Any], Type[TypeEngine[Any]], None] = None,
    existing_type: Union[TypeEngine[Any], Type[TypeEngine[Any]], 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.
    :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
     SQLAlchemy "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.
    :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[Any, ...]] = 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.

.. note:: batch mode requires SQLAlchemy 0.8 or above.

.. seealso::

:ref:`batch_migrations`

"""

def bulk_insert(
    table: Union[Table, TableClause],
    rows: List[Dict[str, Any]],
    *,
    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, ColumnElement[bool], 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, ColumnElement[Any]]],
    *,
    schema: Optional[str] = None,
    unique: bool = False,
    if_not_exists: Optional[bool] = None,
    **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 if_not_exists: If True, adds IF NOT EXISTS operator when
     creating the new index.

.. versionadded:: 1.12.0

: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,
    if_not_exists: Optional[bool] = None,
    **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 if_not_exists: If True, adds IF NOT EXISTS operator when
     creating the new table.

.. versionadded:: 1.13.3
    :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.

: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,
    if_exists: Optional[bool] = 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 if_exists: If True, adds IF EXISTS operator when
     dropping the index.

.. versionadded:: 1.12.0

"""

def drop_table(
    table_name: str,
    *,
    schema: Optional[str] = None,
    if_exists: Optional[bool] = 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 if_exists: If True, adds IF EXISTS operator when
     dropping the table.

.. versionadded:: 1.13.3
    :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.

: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[Executable, str],
    *,
    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
    coerced 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` construct.
    * a :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[[_C], _C]:
    """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[Any]] = 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`

"""

@overload
def invoke(operation: CreateTableOp) -> Table: ...
@overload
def invoke(
    operation: Union[
        AddConstraintOp,
        DropConstraintOp,
        CreateIndexOp,
        DropIndexOp,
        AddColumnOp,
        AlterColumnOp,
        AlterTableOp,
        CreateTableCommentOp,
        DropTableCommentOp,
        DropColumnOp,
        BulkInsertOp,
        DropTableOp,
        ExecuteSQLOp,
    ]
) -> None: ...
@overload
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[[Type[_T]], Type[_T]]:
    """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.
    """

📁 Sid Gifari File Manager