Support IF EXISTS/IF NOT EXISTS for DDL constructs

Added parameters :paramref:`_ddl.CreateTable.if_not_exists`,
:paramref:`_ddl.CreateIndex.if_not_exists`,
:paramref:`_ddl.DropTable.if_exists` and
:paramref:`_ddl.DropIndex.if_exists` to the :class:`_ddl.CreateTable`,
:class:`_ddl.DropTable`, :class:`_ddl.CreateIndex` and
:class:`_ddl.DropIndex` constructs which result in "IF NOT EXISTS" / "IF
EXISTS" DDL being added to the CREATE/DROP. These phrases are not accepted
by all databases and the operation will fail on a database that does not
support it as there is no similarly compatible fallback within the scope of
a single DDL statement.  Pull request courtesy Ramon Williams.

Fixes: #2843
Closes: #5663
Pull-request: https://github.com/sqlalchemy/sqlalchemy/pull/5663
Pull-request-sha: 748b8472345d96efb446e2a444fbe020b313669f

Change-Id: I6a2b1f697993ed49c31584f0a31887fb0a868ed3

10 files changed, 272 insertions, 12 deletions

diff --git a/doc/build/changelog/unreleased_14/2843.rst b/doc/build/changelog/unreleased_14/2843.rst
new file mode 100644
index 000000000..9d227754e
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/2843.rst
@@ -0,0 +1,14 @@
+.. change::
+    :tags: schema, usecase
+    :tickets: 2843
+
+    Added parameters :paramref:`_ddl.CreateTable.if_not_exists`,
+    :paramref:`_ddl.CreateIndex.if_not_exists`,
+    :paramref:`_ddl.DropTable.if_exists` and
+    :paramref:`_ddl.DropIndex.if_exists` to the :class:`_ddl.CreateTable`,
+    :class:`_ddl.DropTable`, :class:`_ddl.CreateIndex` and
+    :class:`_ddl.DropIndex` constructs which result in "IF NOT EXISTS" / "IF
+    EXISTS" DDL being added to the CREATE/DROP. These phrases are not accepted
+    by all databases and the operation will fail on a database that does not
+    support it as there is no similarly compatible fallback within the scope of
+    a single DDL statement.  Pull request courtesy Ramon Williams.
\ No newline at end of file
diff --git a/lib/sqlalchemy/dialects/mysql/base.py b/lib/sqlalchemy/dialects/mysql/base.py
index 27c7b2239..ad029299b 100644
--- a/lib/sqlalchemy/dialects/mysql/base.py
+++ b/lib/sqlalchemy/dialects/mysql/base.py
@@ -2031,7 +2031,10 @@ class MySQLDDLCompiler(compiler.DDLCompiler):
         if index_prefix:
             text += index_prefix + " "
 
-        text += "INDEX %s ON %s " % (name, table)
+        text += "INDEX "
+        if create.if_not_exists:
+            text += "IF NOT EXISTS "
+        text += "%s ON %s " % (name, table)
 
         length = index.dialect_options[self.dialect.name]["length"]
         if length is not None:
@@ -2081,8 +2084,11 @@ class MySQLDDLCompiler(compiler.DDLCompiler):
 
     def visit_drop_index(self, drop):
         index = drop.element
+        text = "\nDROP INDEX "
+        if drop.if_exists:
+            text += "IF EXISTS "
 
-        return "\nDROP INDEX %s ON %s" % (
+        return text + "%s ON %s" % (
             self._prepared_index_name(index, include_schema=False),
             self.preparer.format_table(index.table),
         )
diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index bf257ab3f..5292b3901 100644
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -2322,6 +2322,9 @@ class PGDDLCompiler(compiler.DDLCompiler):
             if concurrently:
                 text += "CONCURRENTLY "
 
+        if create.if_not_exists:
+            text += "IF NOT EXISTS "
+
         text += "%s ON %s " % (
             self._prepared_index_name(index, include_schema=False),
             preparer.format_table(index.table),
@@ -2405,6 +2408,9 @@ class PGDDLCompiler(compiler.DDLCompiler):
             if concurrently:
                 text += "CONCURRENTLY "
 
+        if drop.if_exists:
+            text += "IF EXISTS "
+
         text += self._prepared_index_name(index, include_schema=True)
         return text
 
diff --git a/lib/sqlalchemy/dialects/sqlite/base.py b/lib/sqlalchemy/dialects/sqlite/base.py
index fc08b4b5e..1bd249efc 100644
--- a/lib/sqlalchemy/dialects/sqlite/base.py
+++ b/lib/sqlalchemy/dialects/sqlite/base.py
@@ -1543,7 +1543,13 @@ class SQLiteDDLCompiler(compiler.DDLCompiler):
         text = "CREATE "
         if index.unique:
             text += "UNIQUE "
-        text += "INDEX %s ON %s (%s)" % (
+
+        text += "INDEX "
+
+        if create.if_not_exists:
+            text += "IF NOT EXISTS "
+
+        text += "%s ON %s (%s)" % (
             self._prepared_index_name(index, include_schema=True),
             preparer.format_table(index.table, use_schema=False),
             ", ".join(
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index 9b90bf868..9c3d1e080 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -3835,7 +3835,12 @@ class DDLCompiler(Compiled):
         text = "\nCREATE "
         if table._prefixes:
             text += " ".join(table._prefixes) + " "
-        text += "TABLE " + preparer.format_table(table) + " "
+
+        text += "TABLE "
+        if create.if_not_exists:
+            text += "IF NOT EXISTS "
+
+        text += preparer.format_table(table) + " "
 
         create_table_suffix = self.create_table_suffix(table)
         if create_table_suffix:
@@ -3935,7 +3940,10 @@ class DDLCompiler(Compiled):
         )
 
     def visit_drop_table(self, drop, **kw):
-        return "\nDROP TABLE " + self.preparer.format_table(drop.element)
+        text = "\nDROP TABLE "
+        if drop.if_exists:
+            text += "IF EXISTS "
+        return text + self.preparer.format_table(drop.element)
 
     def visit_drop_view(self, drop, **kw):
         return "\nDROP VIEW " + self.preparer.format_table(drop.element)
@@ -3959,7 +3967,12 @@ class DDLCompiler(Compiled):
             raise exc.CompileError(
                 "CREATE INDEX requires that the index have a name"
             )
-        text += "INDEX %s ON %s (%s)" % (
+
+        text += "INDEX "
+        if create.if_not_exists:
+            text += "IF NOT EXISTS "
+
+        text += "%s ON %s (%s)" % (
             self._prepared_index_name(index, include_schema=include_schema),
             preparer.format_table(
                 index.table, use_schema=include_table_schema
@@ -3980,9 +3993,11 @@ class DDLCompiler(Compiled):
             raise exc.CompileError(
                 "DROP INDEX requires that the index have a name"
             )
-        return "\nDROP INDEX " + self._prepared_index_name(
-            index, include_schema=True
-        )
+        text = "\nDROP INDEX "
+        if drop.if_exists:
+            text += "IF EXISTS "
+
+        return text + self._prepared_index_name(index, include_schema=True)
 
     def _prepared_index_name(self, index, include_schema=False):
         if index.table is not None:
diff --git a/lib/sqlalchemy/sql/ddl.py b/lib/sqlalchemy/sql/ddl.py
index f1012292b..564b53c55 100644
--- a/lib/sqlalchemy/sql/ddl.py
+++ b/lib/sqlalchemy/sql/ddl.py
@@ -369,9 +369,13 @@ class _CreateDropBase(DDLElement):
 
     """
 
-    def __init__(self, element, bind=None):
+    def __init__(
+        self, element, bind=None, if_exists=False, if_not_exists=False
+    ):
         self.element = element
         self.bind = bind
+        self.if_exists = if_exists
+        self.if_not_exists = if_not_exists
 
     @property
     def stringify_dialect(self):
@@ -427,7 +431,11 @@ class CreateTable(_CreateDropBase):
     __visit_name__ = "create_table"
 
     def __init__(
-        self, element, bind=None, include_foreign_key_constraints=None
+        self,
+        element,
+        bind=None,
+        include_foreign_key_constraints=None,
+        if_not_exists=False,
     ):
         """Create a :class:`.CreateTable` construct.
 
@@ -442,8 +450,15 @@ class CreateTable(_CreateDropBase):
 
          .. versionadded:: 1.0.0
 
+        :param if_not_exists: if True, an IF NOT EXISTS operator will be
+         applied to the construct.
+
+         .. versionadded:: 1.4.0b2
+
         """
-        super(CreateTable, self).__init__(element, bind=bind)
+        super(CreateTable, self).__init__(
+            element, bind=bind, if_not_exists=if_not_exists
+        )
         self.columns = [CreateColumn(column) for column in element.columns]
         self.include_foreign_key_constraints = include_foreign_key_constraints
 
@@ -573,6 +588,23 @@ class DropTable(_CreateDropBase):
 
     __visit_name__ = "drop_table"
 
+    def __init__(self, element, bind=None, if_exists=False):
+        """Create a :class:`.DropTable` construct.
+
+        :param element: a :class:`_schema.Table` that's the subject
+         of the DROP.
+        :param on: See the description for 'on' in :class:`.DDL`.
+        :param bind: See the description for 'bind' in :class:`.DDL`.
+        :param if_exists: if True, an IF EXISTS operator will be applied to the
+         construct.
+
+         .. versionadded:: 1.4.0b2
+
+        """
+        super(DropTable, self).__init__(
+            element, bind=bind, if_exists=if_exists
+        )
+
 
 class CreateSequence(_CreateDropBase):
     """Represent a CREATE SEQUENCE statement."""
@@ -591,12 +623,46 @@ class CreateIndex(_CreateDropBase):
 
     __visit_name__ = "create_index"
 
+    def __init__(self, element, bind=None, if_not_exists=False):
+        """Create a :class:`.Createindex` construct.
+
+        :param element: a :class:`_schema.Index` that's the subject
+         of the CREATE.
+        :param on: See the description for 'on' in :class:`.DDL`.
+        :param bind: See the description for 'bind' in :class:`.DDL`.
+        :param if_not_exists: if True, an IF NOT EXISTS operator will be
+         applied to the construct.
+
+         .. versionadded:: 1.4.0b2
+
+        """
+        super(CreateIndex, self).__init__(
+            element, bind=bind, if_not_exists=if_not_exists
+        )
+
 
 class DropIndex(_CreateDropBase):
     """Represent a DROP INDEX statement."""
 
     __visit_name__ = "drop_index"
 
+    def __init__(self, element, bind=None, if_exists=False):
+        """Create a :class:`.DropIndex` construct.
+
+        :param element: a :class:`_schema.Index` that's the subject
+         of the DROP.
+        :param on: See the description for 'on' in :class:`.DDL`.
+        :param bind: See the description for 'bind' in :class:`.DDL`.
+        :param if_exists: if True, an IF EXISTS operator will be applied to the
+         construct.
+
+         .. versionadded:: 1.4.0b2
+
+        """
+        super(DropIndex, self).__init__(
+            element, bind=bind, if_exists=if_exists
+        )
+
 
 class AddConstraint(_CreateDropBase):
     """Represent an ALTER TABLE ADD CONSTRAINT statement."""
diff --git a/lib/sqlalchemy/testing/requirements.py b/lib/sqlalchemy/testing/requirements.py
index bd2d4eaf9..8cd5999a2 100644
--- a/lib/sqlalchemy/testing/requirements.py
+++ b/lib/sqlalchemy/testing/requirements.py
@@ -40,6 +40,18 @@ class SuiteRequirements(Requirements):
         return exclusions.open()
 
     @property
+    def table_ddl_if_exists(self):
+        """target platform supports IF NOT EXISTS / IF EXISTS for tables."""
+
+        return exclusions.closed()
+
+    @property
+    def index_ddl_if_exists(self):
+        """target platform supports IF NOT EXISTS / IF EXISTS for indexes."""
+
+        return exclusions.closed()
+
+    @property
     def foreign_keys(self):
         """Target database must support foreign keys."""
 
diff --git a/lib/sqlalchemy/testing/suite/test_ddl.py b/lib/sqlalchemy/testing/suite/test_ddl.py
index a6f15a72d..c3cf854e4 100644
--- a/lib/sqlalchemy/testing/suite/test_ddl.py
+++ b/lib/sqlalchemy/testing/suite/test_ddl.py
@@ -2,8 +2,11 @@ from .. import config
 from .. import fixtures
 from .. import util
 from ..assertions import eq_
+from ..assertions import is_false
+from ..assertions import is_true
 from ..config import requirements
 from ... import Column
+from ... import Index
 from ... import inspect
 from ... import Integer
 from ... import schema
@@ -31,6 +34,11 @@ class TableDDLTest(fixtures.TestBase):
             Column("_data", String(50)),
         )
 
+    def _table_index_fixture(self, schema=None):
+        table = self._simple_fixture(schema=schema)
+        idx = Index("test_index", table.c.data)
+        return table, idx
+
     def _simple_roundtrip(self, table):
         with config.db.begin() as conn:
             conn.execute(table.insert().values((1, "some data")))
@@ -90,6 +98,85 @@ class TableDDLTest(fixtures.TestBase):
             inspect(connection).get_table_comment("test_table"), {"text": None}
         )
 
+    @requirements.table_ddl_if_exists
+    @util.provide_metadata
+    def test_create_table_if_not_exists(self, connection):
+        table = self._simple_fixture()
+
+        connection.execute(schema.CreateTable(table, if_not_exists=True))
+
+        is_true(inspect(connection).has_table("test_table"))
+        connection.execute(schema.CreateTable(table, if_not_exists=True))
+
+    @requirements.index_ddl_if_exists
+    @util.provide_metadata
+    def test_create_index_if_not_exists(self, connection):
+        table, idx = self._table_index_fixture()
+
+        connection.execute(schema.CreateTable(table, if_not_exists=True))
+        is_true(inspect(connection).has_table("test_table"))
+        is_false(
+            "test_index"
+            in [
+                ix["name"]
+                for ix in inspect(connection).get_indexes("test_table")
+            ]
+        )
+
+        connection.execute(schema.CreateIndex(idx, if_not_exists=True))
+
+        is_true(
+            "test_index"
+            in [
+                ix["name"]
+                for ix in inspect(connection).get_indexes("test_table")
+            ]
+        )
+
+        connection.execute(schema.CreateIndex(idx, if_not_exists=True))
+
+    @requirements.table_ddl_if_exists
+    @util.provide_metadata
+    def test_drop_table_if_exists(self, connection):
+        table = self._simple_fixture()
+
+        table.create(connection)
+
+        is_true(inspect(connection).has_table("test_table"))
+
+        connection.execute(schema.DropTable(table, if_exists=True))
+
+        is_false(inspect(connection).has_table("test_table"))
+
+        connection.execute(schema.DropTable(table, if_exists=True))
+
+    @requirements.index_ddl_if_exists
+    @util.provide_metadata
+    def test_drop_index_if_exists(self, connection):
+        table, idx = self._table_index_fixture()
+
+        table.create(connection)
+
+        is_true(
+            "test_index"
+            in [
+                ix["name"]
+                for ix in inspect(connection).get_indexes("test_table")
+            ]
+        )
+
+        connection.execute(schema.DropIndex(idx, if_exists=True))
+
+        is_false(
+            "test_index"
+            in [
+                ix["name"]
+                for ix in inspect(connection).get_indexes("test_table")
+            ]
+        )
+
+        connection.execute(schema.DropIndex(idx, if_exists=True))
+
 
 class FutureTableDDLTest(fixtures.FutureEngineMixin, TableDDLTest):
     pass
diff --git a/test/requirements.py b/test/requirements.py
index 2cdc3baca..fe1ca082e 100644
--- a/test/requirements.py
+++ b/test/requirements.py
@@ -78,6 +78,19 @@ class DefaultRequirements(SuiteRequirements):
         return skip_if(no_support("sqlite", "not supported by database"))
 
     @property
+    def table_ddl_if_exists(self):
+        """target platform supports IF NOT EXISTS / IF EXISTS for tables."""
+
+        return only_on(["postgresql", "mysql", "mariadb", "sqlite"])
+
+    @property
+    def index_ddl_if_exists(self):
+        """target platform supports IF NOT EXISTS / IF EXISTS for indexes."""
+
+        # mariadb but not mysql, tested up to mysql 8
+        return only_on(["postgresql", "mariadb", "sqlite"])
+
+    @property
     def on_update_cascade(self):
         """target database must support ON UPDATE..CASCADE behavior in
         foreign keys."""
diff --git a/test/sql/test_compiler.py b/test/sql/test_compiler.py
index 75ac896e9..fdffe04bf 100644
--- a/test/sql/test_compiler.py
+++ b/test/sql/test_compiler.py
@@ -31,6 +31,7 @@ from sqlalchemy import exists
 from sqlalchemy import Float
 from sqlalchemy import ForeignKey
 from sqlalchemy import func
+from sqlalchemy import Index
 from sqlalchemy import Integer
 from sqlalchemy import intersect
 from sqlalchemy import join
@@ -4471,6 +4472,40 @@ class DDLTest(fixtures.TestBase, AssertsCompiledSQL):
             "PRIMARY KEY (a, b))",
         )
 
+    def test_create_table_exists(self):
+        m = MetaData()
+        t1 = Table("t1", m, Column("q", Integer))
+        self.assert_compile(
+            schema.CreateTable(t1, if_not_exists=True),
+            "CREATE TABLE IF NOT EXISTS t1 (q INTEGER)",
+        )
+
+    def test_drop_table_exists(self):
+        m = MetaData()
+        t1 = Table("t1", m, Column("q", Integer))
+        self.assert_compile(
+            schema.DropTable(t1, if_exists=True),
+            "DROP TABLE IF EXISTS t1",
+        )
+
+    def test_create_index_exists(self):
+        m = MetaData()
+        t1 = Table("t1", m, Column("q", Integer))
+        idx = Index("my_idx", t1.c.q)
+        self.assert_compile(
+            schema.CreateIndex(idx, if_not_exists=True),
+            "CREATE INDEX IF NOT EXISTS my_idx ON t1 (q)",
+        )
+
+    def test_drop_index_exists(self):
+        m = MetaData()
+        t1 = Table("t1", m, Column("q", Integer))
+        idx = Index("my_idx", t1.c.q)
+        self.assert_compile(
+            schema.DropIndex(idx, if_exists=True),
+            "DROP INDEX IF EXISTS my_idx",
+        )
+
     def test_create_table_suffix(self):
         class MyDialect(default.DefaultDialect):
             class MyCompiler(compiler.DDLCompiler):