Deprecate connection branching

The :meth:`.Connection.connect` method is deprecated as is the concept of
"connection branching", which copies a :class:`.Connection` into a new one
that has a no-op ".close()" method.  This pattern is oriented around the
"connectionless execution" concept which is also being removed in 2.0.

As part of this change we begin to move the internals away from
"connectionless execution" overall.   Remove the "connectionless
execution" concept from the reflection internals and replace with
explicit patterns at the Inspector level.

Fixes: #5131
Change-Id: Id23d28a9889212ac5ae7329b85136157815d3e6f

4 files changed, 136 insertions, 90 deletions

diff --git a/test/engine/test_bind.py b/test/engine/test_bind.py
index 39acfed2c..956874846 100644
--- a/test/engine/test_bind.py
+++ b/test/engine/test_bind.py
@@ -26,14 +26,6 @@ class BindTest(fixtures.TestBase):
             assert not conn.closed
         assert conn.closed
 
-    def test_bind_close_conn(self):
-        e = testing.db
-        conn = e.connect()
-        with conn.connect() as c2:
-            assert not c2.closed
-        assert not conn.closed
-        assert c2.closed
-
     def test_create_drop_explicit(self):
         metadata = MetaData()
         table = Table("test_table", metadata, Column("foo", Integer))
diff --git a/test/engine/test_deprecations.py b/test/engine/test_deprecations.py
index 6fbf1be5b..884838351 100644
--- a/test/engine/test_deprecations.py
+++ b/test/engine/test_deprecations.py
@@ -13,6 +13,7 @@ from sqlalchemy import String
 from sqlalchemy import testing
 from sqlalchemy import text
 from sqlalchemy import TypeDecorator
+from sqlalchemy.engine import reflection
 from sqlalchemy.engine.base import Engine
 from sqlalchemy.engine.mock import MockConnection
 from sqlalchemy.testing import assert_raises
@@ -20,6 +21,7 @@ from sqlalchemy.testing import assert_raises_message
 from sqlalchemy.testing import engines
 from sqlalchemy.testing import eq_
 from sqlalchemy.testing import fixtures
+from sqlalchemy.testing import is_
 from sqlalchemy.testing import is_false
 from sqlalchemy.testing import is_true
 from sqlalchemy.testing.mock import Mock
@@ -31,6 +33,52 @@ class SomeException(Exception):
     pass
 
 
+class ConnectionlessDeprecationTest(fixtures.TestBase):
+    """test various things associated with "connectionless" executions."""
+
+    def test_inspector_constructor_engine(self):
+        with testing.expect_deprecated(
+            r"The __init__\(\) method on Inspector is deprecated and will "
+            r"be removed in a future release."
+        ):
+            i1 = reflection.Inspector(testing.db)
+
+        is_(i1.bind, testing.db)
+
+    def test_inspector_constructor_connection(self):
+        with testing.db.connect() as conn:
+            with testing.expect_deprecated(
+                r"The __init__\(\) method on Inspector is deprecated and "
+                r"will be removed in a future release."
+            ):
+                i1 = reflection.Inspector(conn)
+
+            is_(i1.bind, conn)
+            is_(i1.engine, testing.db)
+
+    def test_inspector_from_engine(self):
+        with testing.expect_deprecated(
+            r"The from_engine\(\) method on Inspector is deprecated and will "
+            r"be removed in a future release."
+        ):
+            i1 = reflection.Inspector.from_engine(testing.db)
+
+        is_(i1.bind, testing.db)
+
+    def test_bind_close_conn(self):
+        e = testing.db
+        conn = e.connect()
+
+        with testing.expect_deprecated(
+            r"The .close\(\) method on a so-called 'branched' "
+            "connection is deprecated"
+        ):
+            with conn.connect() as c2:
+                assert not c2.closed
+        assert not conn.closed
+        assert c2.closed
+
+
 class CreateEngineTest(fixtures.TestBase):
     def test_strategy_keyword_mock(self):
         def executor(x, y):
@@ -512,3 +560,17 @@ class DeprecatedReflectionTest(fixtures.TablesTest):
         ):
             table_names = testing.db.table_names()
         is_true(set(table_names).issuperset(metadata.tables))
+
+
+class ExecutionOptionsTest(fixtures.TestBase):
+    def test_branched_connection_execution_options(self):
+        engine = engines.testing_engine("sqlite://")
+
+        conn = engine.connect()
+        c2 = conn.execution_options(foo="bar")
+
+        with testing.expect_deprecated_20(
+            r"The Connection.connect\(\) function/method is considered "
+        ):
+            c2_branch = c2.connect()
+        eq_(c2_branch._execution_options, {"foo": "bar"})
diff --git a/test/engine/test_execute.py b/test/engine/test_execute.py
index ac9f034fe..5acd14177 100644
--- a/test/engine/test_execute.py
+++ b/test/engine/test_execute.py
@@ -1262,14 +1262,6 @@ class ExecutionOptionsTest(fixtures.TestBase):
         eq_(c1._execution_options, {"foo": "bar"})
         eq_(c2._execution_options, {"foo": "bar", "bat": "hoho"})
 
-    def test_branched_connection_execution_options(self):
-        engine = testing_engine("sqlite://")
-
-        conn = engine.connect()
-        c2 = conn.execution_options(foo="bar")
-        c2_branch = c2.connect()
-        eq_(c2_branch._execution_options, {"foo": "bar"})
-
     def test_get_engine_execution_options(self):
         engine = testing_engine("sqlite://")
         engine.dialect = Mock()
diff --git a/test/engine/test_reflection.py b/test/engine/test_reflection.py
index c137488ec..301614061 100644
--- a/test/engine/test_reflection.py
+++ b/test/engine/test_reflection.py
@@ -34,9 +34,6 @@ from sqlalchemy.testing.schema import Table
 from sqlalchemy.util import ue
 
 
-metadata, users = None, None
-
-
 class ReflectionTest(fixtures.TestBase, ComparesTables):
     __backend__ = True
 
@@ -906,15 +903,15 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             test_needs_fk=True,
         )
 
-        meta.create_all()
-        meta2 = MetaData(testing.db)
+        meta.create_all(testing.db)
+        meta2 = MetaData()
         a2 = Table(
             "addresses",
             meta2,
             Column("user_id", sa.Integer, sa.ForeignKey("users.id")),
-            autoload=True,
+            autoload_with=testing.db,
         )
-        u2 = Table("users", meta2, autoload=True)
+        u2 = Table("users", meta2, autoload_with=testing.db)
         s = sa.select([a2]).subquery()
 
         assert s.c.user_id is not None
@@ -926,19 +923,19 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
         assert list(a2.c.user_id.foreign_keys)[0].parent is a2.c.user_id
         assert u2.join(a2).onclause.compare(u2.c.id == a2.c.user_id)
 
-        meta2 = MetaData(testing.db)
+        meta2 = MetaData()
         u2 = Table(
             "users",
             meta2,
             Column("id", sa.Integer, primary_key=True),
-            autoload=True,
+            autoload_with=testing.db,
         )
         a2 = Table(
             "addresses",
             meta2,
             Column("id", sa.Integer, primary_key=True),
             Column("user_id", sa.Integer, sa.ForeignKey("users.id")),
-            autoload=True,
+            autoload_with=testing.db,
         )
         s = sa.select([a2]).subquery()
 
@@ -1012,29 +1009,28 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             for attr in test_attrs:
                 eq_(getattr(fk, attr), getattr(ref, attr))
 
+    @testing.provide_metadata
     def test_pks_not_uniques(self):
         """test that primary key reflection not tripped up by unique
         indexes"""
 
-        testing.db.execute(
-            """
-        CREATE TABLE book (
-            id INTEGER NOT NULL,
-            title VARCHAR(100) NOT NULL,
-            series INTEGER,
-            series_id INTEGER,
-            UNIQUE(series, series_id),
-            PRIMARY KEY(id)
-        )"""
-        )
-        try:
-            metadata = MetaData(bind=testing.db)
-            book = Table("book", metadata, autoload=True)
-            assert book.primary_key.contains_column(book.c.id)
-            assert not book.primary_key.contains_column(book.c.series)
-            assert len(book.primary_key) == 1
-        finally:
-            testing.db.execute("drop table book")
+        with testing.db.begin() as conn:
+            conn.execute(
+                """
+                CREATE TABLE book (
+                    id INTEGER NOT NULL,
+                    title VARCHAR(100) NOT NULL,
+                    series INTEGER,
+                    series_id INTEGER,
+                    UNIQUE(series, series_id),
+                    PRIMARY KEY(id)
+                )"""
+            )
+
+        book = Table("book", self.metadata, autoload_with=testing.db)
+        assert book.primary_key.contains_column(book.c.id)
+        assert not book.primary_key.contains_column(book.c.series)
+        eq_(len(book.primary_key), 1)
 
     def test_fk_error(self):
         metadata = MetaData(testing.db)
@@ -1054,30 +1050,28 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             metadata.create_all,
         )
 
+    @testing.provide_metadata
     def test_composite_pks(self):
         """test reflection of a composite primary key"""
 
-        testing.db.execute(
-            """
-        CREATE TABLE book (
-            id INTEGER NOT NULL,
-            isbn VARCHAR(50) NOT NULL,
-            title VARCHAR(100) NOT NULL,
-            series INTEGER NOT NULL,
-            series_id INTEGER NOT NULL,
-            UNIQUE(series, series_id),
-            PRIMARY KEY(id, isbn)
-        )"""
-        )
-        try:
-            metadata = MetaData(bind=testing.db)
-            book = Table("book", metadata, autoload=True)
-            assert book.primary_key.contains_column(book.c.id)
-            assert book.primary_key.contains_column(book.c.isbn)
-            assert not book.primary_key.contains_column(book.c.series)
-            assert len(book.primary_key) == 2
-        finally:
-            testing.db.execute("drop table book")
+        with testing.db.begin() as conn:
+            conn.execute(
+                """
+                CREATE TABLE book (
+                    id INTEGER NOT NULL,
+                    isbn VARCHAR(50) NOT NULL,
+                    title VARCHAR(100) NOT NULL,
+                    series INTEGER NOT NULL,
+                    series_id INTEGER NOT NULL,
+                    UNIQUE(series, series_id),
+                    PRIMARY KEY(id, isbn)
+                )"""
+            )
+        book = Table("book", self.metadata, autoload_with=testing.db)
+        assert book.primary_key.contains_column(book.c.id)
+        assert book.primary_key.contains_column(book.c.isbn)
+        assert not book.primary_key.contains_column(book.c.series)
+        eq_(len(book.primary_key), 2)
 
     @testing.exclude("mysql", "<", (4, 1, 1), "innodb funkiness")
     @testing.provide_metadata
@@ -1747,16 +1741,17 @@ class SchemaTest(fixtures.TestBase):
                 "dialect %s doesn't have a has_schema method"
                 % testing.db.dialect.name
             )
-        eq_(
-            testing.db.dialect.has_schema(
-                testing.db, testing.config.test_schema
-            ),
-            True,
-        )
-        eq_(
-            testing.db.dialect.has_schema(testing.db, "sa_fake_schema_123"),
-            False,
-        )
+        with testing.db.connect() as conn:
+            eq_(
+                testing.db.dialect.has_schema(
+                    conn, testing.config.test_schema
+                ),
+                True,
+            )
+            eq_(
+                testing.db.dialect.has_schema(conn, "sa_fake_schema_123"),
+                False,
+            )
 
     @testing.requires.schemas
     @testing.requires.cross_schema_fk_reflection
@@ -2018,24 +2013,29 @@ def createIndexes(con, schema=None):
 
 @testing.requires.views
 def _create_views(con, schema=None):
-    for table_name in ("users", "email_addresses"):
-        fullname = table_name
-        if schema:
-            fullname = "%s.%s" % (schema, table_name)
-        view_name = fullname + "_v"
-        query = "CREATE VIEW %s AS SELECT * FROM %s" % (view_name, fullname)
-        con.execute(sa.sql.text(query))
+    with testing.db.connect() as conn:
+        for table_name in ("users", "email_addresses"):
+            fullname = table_name
+            if schema:
+                fullname = "%s.%s" % (schema, table_name)
+            view_name = fullname + "_v"
+            query = "CREATE VIEW %s AS SELECT * FROM %s" % (
+                view_name,
+                fullname,
+            )
+            conn.execute(sa.sql.text(query))
 
 
 @testing.requires.views
 def _drop_views(con, schema=None):
-    for table_name in ("email_addresses", "users"):
-        fullname = table_name
-        if schema:
-            fullname = "%s.%s" % (schema, table_name)
-        view_name = fullname + "_v"
-        query = "DROP VIEW %s" % view_name
-        con.execute(sa.sql.text(query))
+    with testing.db.connect() as conn:
+        for table_name in ("email_addresses", "users"):
+            fullname = table_name
+            if schema:
+                fullname = "%s.%s" % (schema, table_name)
+            view_name = fullname + "_v"
+            query = "DROP VIEW %s" % view_name
+            conn.execute(sa.sql.text(query))
 
 
 class ReverseCasingReflectTest(fixtures.TestBase, AssertsCompiledSQL):