### end imports ###
def add_column(
- table_name: str, column: Column, schema: Optional[str] = None
+ table_name: str, column: Column, *, schema: Optional[str] = None
) -> None:
"""Issue an "add column" instruction using the current
migration context.
def alter_column(
table_name: str,
column_name: str,
+ *,
nullable: Optional[bool] = None,
comment: Union[str, Literal[False], None] = False,
server_default: Any = False,
constraint_name: Optional[str],
table_name: str,
condition: Union[str, BinaryExpression, TextClause],
+ *,
schema: Optional[str] = None,
**kw: Any,
) -> None:
referent_table: str,
local_cols: List[str],
remote_cols: List[str],
+ *,
onupdate: Optional[str] = None,
ondelete: Optional[str] = None,
deferrable: Optional[bool] = None,
index_name: Optional[str],
table_name: str,
columns: Sequence[Union[str, TextClause, Function[Any]]],
+ *,
schema: Optional[str] = None,
unique: bool = False,
**kw: Any,
constraint_name: Optional[str],
table_name: str,
columns: List[str],
+ *,
schema: Optional[str] = None,
) -> None:
"""Issue a "create primary key" instruction using the current
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.
constraint_name: Optional[str],
table_name: str,
columns: Sequence[str],
+ *,
schema: Optional[str] = None,
**kw: Any,
) -> Any:
"""
def drop_column(
- table_name: str, column_name: str, schema: Optional[str] = None, **kw: Any
+ table_name: str,
+ column_name: str,
+ *,
+ schema: Optional[str] = None,
+ **kw: Any,
) -> None:
"""Issue a "drop column" instruction using the current
migration context.
def drop_constraint(
constraint_name: str,
table_name: str,
+ *,
type_: Optional[str] = None,
schema: Optional[str] = None,
) -> None:
def drop_index(
index_name: str,
+ *,
table_name: Optional[str] = None,
schema: Optional[str] = None,
**kw: Any,
"""
def drop_table(
- table_name: str, schema: Optional[str] = None, **kw: Any
+ table_name: str, *, schema: Optional[str] = None, **kw: Any
) -> None:
r"""Issue a "drop table" instruction using the current
migration context.
def drop_table_comment(
table_name: str,
+ *,
existing_comment: Optional[str] = None,
schema: Optional[str] = None,
) -> None:
"""
def rename_table(
- old_table_name: str, new_table_name: str, schema: Optional[str] = None
+ old_table_name: str, new_table_name: str, *, schema: Optional[str] = None
) -> None:
"""Emit an ALTER TABLE to rename a table.
else:
defaulted_vals = ()
+ defaulted_vals += tuple(spec[4])
+ # here, we are using formatargspec in a different way in order
+ # to get a string that will re-apply incoming arguments to a new
+ # function call
+
apply_kw = inspect_formatargspec(
- name_args,
+ name_args + spec[4],
spec[1],
spec[2],
defaulted_vals,
"doc": fn.__doc__,
}
)
+
globals_ = dict(globals())
globals_.update({"op_cls": op_cls})
lcl = {}
# ### do not edit ###
def add_column(
- self, table_name: str, column: Column, schema: Optional[str] = None
+ self,
+ table_name: str,
+ column: Column,
+ *,
+ schema: Optional[str] = None,
) -> None:
"""Issue an "add column" instruction using the current
migration context.
self,
table_name: str,
column_name: str,
+ *,
nullable: Optional[bool] = None,
comment: Union[str, Literal[False], None] = False,
server_default: Any = False,
constraint_name: Optional[str],
table_name: str,
condition: Union[str, BinaryExpression, TextClause],
+ *,
schema: Optional[str] = None,
**kw: Any,
) -> None:
referent_table: str,
local_cols: List[str],
remote_cols: List[str],
+ *,
onupdate: Optional[str] = None,
ondelete: Optional[str] = None,
deferrable: Optional[bool] = None,
index_name: Optional[str],
table_name: str,
columns: Sequence[Union[str, TextClause, Function[Any]]],
+ *,
schema: Optional[str] = None,
unique: bool = False,
**kw: Any,
constraint_name: Optional[str],
table_name: str,
columns: List[str],
+ *,
schema: Optional[str] = None,
) -> None:
"""Issue a "create primary key" instruction using the current
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.
constraint_name: Optional[str],
table_name: str,
columns: Sequence[str],
+ *,
schema: Optional[str] = None,
**kw: Any,
) -> Any:
self,
table_name: str,
column_name: str,
+ *,
schema: Optional[str] = None,
**kw: Any,
) -> None:
self,
constraint_name: str,
table_name: str,
+ *,
type_: Optional[str] = None,
schema: Optional[str] = None,
) -> None:
def drop_index(
self,
index_name: str,
+ *,
table_name: Optional[str] = None,
schema: Optional[str] = None,
**kw: Any,
...
def drop_table(
- self, table_name: str, schema: Optional[str] = None, **kw: Any
+ self, table_name: str, *, schema: Optional[str] = None, **kw: Any
) -> None:
r"""Issue a "drop table" instruction using the current
migration context.
def drop_table_comment(
self,
table_name: str,
+ *,
existing_comment: Optional[str] = None,
schema: Optional[str] = None,
) -> None:
self,
old_table_name: str,
new_table_name: str,
+ *,
schema: Optional[str] = None,
) -> None:
"""Emit an ALTER TABLE to rename a table.
def add_column(
self,
column: Column,
+ *,
insert_before: Optional[str] = None,
insert_after: Optional[str] = None,
) -> None:
def alter_column(
self,
column_name: str,
+ *,
nullable: Optional[bool] = None,
comment: Union[str, Literal[False], None] = False,
server_default: Any = False,
referent_table: str,
local_cols: List[str],
remote_cols: List[str],
+ *,
referent_schema: Optional[str] = None,
onupdate: Optional[str] = None,
ondelete: Optional[str] = None,
...
def drop_constraint(
- self, constraint_name: str, type_: Optional[str] = None
+ self, constraint_name: str, *, type_: Optional[str] = None
) -> None:
"""Issue a "drop constraint" instruction using the
current batch migration context.
self,
constraint_name: Optional[sqla_compat._ConstraintNameDefined],
table_name: str,
+ *,
type_: Optional[str] = None,
schema: Optional[str] = None,
_reverse: Optional[AddConstraintOp] = None,
operations: Operations,
constraint_name: str,
table_name: str,
+ *,
type_: Optional[str] = None,
schema: Optional[str] = None,
) -> None:
cls,
operations: BatchOperations,
constraint_name: str,
+ *,
type_: Optional[str] = None,
) -> None:
"""Issue a "drop constraint" instruction using the
constraint_name: Optional[sqla_compat._ConstraintNameDefined],
table_name: str,
columns: Sequence[str],
+ *,
schema: Optional[str] = None,
**kw: Any,
) -> None:
constraint_name: Optional[str],
table_name: str,
columns: List[str],
+ *,
schema: Optional[str] = None,
) -> None:
"""Issue a "create primary key" instruction using the current
:class:`~sqlalchemy.sql.elements.quoted_name`.
"""
- op = cls(constraint_name, table_name, columns, schema)
+ op = cls(constraint_name, table_name, columns, schema=schema)
return operations.invoke(op)
@classmethod
constraint_name: Optional[sqla_compat._ConstraintNameDefined],
table_name: str,
columns: Sequence[str],
+ *,
schema: Optional[str] = None,
**kw: Any,
) -> None:
constraint_name: Optional[str],
table_name: str,
columns: Sequence[str],
+ *,
schema: Optional[str] = None,
**kw: Any,
) -> Any:
referent_table: str,
local_cols: List[str],
remote_cols: List[str],
+ *,
onupdate: Optional[str] = None,
ondelete: Optional[str] = None,
deferrable: Optional[bool] = None,
referent_table: str,
local_cols: List[str],
remote_cols: List[str],
+ *,
referent_schema: Optional[str] = None,
onupdate: Optional[str] = None,
ondelete: Optional[str] = None,
constraint_name: Optional[sqla_compat._ConstraintNameDefined],
table_name: str,
condition: Union[str, TextClause, ColumnElement[Any]],
+ *,
schema: Optional[str] = None,
**kw: Any,
) -> None:
constraint_name: Optional[str],
table_name: str,
condition: Union[str, BinaryExpression, TextClause],
+ *,
schema: Optional[str] = None,
**kw: Any,
) -> None:
index_name: Optional[str],
table_name: str,
columns: Sequence[Union[str, TextClause, ColumnElement[Any]]],
+ *,
schema: Optional[str] = None,
unique: bool = False,
**kw: Any,
index_name: Optional[str],
table_name: str,
columns: Sequence[Union[str, TextClause, Function[Any]]],
+ *,
schema: Optional[str] = None,
unique: bool = False,
**kw: Any,
def __init__(
self,
index_name: Union[quoted_name, str, conv],
+ *,
table_name: Optional[str] = None,
schema: Optional[str] = None,
_reverse: Optional[CreateIndexOp] = None,
assert index.table is not None
return cls(
index.name, # type: ignore[arg-type]
- index.table.name,
+ table_name=index.table.name,
schema=index.table.schema,
_reverse=CreateIndexOp.from_index(index),
**index.kwargs,
cls,
operations: Operations,
index_name: str,
+ *,
table_name: Optional[str] = None,
schema: Optional[str] = None,
**kw: Any,
self,
table_name: str,
columns: Sequence[SchemaItem],
+ *,
schema: Optional[str] = None,
_namespace_metadata: Optional[MetaData] = None,
_constraints_included: bool = False,
@classmethod
def from_table(
- cls, table: Table, _namespace_metadata: Optional[MetaData] = None
+ cls, table: Table, *, _namespace_metadata: Optional[MetaData] = None
) -> CreateTableOp:
if _namespace_metadata is None:
_namespace_metadata = table.metadata
def __init__(
self,
table_name: str,
+ *,
schema: Optional[str] = None,
table_kw: Optional[MutableMapping[Any, Any]] = None,
_reverse: Optional[CreateTableOp] = None,
@classmethod
def from_table(
- cls, table: Table, _namespace_metadata: Optional[MetaData] = None
+ cls, table: Table, *, _namespace_metadata: Optional[MetaData] = None
) -> DropTableOp:
return cls(
table.name,
cls,
operations: Operations,
table_name: str,
+ *,
schema: Optional[str] = None,
**kw: Any,
) -> None:
def __init__(
self,
table_name: str,
+ *,
schema: Optional[str] = None,
) -> None:
self.table_name = table_name
self,
old_table_name: str,
new_table_name: str,
+ *,
schema: Optional[str] = None,
) -> None:
super().__init__(old_table_name, schema=schema)
operations: Operations,
old_table_name: str,
new_table_name: str,
+ *,
schema: Optional[str] = None,
) -> None:
"""Emit an ALTER TABLE to rename a table.
self,
table_name: str,
comment: Optional[str],
+ *,
schema: Optional[str] = None,
existing_comment: Optional[str] = None,
) -> None:
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.
def __init__(
self,
table_name: str,
+ *,
schema: Optional[str] = None,
existing_comment: Optional[str] = None,
) -> None:
cls,
operations: Operations,
table_name: str,
+ *,
existing_comment: Optional[str] = None,
schema: Optional[str] = None,
) -> None:
self,
table_name: str,
column_name: str,
+ *,
schema: Optional[str] = None,
existing_type: Optional[Any] = None,
existing_server_default: Any = False,
operations: Operations,
table_name: str,
column_name: str,
+ *,
nullable: Optional[bool] = None,
comment: Optional[Union[str, Literal[False]]] = False,
server_default: Any = False,
cls,
operations: BatchOperations,
column_name: str,
+ *,
nullable: Optional[bool] = None,
comment: Optional[Union[str, Literal[False]]] = False,
server_default: Any = False,
self,
table_name: str,
column: Column,
+ *,
schema: Optional[str] = None,
**kw: Any,
) -> None:
operations: Operations,
table_name: str,
column: Column,
+ *,
schema: Optional[str] = None,
) -> None:
"""Issue an "add column" instruction using the current
cls,
operations: BatchOperations,
column: Column,
+ *,
insert_before: Optional[str] = None,
insert_after: Optional[str] = None,
) -> None:
self,
table_name: str,
column_name: str,
+ *,
schema: Optional[str] = None,
_reverse: Optional[AddColumnOp] = None,
**kw: Any,
operations: Operations,
table_name: str,
column_name: str,
+ *,
schema: Optional[str] = None,
**kw: Any,
) -> None:
self,
table_name: str,
ops: Sequence[MigrateOperation],
+ *,
schema: Optional[str] = None,
) -> None:
super().__init__(ops)
rev_id: Optional[str],
upgrade_ops: UpgradeOps,
downgrade_ops: DowngradeOps,
+ *,
message: Optional[str] = None,
imports: Set[str] = set(),
head: Optional[str] = None,
--- /dev/null
+.. change::
+ :tags: change, py3k
+ :tickets: 1130
+
+ Argument signatures of Alembic operations now enforce keyword-only
+ arguments as passed as keyword and not positionally, such as
+ :paramref:`.Operations.create_table.schema`,
+ :paramref:`.Operations.add_column.type_`, etc.
with self.op.batch_alter_table(
"ck_table", recreate=recreate
) as batch_op:
- batch_op.drop_constraint("ck", "check")
+ batch_op.drop_constraint("ck", type_="check")
ck_consts = inspect(self.conn).get_check_constraints("ck_table")
eq_(ck_consts, [])
def test_drop_index(self):
context = op_fixture("mssql")
- op.drop_index("my_idx", "my_table")
+ op.drop_index("my_idx", table_name="my_table")
context.assert_contains("DROP INDEX my_idx ON my_table")
def test_drop_column_w_default(self):
def test_drop_fk(self):
context = op_fixture("mysql")
- op.drop_constraint("f1", "t1", "foreignkey")
+ op.drop_constraint("f1", "t1", type_="foreignkey")
context.assert_("ALTER TABLE t1 DROP FOREIGN KEY f1")
def test_drop_fk_quoted(self):
context = op_fixture("mysql")
- op.drop_constraint("MyFk", "MyTable", "foreignkey")
+ op.drop_constraint("MyFk", "MyTable", type_="foreignkey")
context.assert_("ALTER TABLE `MyTable` DROP FOREIGN KEY `MyFk`")
def test_drop_constraint_primary(self):
def test_drop_unique(self):
context = op_fixture("mysql")
- op.drop_constraint("f1", "t1", "unique")
+ op.drop_constraint("f1", "t1", type_="unique")
context.assert_("ALTER TABLE t1 DROP INDEX f1")
def test_drop_unique_quoted(self):
context = op_fixture("mysql")
- op.drop_constraint("MyUnique", "MyTable", "unique")
+ op.drop_constraint("MyUnique", "MyTable", type_="unique")
context.assert_("ALTER TABLE `MyTable` DROP INDEX `MyUnique`")
def test_drop_check_mariadb(self):
context = op_fixture("mariadb")
- op.drop_constraint("f1", "t1", "check")
+ op.drop_constraint("f1", "t1", type_="check")
context.assert_("ALTER TABLE t1 DROP CONSTRAINT f1")
def test_drop_check_quoted_mariadb(self):
context = op_fixture("mariadb")
- op.drop_constraint("MyCheck", "MyTable", "check")
+ op.drop_constraint("MyCheck", "MyTable", type_="check")
context.assert_("ALTER TABLE `MyTable` DROP CONSTRAINT `MyCheck`")
def test_drop_check_mysql(self):
context = op_fixture("mysql")
- op.drop_constraint("f1", "t1", "check")
+ op.drop_constraint("f1", "t1", type_="check")
context.assert_("ALTER TABLE t1 DROP CHECK f1")
def test_drop_check_quoted_mysql(self):
context = op_fixture("mysql")
- op.drop_constraint("MyCheck", "MyTable", "check")
+ op.drop_constraint("MyCheck", "MyTable", type_="check")
context.assert_("ALTER TABLE `MyTable` DROP CHECK `MyCheck`")
def test_drop_unknown(self):
op.drop_constraint,
"f1",
"t1",
- "typo",
+ type_="typo",
)
def test_drop_generic_constraint(self):
def test_drop_index(self):
context = op_fixture("oracle")
- op.drop_index("my_idx", "my_table")
+ op.drop_index("my_idx", table_name="my_table")
context.assert_contains("DROP INDEX my_idx")
def test_drop_column_w_default(self):
op.create_index("i", "t", ["c1", "c2"], unique=False)
context.assert_("CREATE INDEX i ON t (c1, c2)")
- def test_drop_index_postgresql_concurrently(self):
+ @config.combinations("include_table", "no_table", argnames="include_table")
+ def test_drop_index_postgresql_concurrently(self, include_table):
context = op_fixture("postgresql")
- op.drop_index("geocoded", "locations", postgresql_concurrently=True)
+ if include_table == "include_table":
+ op.drop_index(
+ "geocoded",
+ table_name="locations",
+ postgresql_concurrently=True,
+ )
+ else:
+ op.drop_index("geocoded", postgresql_concurrently=True)
context.assert_("DROP INDEX CONCURRENTLY geocoded")
def test_alter_column_type_using(self):