table, self.connection, checkfirst=False, _ddl_runner=self
)
- def create_index(self, index: Index) -> None:
- self._exec(schema.CreateIndex(index))
+ def create_index(self, index: Index, **kw: Any) -> None:
+ self._exec(schema.CreateIndex(index, **kw))
def create_table_comment(self, table: Table) -> None:
self._exec(schema.SetTableComment(table))
def create_column_comment(self, column: ColumnElement[Any]) -> None:
self._exec(schema.SetColumnComment(column))
- def drop_index(self, index: Index) -> None:
- self._exec(schema.DropIndex(index))
+ def drop_index(self, index: Index, **kw: Any) -> None:
+ self._exec(schema.DropIndex(index, **kw))
def bulk_insert(
self,
table_name, column_name, schema=schema, name=name
)
- def create_index(self, index: Index) -> None:
+ def create_index(self, index: Index, **kw: Any) -> None:
# this likely defaults to None if not present, so get()
# should normally not return the default value. being
# defensive in any case
for col in mssql_include:
if col not in index.table.c:
index.table.append_column(Column(col, sqltypes.NullType))
- self._exec(CreateIndex(index))
+ self._exec(CreateIndex(index, **kw))
def bulk_insert( # type:ignore[override]
self, table: Union[TableClause, Table], rows: List[dict], **kw: Any
)
identity_attrs_ignore = ("on_null", "order")
- def create_index(self, index):
+ def create_index(self, index: Index, **kw: Any) -> None:
# this likely defaults to None if not present, so get()
# should normally not return the default value. being
# defensive in any case
postgresql_include = index.kwargs.get("postgresql_include", None) or ()
for col in postgresql_include:
- if col not in index.table.c:
- index.table.append_column(Column(col, sqltypes.NullType))
- self._exec(CreateIndex(index))
+ if col not in index.table.c: # type: ignore[union-attr]
+ index.table.append_column( # type: ignore[union-attr]
+ Column(col, sqltypes.NullType)
+ )
+ self._exec(CreateIndex(index, **kw))
def prep_table_for_batch(self, batch_impl, table):
for constraint in table.constraints:
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>``.
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
:param \**kw: Additional keyword arguments not mentioned above are
dialect specific, and passed in the form
``<dialectname>_<argname>``.
def rename_table(self, *arg, **kw):
self.batch.append(("rename_table", arg, kw))
- def create_index(self, idx: Index) -> None:
- self.batch.append(("create_index", (idx,), {}))
+ def create_index(self, idx: Index, **kw: Any) -> None:
+ self.batch.append(("create_index", (idx,), kw))
- def drop_index(self, idx: Index) -> None:
- self.batch.append(("drop_index", (idx,), {}))
+ def drop_index(self, idx: Index, **kw: Any) -> None:
+ self.batch.append(("drop_index", (idx,), kw))
def create_table_comment(self, table):
self.batch.append(("create_table_comment", (table,), {}))
*,
schema: Optional[str] = None,
unique: bool = False,
+ if_not_exists: Optional[bool] = None,
**kw: Any,
) -> None:
self.index_name = index_name
self.columns = columns
self.schema = schema
self.unique = unique
+ self.if_not_exists = if_not_exists
self.kw = kw
def reverse(self) -> DropIndexOp:
*,
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
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>``.
"""
op = cls(
- index_name, table_name, columns, schema=schema, unique=unique, **kw
+ index_name,
+ table_name,
+ columns,
+ schema=schema,
+ unique=unique,
+ if_not_exists=if_not_exists,
+ **kw,
)
return operations.invoke(op)
table_name: Optional[str] = None,
*,
schema: Optional[str] = None,
+ if_exists: Optional[bool] = None,
_reverse: Optional[CreateIndexOp] = None,
**kw: Any,
) -> None:
self.index_name = index_name
self.table_name = table_name
self.schema = schema
+ self.if_exists = if_exists
self._reverse = _reverse
self.kw = kw
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
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
+
:param \**kw: Additional keyword arguments not mentioned above are
dialect specific, and passed in the form
``<dialectname>_<argname>``.
:ref:`dialect_toplevel` for detail on documented arguments.
"""
- op = cls(index_name, table_name=table_name, schema=schema, **kw)
+ op = cls(
+ index_name,
+ table_name=table_name,
+ schema=schema,
+ if_exists=if_exists,
+ **kw,
+ )
return operations.invoke(op)
@classmethod
from . import ops
from .base import Operations
from ..util.sqla_compat import _copy
+from ..util.sqla_compat import sqla_2
if TYPE_CHECKING:
from sqlalchemy.sql.schema import Table
existing_nullable=existing_nullable,
comment=comment,
existing_comment=existing_comment,
- **operation.kw
+ **operation.kw,
)
if type_:
operations: "Operations", operation: "ops.CreateIndexOp"
) -> None:
idx = operation.to_index(operations.migration_context)
- operations.impl.create_index(idx)
+ kw = {}
+ if operation.if_not_exists is not None:
+ if not sqla_2:
+ raise NotImplementedError("SQLAlchemy 2.0+ required")
+
+ kw["if_not_exists"] = operation.if_not_exists
+ operations.impl.create_index(idx, **kw)
@Operations.implementation_for(ops.DropIndexOp)
def drop_index(operations: "Operations", operation: "ops.DropIndexOp") -> None:
+ kw = {}
+ if operation.if_exists is not None:
+ if not sqla_2:
+ raise NotImplementedError("SQLAlchemy 2.0+ required")
+
+ kw["if_exists"] = operation.if_exists
+
operations.impl.drop_index(
- operation.to_index(operations.migration_context)
+ operation.to_index(operations.migration_context),
+ **kw,
)
op.create_index("ik_test", "t1", ["foo", "bar"])
context.assert_("CREATE INDEX ik_test ON t1 (foo, bar)")
+ @config.requirements.sqlalchemy_2
+ def test_create_index_if_not_exists(self):
+ context = op_fixture()
+ op.create_index("ik_test", "t1", ["foo", "bar"], if_not_exists=True)
+ context.assert_("CREATE INDEX IF NOT EXISTS ik_test ON t1 (foo, bar)")
+
def test_create_unique_index(self):
context = op_fixture()
op.create_index("ik_test", "t1", ["foo", "bar"], unique=True)
op.drop_index("ik_test", schema="foo")
context.assert_("DROP INDEX foo.ik_test")
+ @config.requirements.sqlalchemy_2
+ def test_drop_index_if_exists(self):
+ context = op_fixture()
+ op.drop_index("ik_test", if_exists=True)
+ context.assert_("DROP INDEX IF EXISTS ik_test")
+
def test_drop_table(self):
context = op_fixture()
op.drop_table("tb_test")
op.create_index("i", "t", ["c1", "c2"], unique=False)
context.assert_("CREATE INDEX i ON t (c1, c2)")
+ @config.requirements.sqlalchemy_2
+ def test_create_index_postgresql_if_not_exists(self):
+ context = op_fixture("postgresql")
+ op.create_index("i", "t", ["c1", "c2"], if_not_exists=True)
+ context.assert_("CREATE INDEX IF NOT EXISTS i ON t (c1, c2)")
+
@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", postgresql_concurrently=True)
context.assert_("DROP INDEX CONCURRENTLY geocoded")
+ @config.requirements.sqlalchemy_2
+ def test_drop_index_postgresql_if_exists(self):
+ context = op_fixture("postgresql")
+ op.drop_index("geocoded", if_exists=True)
+ context.assert_("DROP INDEX IF EXISTS geocoded")
+
def test_alter_column_type_using(self):
context = op_fixture("postgresql")
op.alter_column("t", "c", type_=Integer, postgresql_using="c::integer")
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
