from typing import Set
from typing import Tuple
from typing import TYPE_CHECKING
-from typing import Union
from sqlalchemy import inspect
from sqlalchemy.engine import Connection
from sqlalchemy.engine import Dialect
from sqlalchemy.engine import Inspector
- from sqlalchemy.sql.schema import Column
- from sqlalchemy.sql.schema import ForeignKeyConstraint
- from sqlalchemy.sql.schema import Index
from sqlalchemy.sql.schema import MetaData
- from sqlalchemy.sql.schema import Table
- from sqlalchemy.sql.schema import UniqueConstraint
+ from sqlalchemy.sql.schema import SchemaItem
- from alembic.config import Config
- from alembic.operations.ops import MigrationScript
- from alembic.operations.ops import UpgradeOps
- from alembic.runtime.migration import MigrationContext
- from alembic.script.base import Script
- from alembic.script.base import ScriptDirectory
+ from ..config import Config
+ from ..operations.ops import MigrationScript
+ from ..operations.ops import UpgradeOps
+ from ..runtime.environment import NameFilterParentNames
+ from ..runtime.environment import NameFilterType
+ from ..runtime.environment import RenderItemFn
+ from ..runtime.migration import MigrationContext
+ from ..script.base import Script
+ from ..script.base import ScriptDirectory
def compare_metadata(context: MigrationContext, metadata: MetaData) -> Any:
alembic_module_prefix: str = "op.",
render_as_batch: bool = False,
imports: Tuple[str, ...] = (),
- render_item: None = None,
+ render_item: Optional[RenderItemFn] = None,
migration_context: Optional[MigrationContext] = None,
) -> str:
"""Render Python code given an :class:`.UpgradeOps` or
def run_name_filters(
self,
name: Optional[str],
- type_: str,
- parent_names: Dict[str, Optional[str]],
+ type_: NameFilterType,
+ parent_names: NameFilterParentNames,
) -> bool:
"""Run the context's name filters and return True if the targets
should be part of the autogenerate operation.
def run_object_filters(
self,
- object_: Union[
- Table,
- Index,
- Column,
- UniqueConstraint,
- ForeignKeyConstraint,
- ],
+ object_: SchemaItem,
name: Optional[str],
- type_: str,
+ type_: NameFilterType,
reflected: bool,
- compare_to: Optional[Union[Table, Index, Column, UniqueConstraint]],
+ compare_to: Optional[SchemaItem],
) -> bool:
"""Run the context's object filters and return True if the targets
should be part of the autogenerate operation.
(inspector),
# fmt: on
)
- sqla_compat._reflect_table(inspector, t, None)
+ sqla_compat._reflect_table(inspector, t)
if autogen_context.run_object_filters(t, tname, "table", True, None):
modify_table_ops = ops.ModifyTableOps(tname, [], schema=s)
_compat_autogen_column_reflect(inspector),
# fmt: on
)
- sqla_compat._reflect_table(inspector, t, None)
+ sqla_compat._reflect_table(inspector, t)
conn_column_info[(s, tname)] = t
for s, tname in sorted(existing_tables, key=lambda x: (x[0] or "", x[1])):
from __future__ import annotations
from typing import Any
-from typing import Callable
from typing import ContextManager
from typing import Dict
from typing import List
from sqlalchemy.sql.schema import MetaData
from .config import Config
- from .operations import MigrateOperation
+ from .runtime.environment import IncludeNameFn
+ from .runtime.environment import IncludeObjectFn
+ from .runtime.environment import OnVersionApplyFn
+ from .runtime.environment import ProcessRevisionDirectiveFn
+ from .runtime.environment import RenderItemFn
from .runtime.migration import _ProxyTransaction
from .runtime.migration import MigrationContext
from .script import ScriptDirectory
def configure(
connection: Optional[Connection] = None,
- url: Union[str, URL, None] = None,
+ url: Optional[Union[str, URL]] = None,
dialect_name: Optional[str] = None,
dialect_opts: Optional[Dict[str, Any]] = None,
transactional_ddl: Optional[bool] = None,
template_args: Optional[Dict[str, Any]] = None,
render_as_batch: bool = False,
target_metadata: Optional[MetaData] = None,
- include_name: Optional[Callable[..., bool]] = None,
- include_object: Optional[Callable[..., bool]] = None,
+ include_name: Optional[IncludeNameFn] = None,
+ include_object: Optional[IncludeObjectFn] = None,
include_schemas: bool = False,
- process_revision_directives: Optional[
- Callable[
- [MigrationContext, Tuple[str, str], List[MigrateOperation]], None
- ]
- ] = None,
+ process_revision_directives: Optional[ProcessRevisionDirectiveFn] = None,
compare_type: bool = False,
compare_server_default: bool = False,
- render_item: Optional[Callable[..., bool]] = None,
+ render_item: Optional[RenderItemFn] = 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[..., None]] = None,
+ on_version_apply: Optional[OnVersionApplyFn] = None,
**kw: Any,
) -> None:
"""Configure a :class:`.MigrationContext` within this
``"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"``.
+ include: ``"schema_name"``, ``"table_name"`` or
+ ``"schema_qualified_table_name"``.
E.g.::
def _exec(
self,
construct: Union[ClauseElement, str],
- execution_options: Optional[dict] = None,
+ execution_options: Optional[dict[str, Any]] = None,
multiparams: Sequence[dict] = (),
- params: Dict[str, int] = util.immutabledict(),
+ params: Dict[str, Any] = util.immutabledict(),
) -> Optional[CursorResult]:
if isinstance(construct, str):
construct = text(construct)
def execute(
self,
sql: Union[ClauseElement, str],
- execution_options: None = None,
+ execution_options: Optional[dict[str, Any]] = None,
) -> None:
self._exec(sql, execution_options)
"""
def execute(
- sqltext: Union[str, TextClause, Update], execution_options: None = None
+ sqltext: Union[str, TextClause, Update],
+ execution_options: Optional[dict[str, Any]] = None,
) -> Optional[Table]:
r"""Execute the given SQL using the current migration context.
"""
def inline_literal(
- value: Union[str, int], type_: None = None
+ 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.
if TYPE_CHECKING:
from typing import Literal
- from sqlalchemy import Table # noqa
+ from sqlalchemy import Table
from sqlalchemy.engine import Connection
+ from sqlalchemy.types import TypeEngine
from .batch import BatchOperationsImpl
from .ops import MigrateOperation
return conv(name)
def inline_literal(
- self, value: Union[str, int], type_: None = None
+ self, 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.
server_default: Optional[Union[Function[Any], str, bool]] = False,
name: Optional[str] = None,
type_: Optional[TypeEngine] = None,
- autoincrement: None = None,
+ autoincrement: Optional[Union[bool, Literal["auto"]]] = None,
comment: Union[str, Literal[False]] = False,
**kw,
) -> None:
local_cols: List[str],
remote_cols: List[str],
referent_schema: Optional[str] = None,
- onupdate: None = None,
- ondelete: None = None,
- deferrable: None = None,
- initially: None = None,
- match: None = None,
+ onupdate: Optional[str] = None,
+ ondelete: Optional[str] = None,
+ deferrable: Optional[bool] = None,
+ initially: Optional[str] = None,
+ match: Optional[str] = None,
**dialect_kw: Any,
) -> None:
"""Issue a "create foreign key" instruction using the
type_: Optional[Union[TypeEngine, Type[TypeEngine]]] = None,
existing_type: Optional[Union[TypeEngine, Type[TypeEngine]]] = None,
existing_server_default: bool = False,
- existing_nullable: None = None,
- existing_comment: None = None,
- insert_before: None = None,
- insert_after: None = None,
+ existing_nullable: Optional[bool] = None,
+ existing_comment: Optional[str] = None,
+ insert_before: Optional[str] = None,
+ insert_after: Optional[str] = None,
**kw: Any,
) -> Optional[Table]:
"""Issue an "alter column" instruction using the current
modify_server_default=server_default,
modify_nullable=nullable,
modify_comment=comment,
+ insert_before=insert_before,
+ insert_after=insert_after,
**kw,
)
def __init__(
self,
sqltext: Union[Update, str, Insert, TextClause],
- execution_options: None = None,
+ execution_options: Optional[dict[str, Any]] = None,
) -> None:
self.sqltext = sqltext
self.execution_options = execution_options
cls,
operations: Operations,
sqltext: Union[str, TextClause, Update],
- execution_options: None = None,
+ execution_options: Optional[dict[str, Any]] = None,
) -> Optional[Table]:
r"""Execute the given SQL using the current migration context.
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 Mapping
+from typing import MutableMapping
from typing import Optional
from typing import overload
from typing import TextIO
from typing import TYPE_CHECKING
from typing import Union
+from typing_extensions import Literal
+
from .migration import _ProxyTransaction
from .migration import MigrationContext
from .. import util
from ..operations import Operations
if TYPE_CHECKING:
- from typing import Literal
from sqlalchemy.engine import URL
from sqlalchemy.engine.base import Connection
from sqlalchemy.sql.elements import ClauseElement
from sqlalchemy.sql.schema import MetaData
+ from sqlalchemy.sql.schema import SchemaItem
+ from .migration import MigrationInfo
+ from ..autogenerate.api import AutogenContext
from ..config import Config
from ..ddl import DefaultImpl
from ..operations.ops import MigrateOperation
[MigrationContext, Tuple[str, str], List["MigrateOperation"]], None
]
+RenderItemFn = Callable[
+ [str, Any, "AutogenContext"], Union[str, Literal[False]]
+]
+
+NameFilterType = Literal[
+ "schema",
+ "table",
+ "column",
+ "index",
+ "unique_constraint",
+ "foreign_key_constraint",
+]
+NameFilterParentNames = MutableMapping[
+ Literal["schema_name", "table_name", "schema_qualified_table_name"],
+ Optional[str],
+]
+IncludeNameFn = Callable[
+ [Optional[str], NameFilterType, NameFilterParentNames], bool
+]
+
+IncludeObjectFn = Callable[
+ [
+ "SchemaItem",
+ Optional[str],
+ NameFilterType,
+ bool,
+ Optional["SchemaItem"],
+ ],
+ bool,
+]
+
+OnVersionApplyFn = Callable[
+ [MigrationContext, "MigrationInfo", Collection[Any], Mapping[str, Any]],
+ None,
+]
+
class EnvironmentContext(util.ModuleClsProxy):
template_args: Optional[Dict[str, Any]] = None,
render_as_batch: bool = False,
target_metadata: Optional[MetaData] = None,
- include_name: Optional[Callable[..., bool]] = None,
- include_object: Optional[Callable[..., bool]] = None,
+ include_name: Optional[IncludeNameFn] = None,
+ include_object: Optional[IncludeObjectFn] = None,
include_schemas: bool = False,
process_revision_directives: Optional[
ProcessRevisionDirectiveFn
] = None,
compare_type: bool = False,
compare_server_default: bool = False,
- render_item: Optional[Callable[..., bool]] = None,
+ render_item: Optional[RenderItemFn] = 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[..., None]] = None,
+ on_version_apply: Optional[OnVersionApplyFn] = None,
**kw: Any,
) -> None:
"""Configure a :class:`.MigrationContext` within this
``"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"``.
+ include: ``"schema_name"``, ``"table_name"`` or
+ ``"schema_qualified_table_name"``.
E.g.::
import logging
import sys
from typing import Any
+from typing import Callable
from typing import cast
from typing import Collection
from typing import ContextManager
from typing import Dict
+from typing import Iterable
from typing import Iterator
from typing import List
from typing import Optional
def __enter__(self) -> _ProxyTransaction:
return self
- def __exit__(self, type_: None, value: None, traceback: None) -> None:
+ def __exit__(self, type_: Any, value: Any, traceback: Any) -> None:
if self._proxied_transaction is not None:
self._proxied_transaction.__exit__(type_, value, traceback)
self.migration_context._transaction = None
sqla_compat._get_connection_in_transaction(connection)
)
- self._migrations_fn = opts.get("fn")
+ self._migrations_fn: Optional[
+ Callable[..., Iterable[RevisionStep]]
+ ] = opts.get("fn")
self.as_sql = as_sql
self.purge = opts.get("purge", False)
self.migration_fn = self.stamp_revision
self.revision_map = revision_map
- doc: None = None
+ doc: Optional[str] = None
def stamp_revision(self, **kw: Any) -> None:
return None
return list(constraint.columns)
-def _reflect_table(
- inspector: Inspector, table: Table, include_cols: None
-) -> None:
+def _reflect_table(inspector: Inspector, table: Table) -> None:
if sqla_14:
return inspector.reflect_table(table, None)
else:
--- /dev/null
+.. change::
+ :tags: bug, batch
+ :tickets: 1221
+
+ Correctly pass previously ignored arguments ``insert_before`` and
+ ``insert_after`` in ``batch_alter_column``
insp = inspect(self.bind)
cols = insp.get_columns(t1.name)
refl = Table(t1.name, MetaData())
- sqla_compat._reflect_table(insp, refl, None)
+ sqla_compat._reflect_table(insp, refl)
ctx = self.autogen_context["context"]
return ctx.impl.compare_server_default(
refl.c[cols[0]["name"]], col, rendered, cols[0]["default"]
# Do not generate the base implementation to avoid mypy errors
overloads = typing.get_overloads(meth)
if overloads:
- # use enumerate so we can generate docs on the last overload
+ # use enumerate so we can generate docs on the
+ # last overload
for i, ovl in enumerate(overloads, 1):
_generate_stub_for_meth(
ovl,
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
