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

2 files changed, 99 insertions, 0 deletions

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