Add 'schema' parameter to table

Added a "schema" parameter to the :func:`_expression.table` construct,
allowing ad-hoc table expressions to also include a schema name.
Pull request courtesy Dylan Modesitt.

Fixes: #5309
Closes: #5310
Pull-request: https://github.com/sqlalchemy/sqlalchemy/pull/5310
Pull-request-sha: ce85681050500186678131f948b6ea277a65dc17
Change-Id: I32015d593e1ee1121c7426fbffdcc565d025fad1

4 files changed, 96 insertions, 4 deletions

diff --git a/doc/build/changelog/unreleased_13/5309.rst b/doc/build/changelog/unreleased_13/5309.rst
new file mode 100644
index 000000000..89ab14b06
--- /dev/null
+++ b/doc/build/changelog/unreleased_13/5309.rst
@@ -0,0 +1,7 @@
+.. change::
+    :tags: usecase, sql
+    :tickets: 5309
+
+    Added a ".schema" parameter to the :func:`_expression.table` construct,
+    allowing ad-hoc table expressions to also include a schema name.
+    Pull request courtesy Dylan Modesitt.
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index c8df637ba..27b9425ec 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -1878,9 +1878,9 @@ class FromGrouping(GroupedElement, FromClause):
 class TableClause(Immutable, FromClause):
     """Represents a minimal "table" construct.
 
-    This is a lightweight table object that has only a name and a
+    This is a lightweight table object that has only a name, a
     collection of columns, which are typically produced
-    by the :func:`_expression.column` function::
+    by the :func:`_expression.column` function, and a schema::
 
         from sqlalchemy import table, column
 
@@ -1925,7 +1925,7 @@ class TableClause(Immutable, FromClause):
     _autoincrement_column = None
     """No PK or default support so no autoincrement column."""
 
-    def __init__(self, name, *columns):
+    def __init__(self, name, *columns, **kw):
         """Produce a new :class:`_expression.TableClause`.
 
         The object returned is an instance of :class:`_expression.TableClause`
@@ -1938,10 +1938,15 @@ class TableClause(Immutable, FromClause):
            be imported from the plain ``sqlalchemy`` namespace like any
            other SQL element.
 
+
         :param name: Name of the table.
 
         :param columns: A collection of :func:`_expression.column` constructs.
 
+        :param schema: The schema name for this table.
+
+            .. versionadded:: 1.3.17 :func:`_expression.table` can now
+               accept a ``schema`` argument.
         """
 
         super(TableClause, self).__init__()
@@ -1952,6 +1957,12 @@ class TableClause(Immutable, FromClause):
         for c in columns:
             self.append_column(c)
 
+        schema = kw.pop("schema", None)
+        if schema is not None:
+            self.schema = schema
+        if kw:
+            raise exc.ArgumentError("Unsupported argument(s): %s" % list(kw))
+
     def _refresh_for_new_column(self, column):
         pass
 
diff --git a/test/orm/test_transaction.py b/test/orm/test_transaction.py
index 2b32282ba..4467542f6 100644
--- a/test/orm/test_transaction.py
+++ b/test/orm/test_transaction.py
@@ -595,7 +595,7 @@ class SessionTransactionTest(fixtures.RemovesEvents, FixtureTest):
         x = [1]
 
         @event.listens_for(sess, "after_commit")  # noqa
-        def add_another_user(session):
+        def add_another_user(session):  # noqa
             x[0] += 1
 
         sess.add(to_flush.pop())
diff --git a/test/sql/test_compiler.py b/test/sql/test_compiler.py
index 4b0b58b7e..b3ae7e12d 100644
--- a/test/sql/test_compiler.py
+++ b/test/sql/test_compiler.py
@@ -4448,6 +4448,80 @@ class SchemaTest(fixtures.TestBase, AssertsCompiledSQL):
             "(:rem_id, :datatype_id, :value)",
         )
 
+    def test_schema_lowercase_select(self):
+        # test that "schema" works correctly when passed to table
+        t1 = table("foo", column("a"), column("b"), schema="bar")
+        self.assert_compile(
+            select([t1]).select_from(t1),
+            "SELECT bar.foo.a, bar.foo.b FROM bar.foo",
+        )
+
+    def test_schema_lowercase_select_alias(self):
+        # test alias behavior
+        t1 = table("foo", schema="bar")
+        self.assert_compile(
+            select(["*"]).select_from(t1.alias("t")),
+            "SELECT * FROM bar.foo AS t",
+        )
+
+    def test_schema_lowercase_select_labels(self):
+        # test "schema" with extended_labels
+        t1 = table(
+            "baz",
+            column("id", Integer),
+            column("name", String),
+            column("meta", String),
+            schema="here",
+        )
+
+        self.assert_compile(
+            select([t1]).select_from(t1).apply_labels(),
+            "SELECT here.baz.id AS here_baz_id, here.baz.name AS "
+            "here_baz_name, here.baz.meta AS here_baz_meta FROM here.baz",
+        )
+
+    def test_schema_lowercase_select_subquery(self):
+        # test schema plays well with subqueries
+        t1 = table(
+            "yetagain",
+            column("anotherid", Integer),
+            column("anothername", String),
+            schema="here",
+        )
+        s = (
+            text("select id, name from user")
+            .columns(id=Integer, name=String)
+            .subquery()
+        )
+        stmt = select([t1.c.anotherid]).select_from(
+            t1.join(s, t1.c.anotherid == s.c.id)
+        )
+        compiled = stmt.compile()
+        eq_(
+            compiled._create_result_map(),
+            {
+                "anotherid": (
+                    "anotherid",
+                    (
+                        t1.c.anotherid,
+                        "anotherid",
+                        "anotherid",
+                        "here_yetagain_anotherid",
+                    ),
+                    t1.c.anotherid.type,
+                )
+            },
+        )
+
+    def test_schema_lowercase_invalid(self):
+        assert_raises_message(
+            exc.ArgumentError,
+            r"Unsupported argument\(s\): \['not_a_schema'\]",
+            table,
+            "foo",
+            not_a_schema="bar",
+        )
+
 
 class CorrelateTest(fixtures.TestBase, AssertsCompiledSQL):
     __dialect__ = "default"