MCPcopy
hub / github.com/sqlalchemy/sqlalchemy / test_union

Method test_union

test/sql/test_statement_params.py:133–183  ·  view source on GitHub ↗
(self, connection)

Source from the content-addressed store, hash-verified

131 eq_(cs(stmt3.params({"x": 9})).all(), [9])
132
133 def test_union(self, connection):
134 a = self.tables.a
135
136 cs = connection.scalars
137 for _ in range(3):
138 x1 = random.randint(1, 10)
139 x2 = random.randint(1, 10)
140
141 eq_(
142 cs(
143 select(a)
144 .where(a.c.data == x1)
145 .union_all(select(a).where(a.c.data == x2))
146 .order_by(a.c.data)
147 ).all(),
148 sorted([x1, x2]),
149 )
150
151 x1 = random.randint(1, 10)
152 x2 = random.randint(1, 10)
153 stmt = (
154 select(a, literal(1).label("ord"))
155 .where(a.c.data == bindparam("x", x1))
156 .union_all(
157 select(a, literal(2)).where(a.c.data == bindparam("y", x2))
158 )
159 .order_by("ord")
160 )
161 eq_(cs(stmt).all(), [x1, x2])
162
163 x1a = random.randint(1, 10)
164 eq_(cs(stmt.params({"x": x1a})).all(), [x1a, x2])
165
166 x2 = random.randint(1, 10)
167 eq_(cs(stmt, {"y": x2}).all(), [x1, x2])
168
169 x1 = random.randint(1, 10)
170 x2 = random.randint(1, 10)
171 eq_(cs(stmt.params({"x": x1}), {"y": x2}).all(), [x1, x2])
172
173 x1 = random.randint(1, 10)
174 x2 = random.randint(1, 10)
175 stmt2 = (
176 stmt.params(x=x1)
177 .subquery()
178 .select()
179 .params(y=x2)
180 .order_by("ord")
181 )
182 eq_(cs(stmt2).all(), [x1, x2])
183 eq_(cs(stmt2.params({"x": x1}).params({"y": x2})).all(), [x1, x2])
184
185 def test_text(self, connection):
186 a = self.tables.a

Callers

nothing calls this directly

Calls 12

eq_Function · 0.90
selectFunction · 0.90
literalFunction · 0.90
bindparamFunction · 0.90
allMethod · 0.45
order_byMethod · 0.45
union_allMethod · 0.45
whereMethod · 0.45
labelMethod · 0.45
paramsMethod · 0.45
selectMethod · 0.45
subqueryMethod · 0.45

Tested by

no test coverage detected