base.RenameTable(old_table_name, new_table_name, schema=schema)
)
- def create_table(self, table: Table) -> None:
+ def create_table(self, table: Table, **kw: Any) -> None:
table.dispatch.before_create(
table, self.connection, checkfirst=False, _ddl_runner=self
)
- self._exec(schema.CreateTable(table))
+ self._exec(schema.CreateTable(table, **kw))
table.dispatch.after_create(
table, self.connection, checkfirst=False, _ddl_runner=self
)
if comment and with_comment:
self.create_column_comment(column)
- def drop_table(self, table: Table) -> None:
+ def drop_table(self, table: Table, **kw: Any) -> None:
table.dispatch.before_drop(
table, self.connection, checkfirst=False, _ddl_runner=self
)
- self._exec(schema.DropTable(table))
+ self._exec(schema.DropTable(table, **kw))
table.dispatch.after_drop(
table, self.connection, checkfirst=False, _ddl_runner=self
)
"""
-def create_table(table_name: str, *columns: SchemaItem, **kw: Any) -> Table:
+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.
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.
"""
def drop_table(
- table_name: str, *, schema: Optional[str] = None, **kw: Any
+ 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.
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 create_table(
- self, table_name: str, *columns: SchemaItem, **kw: Any
+ self,
+ 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.
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.
...
def drop_table(
- self, table_name: str, *, schema: Optional[str] = None, **kw: Any
+ self,
+ 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.
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.
columns: Sequence[SchemaItem],
*,
schema: Optional[str] = None,
+ if_not_exists: Optional[bool] = None,
_namespace_metadata: Optional[MetaData] = None,
_constraints_included: bool = False,
**kw: Any,
self.table_name = table_name
self.columns = columns
self.schema = schema
+ self.if_not_exists = if_not_exists
self.info = kw.pop("info", {})
self.comment = kw.pop("comment", None)
self.prefixes = kw.pop("prefixes", None)
operations: Operations,
table_name: str,
*columns: SchemaItem,
+ if_not_exists: Optional[bool] = None,
**kw: Any,
) -> Table:
r"""Issue a "create table" instruction using the current migration
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.
to the parameters given.
"""
- op = cls(table_name, columns, **kw)
+ op = cls(table_name, columns, if_not_exists=if_not_exists, **kw)
return operations.invoke(op)
table_name: str,
*,
schema: Optional[str] = None,
+ if_exists: Optional[bool] = None,
table_kw: Optional[MutableMapping[Any, Any]] = None,
_reverse: Optional[CreateTableOp] = None,
) -> None:
self.table_name = table_name
self.schema = schema
+ self.if_exists = if_exists
self.table_kw = table_kw or {}
self.comment = self.table_kw.pop("comment", None)
self.info = self.table_kw.pop("info", None)
table_name: str,
*,
schema: Optional[str] = None,
+ if_exists: Optional[bool] = None,
**kw: Any,
) -> None:
r"""Issue a "drop table" instruction using the current
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.
"""
- op = cls(table_name, schema=schema, table_kw=kw)
+ op = cls(table_name, schema=schema, if_exists=if_exists, table_kw=kw)
operations.invoke(op)
@Operations.implementation_for(ops.DropTableOp)
def drop_table(operations: "Operations", operation: "ops.DropTableOp") -> None:
+ kw = {}
+ if operation.if_exists is not None:
+ if not sqla_14:
+ raise NotImplementedError("SQLAlchemy 1.4+ required")
+
+ kw["if_exists"] = operation.if_exists
operations.impl.drop_table(
- operation.to_table(operations.migration_context)
+ operation.to_table(operations.migration_context), **kw
)
def create_table(
operations: "Operations", operation: "ops.CreateTableOp"
) -> "Table":
+ kw = {}
+ if operation.if_not_exists is not None:
+ if not sqla_14:
+ raise NotImplementedError("SQLAlchemy 1.4+ required")
+
+ kw["if_not_exists"] = operation.if_not_exists
table = operation.to_table(operations.migration_context)
- operations.impl.create_table(table)
+ operations.impl.create_table(table, **kw)
return table
--- /dev/null
+.. change::
+ :tags: usecase, operations
+ :tickets: 1520
+
+ Added support for :paramref:`.Operations.create_table.if_not_exists` and
+ :paramref:`.Operations.drop_table.if_exists`, adding similar functionality
+ to render IF [NOT] EXISTS for table operations in a similar way as with
+ indexes. Pull request courtesy Aaron Griffin.
+
op.drop_table("tb_test", schema="foo")
context.assert_("DROP TABLE foo.tb_test")
+ @config.requirements.sqlalchemy_14
+ def test_drop_table_if_exists(self):
+ context = op_fixture()
+ op.drop_table("tb_test", if_exists=True)
+ context.assert_("DROP TABLE IF EXISTS tb_test")
+
def test_create_table_selfref(self):
context = op_fixture()
op.create_table(
"FOREIGN KEY(foo_bar) REFERENCES foo (bar))"
)
+ @config.requirements.sqlalchemy_14
+ def test_create_table_if_not_exists(self):
+ context = op_fixture()
+ op.create_table(
+ "some_table",
+ Column("id", Integer, primary_key=True),
+ if_not_exists=True,
+ )
+ context.assert_(
+ "CREATE TABLE IF NOT EXISTS some_table ("
+ "id INTEGER NOT NULL, "
+ "PRIMARY KEY (id))"
+ )
+
def test_execute_delete(self):
context = op_fixture()
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
