MCPcopy
hub / github.com/sqlalchemy/sqlalchemy / test_nested_api

Method test_nested_api

test/sql/test_compiler.py:8000–8084  ·  view source on GitHub ↗
(self)

Source from the content-addressed store, hash-verified

7998 )
7999
8000 def test_nested_api(self):
8001 from sqlalchemy.engine.cursor import CursorResultMetaData
8002
8003 stmt2 = select(table2).subquery()
8004
8005 stmt1 = select(table1).select_from(stmt2)
8006
8007 contexts = {}
8008
8009 int_ = Integer()
8010
8011 class MyCompiler(compiler.SQLCompiler):
8012 def visit_select(self, stmt, *arg, **kw):
8013 if stmt is stmt2.element:
8014 with self._nested_result() as nested:
8015 contexts[stmt2.element] = nested
8016 text = super().visit_select(
8017 stmt2.element,
8018 )
8019 self._add_to_result_map("k1", "k1", (1, 2, 3), int_)
8020 else:
8021 text = super().visit_select(stmt, *arg, **kw)
8022 self._add_to_result_map("k2", "k2", (3, 4, 5), int_)
8023 return text
8024
8025 comp = MyCompiler(default.DefaultDialect(), stmt1)
8026 eq_(
8027 CursorResultMetaData._create_description_match_map(
8028 contexts[stmt2.element][0]
8029 ),
8030 {
8031 "otherid": (
8032 "otherid",
8033 (
8034 table2.c.otherid,
8035 "otherid",
8036 "otherid",
8037 "myothertable_otherid",
8038 ),
8039 table2.c.otherid.type,
8040 0,
8041 ),
8042 "othername": (
8043 "othername",
8044 (
8045 table2.c.othername,
8046 "othername",
8047 "othername",
8048 "myothertable_othername",
8049 ),
8050 table2.c.othername.type,
8051 1,
8052 ),
8053 "k1": ("k1", (1, 2, 3), int_, 2),
8054 },
8055 )
8056 eq_(
8057 comp._create_result_map(),

Callers

nothing calls this directly

Calls 8

selectFunction · 0.90
IntegerClass · 0.90
eq_Function · 0.90
_create_result_mapMethod · 0.80
MyCompilerClass · 0.70
subqueryMethod · 0.45
select_fromMethod · 0.45

Tested by

no test coverage detected