summaryrefslogtreecommitdiff
path: root/test
diff options
context:
space:
mode:
authorMike Bayer <mike_mp@zzzcomputing.com>2021-01-21 11:15:06 -0500
committerMike Bayer <mike_mp@zzzcomputing.com>2021-01-21 11:15:06 -0500
commit851a3a362ee5e05b8438f92e2e1df63c68f79d68 (patch)
tree2d862d02a1369d1730d78c933e09b709d2ef8bf6 /test
parent05a31f2708590161d4b3b4c7ff65196c99b4a22b (diff)
downloadsqlalchemy-851a3a362ee5e05b8438f92e2e1df63c68f79d68.tar.gz
Revert "Implement support for functions as FROM with columns clause support"
This reverts commit 05a31f2708590161d4b3b4c7ff65196c99b4a22b. Atom has this little button called "push" and just pushes to master, I wasn't even *on* master. oops
Diffstat (limited to 'test')
-rw-r--r--test/dialect/oracle/test_compiler.py31
-rw-r--r--test/dialect/oracle/test_dialect.py108
-rw-r--r--test/dialect/postgresql/test_query.py230
-rw-r--r--test/requirements.py7
-rw-r--r--test/sql/test_compare.py25
-rw-r--r--test/sql/test_functions.py432
-rw-r--r--test/sql/test_roles.py20
7 files changed, 0 insertions, 853 deletions
diff --git a/test/dialect/oracle/test_compiler.py b/test/dialect/oracle/test_compiler.py
index f12207d2b..f09346eb3 100644
--- a/test/dialect/oracle/test_compiler.py
+++ b/test/dialect/oracle/test_compiler.py
@@ -1481,34 +1481,3 @@ class RegexpTest(fixtures.TestBase, testing.AssertsCompiledSQL):
"REGEXP_REPLACE(mytable.myid, :myid_1, :myid_2, mytable.name)",
checkparams={"myid_1": "pattern", "myid_2": "replacement"},
)
-
-
-class TableValuedFunctionTest(fixtures.TestBase, testing.AssertsCompiledSQL):
- __dialect__ = "oracle"
-
- def test_scalar_alias_column(self):
- fn = func.scalar_strings(5)
- stmt = select(fn.alias().column)
- self.assert_compile(
- stmt,
- "SELECT COLUMN_VALUE anon_1 "
- "FROM TABLE (scalar_strings(:scalar_strings_1)) anon_1",
- )
-
- def test_column_valued(self):
- fn = func.scalar_strings(5)
- stmt = select(fn.column_valued())
- self.assert_compile(
- stmt,
- "SELECT COLUMN_VALUE anon_1 "
- "FROM TABLE (scalar_strings(:scalar_strings_1)) anon_1",
- )
-
- def test_table_valued(self):
- fn = func.three_pairs().table_valued("string1", "string2")
- stmt = select(fn.c.string1, fn.c.string2)
- self.assert_compile(
- stmt,
- "SELECT anon_1.string1, anon_1.string2 "
- "FROM TABLE (three_pairs()) anon_1",
- )
diff --git a/test/dialect/oracle/test_dialect.py b/test/dialect/oracle/test_dialect.py
index cce33d4a2..2ec306842 100644
--- a/test/dialect/oracle/test_dialect.py
+++ b/test/dialect/oracle/test_dialect.py
@@ -7,7 +7,6 @@ from sqlalchemy import Computed
from sqlalchemy import create_engine
from sqlalchemy import exc
from sqlalchemy import Float
-from sqlalchemy import func
from sqlalchemy import Integer
from sqlalchemy import literal_column
from sqlalchemy import outparam
@@ -947,110 +946,3 @@ class CXOracleConnectArgsTest(fixtures.TestBase):
"auto_convert_lobs",
False,
)
-
-
-class TableValuedTest(fixtures.TestBase):
- __backend__ = True
- __only_on__ = "oracle"
-
- @testing.fixture
- def scalar_strings(self, connection):
- connection.exec_driver_sql(
- "CREATE OR REPLACE TYPE strings_t IS TABLE OF VARCHAR2 (100)"
- )
- connection.exec_driver_sql(
- r"""
-CREATE OR REPLACE FUNCTION scalar_strings (
- count_in IN INTEGER)
- RETURN strings_t
- AUTHID DEFINER
-IS
- l_strings strings_t := strings_t ();
-BEGIN
- l_strings.EXTEND (count_in);
-
- FOR indx IN 1 .. count_in
- LOOP
- l_strings (indx) := 'some string';
- END LOOP;
-
- RETURN l_strings;
-END;
- """
- )
- yield
- connection.exec_driver_sql("DROP FUNCTION scalar_strings")
- connection.exec_driver_sql("DROP TYPE strings_t")
-
- @testing.fixture
- def two_strings(self, connection):
- connection.exec_driver_sql(
- """
-CREATE OR REPLACE TYPE two_strings_ot
- AUTHID DEFINER IS OBJECT
-(
- string1 VARCHAR2 (10),
- string2 VARCHAR2 (10)
-)"""
- )
- connection.exec_driver_sql(
- """
- CREATE OR REPLACE TYPE two_strings_nt IS TABLE OF two_strings_ot
-"""
- )
-
- connection.exec_driver_sql(
- """
- CREATE OR REPLACE FUNCTION three_pairs
- RETURN two_strings_nt
- AUTHID DEFINER
-IS
- l_strings two_strings_nt;
-BEGIN
- RETURN two_strings_nt (two_strings_ot ('a', 'b'),
- two_strings_ot ('c', 'd'),
- two_strings_ot ('e', 'f'));
-END;
-"""
- )
- yield
- connection.exec_driver_sql("DROP FUNCTION three_pairs")
- connection.exec_driver_sql("DROP TYPE two_strings_nt")
- connection.exec_driver_sql("DROP TYPE two_strings_ot")
-
- def test_scalar_strings_control(self, scalar_strings, connection):
- result = (
- connection.exec_driver_sql(
- "SELECT COLUMN_VALUE my_string FROM TABLE (scalar_strings (5))"
- )
- .scalars()
- .all()
- )
- eq_(result, ["some string"] * 5)
-
- def test_scalar_strings_named_control(self, scalar_strings, connection):
- result = (
- connection.exec_driver_sql(
- "SELECT COLUMN_VALUE anon_1 "
- "FROM TABLE (scalar_strings (5)) anon_1"
- )
- .scalars()
- .all()
- )
- eq_(result, ["some string"] * 5)
-
- def test_scalar_strings(self, scalar_strings, connection):
- fn = func.scalar_strings(5)
- result = connection.execute(select(fn.column_valued())).scalars().all()
- eq_(result, ["some string"] * 5)
-
- def test_two_strings_control(self, two_strings, connection):
- result = connection.exec_driver_sql(
- "SELECT string1, string2 FROM TABLE (three_pairs ())"
- ).all()
- eq_(result, [("a", "b"), ("c", "d"), ("e", "f")])
-
- def test_two_strings(self, two_strings, connection):
- fn = func.three_pairs().table_valued("string1", "string2")
- result = connection.execute(select(fn.c.string1, fn.c.string2)).all()
- eq_(result, [("a", "b"), ("c", "d"), ("e", "f")])
diff --git a/test/dialect/postgresql/test_query.py b/test/dialect/postgresql/test_query.py
index a2c152501..47f86e791 100644
--- a/test/dialect/postgresql/test_query.py
+++ b/test/dialect/postgresql/test_query.py
@@ -3,13 +3,11 @@
import datetime
from sqlalchemy import and_
-from sqlalchemy import cast
from sqlalchemy import Column
from sqlalchemy import Date
from sqlalchemy import DateTime
from sqlalchemy import exc
from sqlalchemy import extract
-from sqlalchemy import Float
from sqlalchemy import ForeignKey
from sqlalchemy import func
from sqlalchemy import Integer
@@ -24,10 +22,8 @@ from sqlalchemy import Table
from sqlalchemy import testing
from sqlalchemy import text
from sqlalchemy import Time
-from sqlalchemy import true
from sqlalchemy import tuple_
from sqlalchemy.dialects import postgresql
-from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import AssertsExecutionResults
@@ -1101,229 +1097,3 @@ class ExtractTest(fixtures.TablesTest):
datetime.timedelta(days=5) + t.c.dtme,
overrides={"day": 15, "epoch": 1337084125.0},
)
-
-
-class TableValuedRoundTripTest(fixtures.TestBase):
- __backend__ = True
- __only_on__ = "postgresql"
-
- def test_generate_series_scalar(self, connection):
- x = func.generate_series(1, 2).alias("x")
- y = func.generate_series(1, 2).alias("y")
-
- stmt = select(x.column, y.column).join_from(x, y, true())
-
- eq_(connection.execute(stmt).all(), [(1, 1), (1, 2), (2, 1), (2, 2)])
-
- def test_aggregate_scalar_over_table_valued(self, metadata, connection):
- test = Table(
- "test", metadata, Column("id", Integer), Column("data", JSONB)
- )
- test.create(connection)
-
- connection.execute(
- test.insert(),
- [
- {"id": 1, "data": {"key": [23.7, 108.17, 55.98]}},
- {"id": 2, "data": {"key": [2.320, 9.55]}},
- {"id": 3, "data": {"key": [10.5, 6]}},
- ],
- )
-
- elem = (
- func.jsonb_array_elements_text(test.c.data["key"])
- .table_valued("value")
- .alias("elem")
- )
-
- maxdepth = select(func.max(cast(elem.c.value, Float))).label(
- "maxdepth"
- )
-
- stmt = select(test.c.id.label("test_id"), maxdepth).order_by(
- "maxdepth"
- )
-
- eq_(
- connection.execute(stmt).all(), [(2, 9.55), (3, 10.5), (1, 108.17)]
- )
-
- @testing.fixture
- def assets_transactions(self, metadata, connection):
- assets_transactions = Table(
- "assets_transactions",
- metadata,
- Column("id", Integer),
- Column("contents", JSONB),
- )
- assets_transactions.create(connection)
- connection.execute(
- assets_transactions.insert(),
- [
- {"id": 1, "contents": {"k1": "v1"}},
- {"id": 2, "contents": {"k2": "v2"}},
- {"id": 3, "contents": {"k3": "v3"}},
- ],
- )
- return assets_transactions
-
- def test_scalar_table_valued(self, assets_transactions, connection):
- stmt = select(
- assets_transactions.c.id,
- func.jsonb_each(
- assets_transactions.c.contents, type_=JSONB
- ).scalar_table_valued("key"),
- func.jsonb_each(
- assets_transactions.c.contents, type_=JSONB
- ).scalar_table_valued("value"),
- )
-
- eq_(
- connection.execute(stmt).all(),
- [(1, "k1", "v1"), (2, "k2", "v2"), (3, "k3", "v3")],
- )
-
- def test_table_valued(self, assets_transactions, connection):
-
- jb = func.jsonb_each(assets_transactions.c.contents).table_valued(
- "key", "value"
- )
-
- stmt = select(assets_transactions.c.id, jb.c.key, jb.c.value).join(
- jb, true()
- )
- eq_(
- connection.execute(stmt).all(),
- [(1, "k1", "v1"), (2, "k2", "v2"), (3, "k3", "v3")],
- )
-
- @testing.fixture
- def axy_table(self, metadata, connection):
- a = Table(
- "a",
- metadata,
- Column("id", Integer),
- Column("x", Integer),
- Column("y", Integer),
- )
- a.create(connection)
- connection.execute(
- a.insert(),
- [
- {"id": 1, "x": 5, "y": 4},
- {"id": 2, "x": 15, "y": 3},
- {"id": 3, "x": 7, "y": 9},
- ],
- )
-
- return a
-
- def test_function_against_table_record(self, axy_table, connection):
- """
- SELECT row_to_json(anon_1) AS row_to_json_1
- FROM (SELECT a.id AS id, a.x AS x, a.y AS y
- FROM a) AS anon_1
-
- """
-
- stmt = select(func.row_to_json(axy_table.record()))
-
- eq_(
- connection.execute(stmt).scalars().all(),
- [
- {"id": 1, "x": 5, "y": 4},
- {"id": 2, "x": 15, "y": 3},
- {"id": 3, "x": 7, "y": 9},
- ],
- )
-
- def test_function_against_subq_record(self, axy_table, connection):
- """
- SELECT row_to_json(anon_1) AS row_to_json_1
- FROM (SELECT a.id AS id, a.x AS x, a.y AS y
- FROM a) AS anon_1
-
- """
-
- stmt = select(func.row_to_json(axy_table.select().subquery().record()))
-
- eq_(
- connection.execute(stmt).scalars().all(),
- [
- {"id": 1, "x": 5, "y": 4},
- {"id": 2, "x": 15, "y": 3},
- {"id": 3, "x": 7, "y": 9},
- ],
- )
-
- def test_function_against_row_constructor(self, connection):
-
- stmt = select(func.row_to_json(func.row(1, "foo")))
-
- eq_(connection.scalar(stmt), {"f1": 1, "f2": "foo"})
-
- def test_with_ordinality_named(self, connection):
-
- stmt = select(
- func.generate_series(4, 1, -1).named_table_valued(
- "gs", with_ordinality="ordinality"
- )
- )
-
- eq_(connection.execute(stmt).all(), [(4, 1), (3, 2), (2, 3), (1, 4)])
-
- def test_with_ordinality_star(self, connection):
-
- stmt = select("*").select_from(
- func.generate_series(4, 1, -1).table_valued(
- with_ordinality="ordinality"
- )
- )
-
- eq_(connection.execute(stmt).all(), [(4, 1), (3, 2), (2, 3), (1, 4)])
-
- def test_plain_old_unnest(self, connection):
- fn = func.unnest(
- postgresql.array(["one", "two", "three", "four"])
- ).column_valued()
-
- stmt = select(fn)
-
- eq_(
- connection.execute(stmt).all(),
- [("one",), ("two",), ("three",), ("four",)],
- )
-
- def test_unnest_with_ordinality(self, connection):
-
- array_val = postgresql.array(
- [postgresql.array([14, 41, 7]), postgresql.array([54, 9, 49])]
- )
- stmt = select("*").select_from(
- func.unnest(array_val)
- .named_table_valued("elts", with_ordinality="num")
- .alias("t")
- )
- eq_(
- connection.execute(stmt).all(),
- [(14, 1), (41, 2), (7, 3), (54, 4), (9, 5), (49, 6)],
- )
-
- def test_unnest_with_ordinality_named(self, connection):
-
- array_val = postgresql.array(
- [postgresql.array([14, 41, 7]), postgresql.array([54, 9, 49])]
- )
-
- fn = (
- func.unnest(array_val)
- .named_table_valued("elts", with_ordinality="num")
- .alias("t")
- )
-
- stmt = select(fn.c.elts, fn.c.num)
-
- eq_(
- connection.execute(stmt).all(),
- [(14, 1), (41, 2), (7, 3), (54, 4), (9, 5), (49, 6)],
- )
diff --git a/test/requirements.py b/test/requirements.py
index 0098283b3..3c9b39ac7 100644
--- a/test/requirements.py
+++ b/test/requirements.py
@@ -957,13 +957,6 @@ class DefaultRequirements(SuiteRequirements):
return skip_if(["mssql", "sqlite"])
@property
- def tuple_valued_builtin_functions(self):
- return only_on(
- lambda config: self._sqlite_json(config)
- or against(config, "postgresql")
- )
-
- @property
def array_type(self):
return only_on(
[
diff --git a/test/sql/test_compare.py b/test/sql/test_compare.py
index 716ff107a..1ac3613f7 100644
--- a/test/sql/test_compare.py
+++ b/test/sql/test_compare.py
@@ -272,31 +272,6 @@ class CoreFixtures(object):
func.next_value(Sequence("q")),
func.next_value(Sequence("p")),
),
- lambda: (
- func.json_to_recordset("{foo}"),
- func.json_to_recordset("{foo}").table_valued("a", "b"),
- func.jsonb_to_recordset("{foo}").table_valued("a", "b"),
- func.json_to_recordset("{foo}").named_table_valued("a", "b"),
- func.json_to_recordset("{foo}").named_table_valued(
- "a", with_ordinality="b"
- ),
- func.json_to_recordset("{foo}").named_table_valued(
- "a", with_ordinality="c"
- ),
- func.json_to_recordset("{foo}").named_table_valued(
- column("a", Integer), column("b", String)
- ),
- func.json_to_recordset("{foo}").named_table_valued("b", "c"),
- func.json_to_recordset("{foo}")
- .named_table_valued("a", "b")
- .alias("foo"),
- func.json_to_recordset("{foo}")
- .table_valued("a", "b")
- .alias("foo"),
- func.json_to_recordset("{foo}").column_valued(),
- func.json_to_recordset("{foo}").scalar_table_valued("foo"),
- ),
- lambda: (table_a.record(), table_b.record()),
lambda: (True_(), False_()),
lambda: (Null(),),
lambda: (ReturnTypeFromArgs("foo"), ReturnTypeFromArgs(5)),
diff --git a/test/sql/test_functions.py b/test/sql/test_functions.py
index 59accd245..50f50f0f0 100644
--- a/test/sql/test_functions.py
+++ b/test/sql/test_functions.py
@@ -5,15 +5,12 @@ import decimal
from sqlalchemy import ARRAY
from sqlalchemy import bindparam
from sqlalchemy import Boolean
-from sqlalchemy import cast
from sqlalchemy import Column
from sqlalchemy import Date
from sqlalchemy import DateTime
from sqlalchemy import extract
-from sqlalchemy import Float
from sqlalchemy import func
from sqlalchemy import Integer
-from sqlalchemy import JSON
from sqlalchemy import literal
from sqlalchemy import literal_column
from sqlalchemy import Numeric
@@ -23,7 +20,6 @@ from sqlalchemy import sql
from sqlalchemy import String
from sqlalchemy import Table
from sqlalchemy import testing
-from sqlalchemy import true
from sqlalchemy import types as sqltypes
from sqlalchemy import util
from sqlalchemy.dialects import mysql
@@ -1152,431 +1148,3 @@ class RegisterTest(fixtures.TestBase, AssertsCompiledSQL):
assert "not_registered_func" not in functions._registry["_default"]
assert isinstance(func.not_registered_func_child().type, Integer)
-
-
-class TableValuedCompileTest(fixtures.TestBase, AssertsCompiledSQL):
- """test the full set of functions as FROM developed in [ticket:3566]"""
-
- __dialect__ = "default_enhanced"
-
- def test_aggregate_scalar_over_table_valued(self):
- test = table("test", column("id"), column("data", JSON))
-
- elem = (
- func.json_array_elements_text(test.c.data["key"])
- .table_valued("value")
- .alias("elem")
- )
-
- maxdepth = select(func.max(cast(elem.c.value, Float))).label(
- "maxdepth"
- )
-
- stmt = select(test.c.id.label("test_id"), maxdepth).order_by(
- "maxdepth"
- )
-
- self.assert_compile(
- stmt,
- "SELECT test.id AS test_id, "
- "(SELECT max(CAST(elem.value AS FLOAT)) AS max_1 "
- "FROM json_array_elements_text(test.data[:data_1]) AS elem) "
- "AS maxdepth "
- "FROM test ORDER BY maxdepth",
- )
-
- def test_scalar_table_valued(self):
- assets_transactions = table(
- "assets_transactions", column("id"), column("contents", JSON)
- )
-
- stmt = select(
- assets_transactions.c.id,
- func.jsonb_each(
- assets_transactions.c.contents
- ).scalar_table_valued("key"),
- func.jsonb_each(
- assets_transactions.c.contents
- ).scalar_table_valued("value"),
- )
- self.assert_compile(
- stmt,
- "SELECT assets_transactions.id, "
- "(jsonb_each(assets_transactions.contents)).key, "
- "(jsonb_each(assets_transactions.contents)).value "
- "FROM assets_transactions",
- )
-
- def test_table_valued_one(self):
- assets_transactions = table(
- "assets_transactions", column("id"), column("contents", JSON)
- )
-
- jb = func.jsonb_each(assets_transactions.c.contents).table_valued(
- "key", "value"
- )
-
- stmt = select(assets_transactions.c.id, jb.c.key, jb.c.value).join(
- jb, true()
- )
-
- self.assert_compile(
- stmt,
- "SELECT assets_transactions.id, anon_1.key, anon_1.value "
- "FROM assets_transactions "
- "JOIN jsonb_each(assets_transactions.contents) AS anon_1 ON true",
- )
-
- def test_table_valued_two(self):
- """
- SELECT vi.id, vv.value
- FROM value_ids() AS vi JOIN values AS vv ON vv.id = vi.id
-
- """
-
- values = table(
- "values",
- column(
- "id",
- Integer,
- ),
- column("value", String),
- )
- vi = func.value_ids().table_valued(column("id", Integer)).alias("vi")
- vv = values.alias("vv")
-
- stmt = select(vi.c.id, vv.c.value).select_from( # noqa
- vi.join(vv, vv.c.id == vi.c.id)
- )
- self.assert_compile(
- stmt,
- "SELECT vi.id, vv.value FROM value_ids() AS vi "
- "JOIN values AS vv ON vv.id = vi.id",
- )
-
- def test_table_as_record(self):
- a = table(
- "a",
- column("id"),
- column("x"),
- column("y"),
- )
-
- stmt = select(func.row_to_json(a.record()))
-
- self.assert_compile(
- stmt, "SELECT row_to_json(a) AS row_to_json_1 FROM a"
- )
-
- def test_subquery_as_record(self):
- """
- SELECT row_to_json(anon_1) AS row_to_json_1
- FROM (SELECT a.id AS id, a.x AS x, a.y AS y
- FROM a) AS anon_1
-
- """
-
- a = table(
- "a",
- column("id"),
- column("x"),
- column("y"),
- )
-
- stmt = select(func.row_to_json(a.select().subquery().record()))
-
- self.assert_compile(
- stmt,
- "SELECT row_to_json(anon_1) AS row_to_json_1 FROM "
- "(SELECT a.id AS id, a.x AS x, a.y AS y FROM a) AS anon_1",
- )
-
- def test_scalar_subquery(self):
-
- a = table(
- "a",
- column("id"),
- column("x"),
- column("y"),
- )
-
- stmt = select(func.row_to_json(a.select().scalar_subquery()))
-
- self.assert_compile(
- stmt,
- "SELECT row_to_json((SELECT a.id, a.x, a.y FROM a)) "
- "AS row_to_json_1",
- )
-
- def test_named_with_ordinality(self):
- """
- SELECT a.id AS a_id, a.refs AS a_refs,
- unnested.unnested AS unnested_unnested,
- unnested.ordinality AS unnested_ordinality,
- b.id AS b_id, b.ref AS b_ref
- FROM a LEFT OUTER JOIN unnest(a.refs)
- `WITH ORDINALITY AS unnested(unnested, ordinality) ON true
- LEFT OUTER JOIN b ON unnested.unnested = b.ref
-
- """ # noqa 501
-
- a = table("a", column("id"), column("refs"))
- b = table("b", column("id"), column("ref"))
-
- unnested = (
- func.unnest(a.c.refs)
- .named_table_valued("unnested", with_ordinality="ordinality")
- .alias("unnested")
- )
-
- stmt = (
- select(
- a.c.id, a.c.refs, unnested.c.unnested, unnested.c.ordinality
- )
- .outerjoin(unnested, true())
- .outerjoin(
- b,
- unnested.c.unnested == b.c.ref,
- )
- )
- self.assert_compile(
- stmt,
- "SELECT a.id, a.refs, unnested.unnested, unnested.ordinality "
- "FROM a "
- "LEFT OUTER JOIN unnest(a.refs) "
- "WITH ORDINALITY AS unnested(unnested, ordinality) ON true "
- "LEFT OUTER JOIN b ON unnested.unnested = b.ref",
- )
-
- def test_star_with_ordinality(self):
- """
- SELECT * FROM generate_series(4,1,-1) WITH ORDINALITY;
- """
-
- stmt = select("*").select_from( # noqa
- func.generate_series(4, 1, -1).table_valued(
- with_ordinality="ordinality"
- )
- )
- self.assert_compile(
- stmt,
- "SELECT * FROM generate_series"
- "(:generate_series_1, :generate_series_2, :generate_series_3) "
- "WITH ORDINALITY AS anon_1",
- )
-
- def test_json_object_keys_with_ordinality(self):
- """
- SELECT * FROM json_object_keys('{"a1":"1","a2":"2","a3":"3"}')
- WITH ORDINALITY AS t(keys, n);
- """
- stmt = select("*").select_from(
- func.json_object_keys(
- literal({"a1": "1", "a2": "2", "a3": "3"}, type_=JSON)
- )
- .named_table_valued("keys", with_ordinality="n")
- .alias("t")
- )
-
- self.assert_compile(
- stmt,
- "SELECT * FROM json_object_keys(:param_1) "
- "WITH ORDINALITY AS t(keys, n)",
- )
-
- def test_alias_column(self):
- """
-
- ::
-
- SELECT x, y
- FROM
- generate_series(:generate_series_1, :generate_series_2) AS x,
- generate_series(:generate_series_3, :generate_series_4) AS y
-
- """
-
- x = func.generate_series(1, 2).alias("x")
- y = func.generate_series(3, 4).alias("y")
- stmt = select(x.column, y.column)
-
- self.assert_compile(
- stmt,
- "SELECT x, y FROM "
- "generate_series(:generate_series_1, :generate_series_2) AS x, "
- "generate_series(:generate_series_3, :generate_series_4) AS y",
- )
-
- def test_column_valued_one(self):
- fn = func.unnest(["one", "two", "three", "four"]).column_valued()
-
- stmt = select(fn)
-
- self.assert_compile(
- stmt, "SELECT anon_1 FROM unnest(:unnest_1) AS anon_1"
- )
-
- def test_column_valued_two(self):
- """
-
- ::
-
- SELECT x, y
- FROM
- generate_series(:generate_series_1, :generate_series_2) AS x,
- generate_series(:generate_series_3, :generate_series_4) AS y
-
- """
-
- x = func.generate_series(1, 2).column_valued("x")
- y = func.generate_series(3, 4).column_valued("y")
- stmt = select(x, y)
-
- self.assert_compile(
- stmt,
- "SELECT x, y FROM "
- "generate_series(:generate_series_1, :generate_series_2) AS x, "
- "generate_series(:generate_series_3, :generate_series_4) AS y",
- )
-
- def test_column_valued_subquery(self):
- x = func.generate_series(1, 2).column_valued("x")
- y = func.generate_series(3, 4).column_valued("y")
- subq = select(x, y).subquery()
- stmt = select(subq).where(subq.c.x > 2)
-
- self.assert_compile(
- stmt,
- "SELECT anon_1.x, anon_1.y FROM "
- "(SELECT x, y FROM "
- "generate_series(:generate_series_1, :generate_series_2) AS x, "
- "generate_series(:generate_series_3, :generate_series_4) AS y"
- ") AS anon_1 "
- "WHERE anon_1.x > :x_1",
- )
-
- def test_ten(self):
- """
- # this is the "record" type
-
- SELECT
- table1.user_id AS table1_user_id,
- table2.name AS table2_name,
- jsonb_table.name AS jsonb_table_name,
- count(jsonb_table.time) AS count_1
- FROM table1
- JOIN table2 ON table1.user_id = table2.id
- JOIN LATERAL jsonb_to_recordset(table1.jsonb) AS jsonb_table(name TEXT, time FLOAT) ON true
- WHERE
- table2.route_id = %(route_id_1)s
- AND table1.list_id IN (%(list_id_1)s, %(list_id_2)s, %(list_id_3)s)
- AND jsonb_table.name IN (%(name_1)s, %(name_2)s, %(name_3)s)
- GROUP BY table1.user_id, table2.name, jsonb_table.name
- ORDER BY table2.name
-
- """ # noqa
-
- def test_function_alias(self):
- """
- ::
-
- SELECT result_elem -> 'Field' as field
- FROM "check" AS check_, json_array_elements(
- (
- SELECT check_inside.response -> 'Results'
- FROM "check" as check_inside
- WHERE check_inside.id = check_.id
- )
- ) AS result_elem
- WHERE result_elem ->> 'Name' = 'FooBar'
-
- """
- check = table("check", column("id"), column("response", JSON))
-
- check_inside = check.alias("check_inside")
- check_outside = check.alias("_check")
-
- subq = (
- select(check_inside.c.response["Results"])
- .where(check_inside.c.id == check_outside.c.id)
- .scalar_subquery()
- )
-
- fn = func.json_array_elements(subq, type_=JSON).alias("result_elem")
-
- stmt = (
- select(fn.column["Field"].label("field"))
- .where(fn.column["Name"] == "FooBar")
- .select_from(check_outside)
- )
-
- self.assert_compile(
- stmt,
- "SELECT result_elem[:result_elem_1] AS field "
- "FROM json_array_elements("
- "(SELECT check_inside.response[:response_1] AS anon_1 "
- 'FROM "check" AS check_inside '
- "WHERE check_inside.id = _check.id)"
- ') AS result_elem, "check" AS _check '
- "WHERE result_elem[:result_elem_2] = :param_1",
- )
-
- def test_named_table_valued(self):
-
- fn = func.json_to_recordset( # noqa
- '[{"a":1,"b":"foo"},{"a":"2","c":"bar"}]'
- ).named_table_valued(column("a", Integer), column("b", String))
-
- stmt = select(fn.c.a, fn.c.b)
-
- self.assert_compile(
- stmt,
- "SELECT anon_1.a, anon_1.b "
- "FROM json_to_recordset(:json_to_recordset_1) "
- "AS anon_1(a INTEGER, b VARCHAR)",
- )
-
- def test_named_table_valued_subquery(self):
-
- fn = func.json_to_recordset( # noqa
- '[{"a":1,"b":"foo"},{"a":"2","c":"bar"}]'
- ).named_table_valued(column("a", Integer), column("b", String))
-
- stmt = select(fn.c.a, fn.c.b).subquery()
-
- stmt = select(stmt)
-
- self.assert_compile(
- stmt,
- "SELECT anon_1.a, anon_1.b FROM "
- "(SELECT anon_2.a AS a, anon_2.b AS b "
- "FROM json_to_recordset(:json_to_recordset_1) "
- "AS anon_2(a INTEGER, b VARCHAR)"
- ") AS anon_1",
- )
-
- def test_named_table_valued_alias(self):
-
- """select * from json_to_recordset
- ('[{"a":1,"b":"foo"},{"a":"2","c":"bar"}]') as x(a int, b text);"""
-
- fn = (
- func.json_to_recordset( # noqa
- '[{"a":1,"b":"foo"},{"a":"2","c":"bar"}]'
- )
- .named_table_valued(column("a", Integer), column("b", String))
- .alias("jbr")
- )
-
- stmt = select(fn.c.a, fn.c.b)
-
- self.assert_compile(
- stmt,
- "SELECT jbr.a, jbr.b "
- "FROM json_to_recordset(:json_to_recordset_1) "
- "AS jbr(a INTEGER, b VARCHAR)",
- )
-
- # continuing from
- # https://paquier.xyz/postgresql-2/postgres-9-4-feature-highlight-with-ordinality/
- # https://github.com/sqlalchemy/sqlalchemy/issues/3566
diff --git a/test/sql/test_roles.py b/test/sql/test_roles.py
index 3e6f0ecfc..0ef90e89e 100644
--- a/test/sql/test_roles.py
+++ b/test/sql/test_roles.py
@@ -162,26 +162,6 @@ class RoleTest(fixtures.TestBase):
select(column("q")).alias(),
)
- def test_record_advice(self):
- msg = (
- r"SQL expression element expected, got %s. To create a "
- r"column expression from a FROM clause row as a whole, "
- r"use the .record\(\) method."
- )
- assert_raises_message(
- exc.ArgumentError,
- msg % ("Table.*",),
- expect,
- roles.ExpressionElementRole,
- t,
- )
-
- # no record() message here right now, it goes to scalar subquery
- with testing.expect_warnings(
- "implicitly coercing SELECT object to scalar subquery"
- ):
- expect(roles.ExpressionElementRole, t.select().alias())
-
def test_statement_no_text_coercion(self):
assert_raises_message(
exc.ArgumentError,