diff options
| author | Mike Bayer <mike_mp@zzzcomputing.com> | 2021-01-21 11:15:06 -0500 |
|---|---|---|
| committer | Mike Bayer <mike_mp@zzzcomputing.com> | 2021-01-21 11:15:06 -0500 |
| commit | 851a3a362ee5e05b8438f92e2e1df63c68f79d68 (patch) | |
| tree | 2d862d02a1369d1730d78c933e09b709d2ef8bf6 /test | |
| parent | 05a31f2708590161d4b3b4c7ff65196c99b4a22b (diff) | |
| download | sqlalchemy-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.py | 31 | ||||
| -rw-r--r-- | test/dialect/oracle/test_dialect.py | 108 | ||||
| -rw-r--r-- | test/dialect/postgresql/test_query.py | 230 | ||||
| -rw-r--r-- | test/requirements.py | 7 | ||||
| -rw-r--r-- | test/sql/test_compare.py | 25 | ||||
| -rw-r--r-- | test/sql/test_functions.py | 432 | ||||
| -rw-r--r-- | test/sql/test_roles.py | 20 |
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, |
