codekingpro/portable-devtools
114k
1# testing/suite/test_select.py
2# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
7# mypy: ignore-errors
8
9import collections.abc as collections_abc
10import itertools
11
12from .. import AssertsCompiledSQL
13from .. import AssertsExecutionResults
14from .. import config
15from .. import fixtures
16from ..assertions import assert_raises
17from ..assertions import eq_
18from ..assertions import in_
19from ..assertsql import CursorSQL
20from ..schema import Column
21from ..schema import Table
22from ... import bindparam
23from ... import case
24from ... import column
25from ... import Computed
26from ... import exists
27from ... import false
28from ... import ForeignKey
29from ... import func
30from ... import Identity
31from ... import Integer
32from ... import literal
33from ... import literal_column
34from ... import null
35from ... import select
36from ... import String
37from ... import table
38from ... import testing
39from ... import text
40from ... import true
41from ... import tuple_
42from ... import TupleType
43from ... import union
44from ... import values
45from ...exc import DatabaseError
46from ...exc import ProgrammingError
47
48
49class CollateTest(fixtures.TablesTest):
50 __sparse_driver_backend__ = True
51
52 @classmethod
53 def define_tables(cls, metadata):
54 Table(
55 "some_table",
56 metadata,
57 Column("id", Integer, primary_key=True),
58 Column("data", String(100)),
59 )
60
61 @classmethod
62 def insert_data(cls, connection):
63 connection.execute(
64 cls.tables.some_table.insert(),
65 [
66 {"id": 1, "data": "collate data1"},
67 {"id": 2, "data": "collate data2"},
68 ],
69 )
70
71 def _assert_result(self, select, result):
72 with config.db.connect() as conn:
73 eq_(conn.execute(select).fetchall(), result)
74
75 @testing.requires.order_by_collation
76 def test_collate_order_by(self):
77 collation = testing.requires.get_order_by_collation(testing.config)
78
79 self._assert_result(
80 select(self.tables.some_table).order_by(
81 self.tables.some_table.c.data.collate(collation).asc()
82 ),
83 [(1, "collate data1"), (2, "collate data2")],
84 )
85
86
87class OrderByLabelTest(fixtures.TablesTest):
88 """Test the dialect sends appropriate ORDER BY expressions when
89 labels are used.
90
91 This essentially exercises the "supports_simple_order_by_label"
92 setting.
93
94 """
95
96 __sparse_driver_backend__ = True
97
98 @classmethod
99 def define_tables(cls, metadata):
100 Table(
101 "some_table",
102 metadata,
103 Column("id", Integer, primary_key=True),
104 Column("x", Integer),
105 Column("y", Integer),
106 Column("q", String(50)),
107 Column("p", String(50)),
108 )
109
110 @classmethod
111 def insert_data(cls, connection):
112 connection.execute(
113 cls.tables.some_table.insert(),
114 [
115 {"id": 1, "x": 1, "y": 2, "q": "q1", "p": "p3"},
116 {"id": 2, "x": 2, "y": 3, "q": "q2", "p": "p2"},
117 {"id": 3, "x": 3, "y": 4, "q": "q3", "p": "p1"},
118 ],
119 )
120
121 def _assert_result(self, select, result):
122 with config.db.connect() as conn:
123 eq_(conn.execute(select).fetchall(), result)
124
125 def test_plain(self):
126 table = self.tables.some_table
127 lx = table.c.x.label("lx")
128 self._assert_result(select(lx).order_by(lx), [(1,), (2,), (3,)])
129
130 def test_composed_int(self):
131 table = self.tables.some_table
132 lx = (table.c.x + table.c.y).label("lx")
133 self._assert_result(select(lx).order_by(lx), [(3,), (5,), (7,)])
134
135 def test_composed_multiple(self):
136 table = self.tables.some_table
137 lx = (table.c.x + table.c.y).label("lx")
138 ly = (func.lower(table.c.q) + table.c.p).label("ly")
139 self._assert_result(
140 select(lx, ly).order_by(lx, ly.desc()),
141 [(3, "q1p3"), (5, "q2p2"), (7, "q3p1")],
142 )
143
144 def test_plain_desc(self):
145 table = self.tables.some_table
146 lx = table.c.x.label("lx")
147 self._assert_result(select(lx).order_by(lx.desc()), [(3,), (2,), (1,)])
148
149 def test_composed_int_desc(self):
150 table = self.tables.some_table
151 lx = (table.c.x + table.c.y).label("lx")
152 self._assert_result(select(lx).order_by(lx.desc()), [(7,), (5,), (3,)])
153
154 @testing.requires.group_by_complex_expression
155 def test_group_by_composed(self):
156 table = self.tables.some_table
157 expr = (table.c.x + table.c.y).label("lx")
158 stmt = (
159 select(func.count(table.c.id), expr).group_by(expr).order_by(expr)
160 )
161 self._assert_result(stmt, [(1, 3), (1, 5), (1, 7)])
162
163
164class ValuesExpressionTest(fixtures.TestBase):
165 __requires__ = ("table_value_constructor",)
166
167 __sparse_driver_backend__ = True
168
169 def test_tuples(self, connection):
170 value_expr = values(
171 column("id", Integer), column("name", String), name="my_values"
172 ).data([(1, "name1"), (2, "name2"), (3, "name3")])
173
174 eq_(
175 connection.execute(select(value_expr)).all(),
176 [(1, "name1"), (2, "name2"), (3, "name3")],
177 )
178
179
180class FetchLimitOffsetTest(fixtures.TablesTest):
181 __backend__ = True
182
183 @classmethod
184 def define_tables(cls, metadata):
185 Table(
186 "some_table",
187 metadata,
188 Column("id", Integer, primary_key=True),
189 Column("x", Integer),
190 Column("y", Integer),
191 )
192
193 @classmethod
194 def insert_data(cls, connection):
195 connection.execute(
196 cls.tables.some_table.insert(),
197 [
198 {"id": 1, "x": 1, "y": 2},
199 {"id": 2, "x": 2, "y": 3},
200 {"id": 3, "x": 3, "y": 4},
201 {"id": 4, "x": 4, "y": 5},
202 {"id": 5, "x": 4, "y": 6},
203 ],
204 )
205
206 def _assert_result(
207 self, connection, select, result, params=(), set_=False
208 ):
209 if set_:
210 query_res = connection.execute(select, params).fetchall()
211 eq_(len(query_res), len(result))
212 eq_(set(query_res), set(result))
213
214 else:
215 eq_(connection.execute(select, params).fetchall(), result)
216
217 def _assert_result_str(self, select, result, params=()):
218 with config.db.connect() as conn:
219 eq_(conn.exec_driver_sql(select, params).fetchall(), result)
220
221 def test_simple_limit(self, connection):
222 table = self.tables.some_table
223 stmt = select(table).order_by(table.c.id)
224 self._assert_result(
225 connection,
226 stmt.limit(2),
227 [(1, 1, 2), (2, 2, 3)],
228 )
229 self._assert_result(
230 connection,
231 stmt.limit(3),
232 [(1, 1, 2), (2, 2, 3), (3, 3, 4)],
233 )
234
235 def test_limit_render_multiple_times(self, connection):
236 table = self.tables.some_table
237 stmt = (
238 select(table.c.id).order_by(table.c.id).limit(1).scalar_subquery()
239 )
240
241 u = union(select(stmt), select(stmt)).subquery().select()
242
243 self._assert_result(
244 connection,
245 u,
246 [
247 (1,),
248 ],
249 )
250
251 @testing.requires.fetch_first
252 def test_simple_fetch(self, connection):
253 table = self.tables.some_table
254 self._assert_result(
255 connection,
256 select(table).order_by(table.c.id).fetch(2),
257 [(1, 1, 2), (2, 2, 3)],
258 )
259 self._assert_result(
260 connection,
261 select(table).order_by(table.c.id).fetch(3),
262 [(1, 1, 2), (2, 2, 3), (3, 3, 4)],
263 )
264
265 @testing.requires.offset
266 def test_simple_offset(self, connection):
267 table = self.tables.some_table
268 self._assert_result(
269 connection,
270 select(table).order_by(table.c.id).offset(2),
271 [(3, 3, 4), (4, 4, 5), (5, 4, 6)],
272 )
273 self._assert_result(
274 connection,
275 select(table).order_by(table.c.id).offset(3),
276 [(4, 4, 5), (5, 4, 6)],
277 )
278
279 @testing.combinations(
280 ([(2, 0), (2, 1), (3, 2)]),
281 ([(2, 1), (2, 0), (3, 2)]),
282 ([(3, 1), (2, 1), (3, 1)]),
283 argnames="cases",
284 )
285 @testing.requires.offset
286 def test_simple_limit_offset(self, connection, cases):
287 table = self.tables.some_table
288 connection = connection.execution_options(compiled_cache={})
289
290 assert_data = [(1, 1, 2), (2, 2, 3), (3, 3, 4), (4, 4, 5), (5, 4, 6)]
291
292 for limit, offset in cases:
293 expected = assert_data[offset : offset + limit]
294 self._assert_result(
295 connection,
296 select(table).order_by(table.c.id).limit(limit).offset(offset),
297 expected,
298 )
299
300 @testing.requires.fetch_first
301 def test_simple_fetch_offset(self, connection):
302 table = self.tables.some_table
303 self._assert_result(
304 connection,
305 select(table).order_by(table.c.id).fetch(2).offset(1),
306 [(2, 2, 3), (3, 3, 4)],
307 )
308
309 self._assert_result(
310 connection,
311 select(table).order_by(table.c.id).fetch(3).offset(2),
312 [(3, 3, 4), (4, 4, 5), (5, 4, 6)],
313 )
314
315 @testing.requires.fetch_no_order_by
316 def test_fetch_offset_no_order(self, connection):
317 table = self.tables.some_table
318 self._assert_result(
319 connection,
320 select(table).fetch(10),
321 [(1, 1, 2), (2, 2, 3), (3, 3, 4), (4, 4, 5), (5, 4, 6)],
322 set_=True,
323 )
324
325 @testing.requires.offset
326 def test_simple_offset_zero(self, connection):
327 table = self.tables.some_table
328 self._assert_result(
329 connection,
330 select(table).order_by(table.c.id).offset(0),
331 [(1, 1, 2), (2, 2, 3), (3, 3, 4), (4, 4, 5), (5, 4, 6)],
332 )
333
334 self._assert_result(
335 connection,
336 select(table).order_by(table.c.id).offset(1),
337 [(2, 2, 3), (3, 3, 4), (4, 4, 5), (5, 4, 6)],
338 )
339
340 @testing.requires.offset
341 def test_limit_offset_nobinds(self):
342 """test that 'literal binds' mode works - no bound params."""
343
344 table = self.tables.some_table
345 stmt = select(table).order_by(table.c.id).limit(2).offset(1)
346 sql = stmt.compile(
347 dialect=config.db.dialect, compile_kwargs={"literal_binds": True}
348 )
349 sql = str(sql)
350
351 self._assert_result_str(sql, [(2, 2, 3), (3, 3, 4)])
352
353 @testing.requires.fetch_first
354 def test_fetch_offset_nobinds(self):
355 """test that 'literal binds' mode works - no bound params."""
356
357 table = self.tables.some_table
358 stmt = select(table).order_by(table.c.id).fetch(2).offset(1)
359 sql = stmt.compile(
360 dialect=config.db.dialect, compile_kwargs={"literal_binds": True}
361 )
362 sql = str(sql)
363
364 self._assert_result_str(sql, [(2, 2, 3), (3, 3, 4)])
365
366 @testing.requires.bound_limit_offset
367 def test_bound_limit(self, connection):
368 table = self.tables.some_table
369 self._assert_result(
370 connection,
371 select(table).order_by(table.c.id).limit(bindparam("l")),
372 [(1, 1, 2), (2, 2, 3)],
373 params={"l": 2},
374 )
375
376 self._assert_result(
377 connection,
378 select(table).order_by(table.c.id).limit(bindparam("l")),
379 [(1, 1, 2), (2, 2, 3), (3, 3, 4)],
380 params={"l": 3},
381 )
382
383 @testing.requires.bound_limit_offset
384 def test_bound_offset(self, connection):
385 table = self.tables.some_table
386 self._assert_result(
387 connection,
388 select(table).order_by(table.c.id).offset(bindparam("o")),
389 [(3, 3, 4), (4, 4, 5), (5, 4, 6)],
390 params={"o": 2},
391 )
392
393 self._assert_result(
394 connection,
395 select(table).order_by(table.c.id).offset(bindparam("o")),
396 [(2, 2, 3), (3, 3, 4), (4, 4, 5), (5, 4, 6)],
397 params={"o": 1},
398 )
399
400 @testing.requires.bound_limit_offset
401 def test_bound_limit_offset(self, connection):
402 table = self.tables.some_table
403 self._assert_result(
404 connection,
405 select(table)
406 .order_by(table.c.id)
407 .limit(bindparam("l"))
408 .offset(bindparam("o")),
409 [(2, 2, 3), (3, 3, 4)],
410 params={"l": 2, "o": 1},
411 )
412
413 self._assert_result(
414 connection,
415 select(table)
416 .order_by(table.c.id)
417 .limit(bindparam("l"))
418 .offset(bindparam("o")),
419 [(3, 3, 4), (4, 4, 5), (5, 4, 6)],
420 params={"l": 3, "o": 2},
421 )
422
423 @testing.requires.fetch_first
424 def test_bound_fetch_offset(self, connection):
425 table = self.tables.some_table
426 self._assert_result(
427 connection,
428 select(table)
429 .order_by(table.c.id)
430 .fetch(bindparam("f"))
431 .offset(bindparam("o")),
432 [(2, 2, 3), (3, 3, 4)],
433 params={"f": 2, "o": 1},
434 )
435
436 self._assert_result(
437 connection,
438 select(table)
439 .order_by(table.c.id)
440 .fetch(bindparam("f"))
441 .offset(bindparam("o")),
442 [(3, 3, 4), (4, 4, 5), (5, 4, 6)],
443 params={"f": 3, "o": 2},
444 )
445
446 @testing.requires.sql_expression_limit_offset
447 def test_expr_offset(self, connection):
448 table = self.tables.some_table
449 self._assert_result(
450 connection,
451 select(table)
452 .order_by(table.c.id)
453 .offset(literal_column("1") + literal_column("2")),
454 [(4, 4, 5), (5, 4, 6)],
455 )
456
457 @testing.requires.sql_expression_limit_offset
458 def test_expr_limit(self, connection):
459 table = self.tables.some_table
460 self._assert_result(
461 connection,
462 select(table)
463 .order_by(table.c.id)
464 .limit(literal_column("1") + literal_column("2")),
465 [(1, 1, 2), (2, 2, 3), (3, 3, 4)],
466 )
467
468 @testing.requires.sql_expression_limit_offset
469 def test_expr_limit_offset(self, connection):
470 table = self.tables.some_table
471 self._assert_result(
472 connection,
473 select(table)
474 .order_by(table.c.id)
475 .limit(literal_column("1") + literal_column("1"))
476 .offset(literal_column("1") + literal_column("1")),
477 [(3, 3, 4), (4, 4, 5)],
478 )
479
480 @testing.requires.fetch_first
481 @testing.requires.fetch_expression
482 def test_expr_fetch_offset(self, connection):
483 table = self.tables.some_table
484 self._assert_result(
485 connection,
486 select(table)
487 .order_by(table.c.id)
488 .fetch(literal_column("1") + literal_column("1"))
489 .offset(literal_column("1") + literal_column("1")),
490 [(3, 3, 4), (4, 4, 5)],
491 )
492
493 @testing.requires.sql_expression_limit_offset
494 def test_simple_limit_expr_offset(self, connection):
495 table = self.tables.some_table
496 self._assert_result(
497 connection,
498 select(table)
499 .order_by(table.c.id)
500 .limit(2)
501 .offset(literal_column("1") + literal_column("1")),
502 [(3, 3, 4), (4, 4, 5)],
503 )
504
505 self._assert_result(
506 connection,
507 select(table)
508 .order_by(table.c.id)
509 .limit(3)
510 .offset(literal_column("1") + literal_column("1")),
511 [(3, 3, 4), (4, 4, 5), (5, 4, 6)],
512 )
513
514 @testing.requires.sql_expression_limit_offset
515 def test_expr_limit_simple_offset(self, connection):
516 table = self.tables.some_table
517 self._assert_result(
518 connection,
519 select(table)
520 .order_by(table.c.id)
521 .limit(literal_column("1") + literal_column("1"))
522 .offset(2),
523 [(3, 3, 4), (4, 4, 5)],
524 )
525
526 self._assert_result(
527 connection,
528 select(table)
529 .order_by(table.c.id)
530 .limit(literal_column("1") + literal_column("1"))
531 .offset(1),
532 [(2, 2, 3), (3, 3, 4)],
533 )
534
535 @testing.requires.fetch_ties
536 def test_simple_fetch_ties(self, connection):
537 table = self.tables.some_table
538 self._assert_result(
539 connection,
540 select(table).order_by(table.c.x.desc()).fetch(1, with_ties=True),
541 [(4, 4, 5), (5, 4, 6)],
542 set_=True,
543 )
544
545 self._assert_result(
546 connection,
547 select(table).order_by(table.c.x.desc()).fetch(3, with_ties=True),
548 [(3, 3, 4), (4, 4, 5), (5, 4, 6)],
549 set_=True,
550 )
551
552 @testing.requires.fetch_ties
553 @testing.requires.fetch_offset_with_options
554 def test_fetch_offset_ties(self, connection):
555 table = self.tables.some_table
556 fa = connection.execute(
557 select(table)
558 .order_by(table.c.x)
559 .fetch(2, with_ties=True)
560 .offset(2)
561 ).fetchall()
562 eq_(fa[0], (3, 3, 4))
563 eq_(set(fa), {(3, 3, 4), (4, 4, 5), (5, 4, 6)})
564
565 @testing.requires.fetch_ties
566 @testing.requires.fetch_offset_with_options
567 def test_fetch_offset_ties_exact_number(self, connection):
568 table = self.tables.some_table
569 self._assert_result(
570 connection,
571 select(table)
572 .order_by(table.c.x)
573 .fetch(2, with_ties=True)
574 .offset(1),
575 [(2, 2, 3), (3, 3, 4)],
576 )
577
578 self._assert_result(
579 connection,
580 select(table)
581 .order_by(table.c.x)
582 .fetch(3, with_ties=True)
583 .offset(3),
584 [(4, 4, 5), (5, 4, 6)],
585 )
586
587 @testing.requires.fetch_percent
588 def test_simple_fetch_percent(self, connection):
589 table = self.tables.some_table
590 self._assert_result(
591 connection,
592 select(table).order_by(table.c.id).fetch(20, percent=True),
593 [(1, 1, 2)],
594 )
595
596 @testing.requires.fetch_percent
597 @testing.requires.fetch_offset_with_options
598 def test_fetch_offset_percent(self, connection):
599 table = self.tables.some_table
600 self._assert_result(
601 connection,
602 select(table)
603 .order_by(table.c.id)
604 .fetch(40, percent=True)
605 .offset(1),
606 [(2, 2, 3), (3, 3, 4)],
607 )
608
609 @testing.requires.fetch_ties
610 @testing.requires.fetch_percent
611 def test_simple_fetch_percent_ties(self, connection):
612 table = self.tables.some_table
613 self._assert_result(
614 connection,
615 select(table)
616 .order_by(table.c.x.desc())
617 .fetch(20, percent=True, with_ties=True),
618 [(4, 4, 5), (5, 4, 6)],
619 set_=True,
620 )
621
622 @testing.requires.fetch_ties
623 @testing.requires.fetch_percent
624 @testing.requires.fetch_offset_with_options
625 def test_fetch_offset_percent_ties(self, connection):
626 table = self.tables.some_table
627 fa = connection.execute(
628 select(table)
629 .order_by(table.c.x)
630 .fetch(40, percent=True, with_ties=True)
631 .offset(2)
632 ).fetchall()
633 eq_(fa[0], (3, 3, 4))
634 eq_(set(fa), {(3, 3, 4), (4, 4, 5), (5, 4, 6)})
635
636
637class SameNamedSchemaTableTest(fixtures.TablesTest):
638 """tests for #7471"""
639
640 __sparse_driver_backend__ = True
641
642 __requires__ = ("schemas",)
643
644 @classmethod
645 def define_tables(cls, metadata):
646 Table(
647 "some_table",
648 metadata,
649 Column("id", Integer, primary_key=True),
650 schema=config.test_schema,
651 )
652 Table(
653 "some_table",
654 metadata,
655 Column("id", Integer, primary_key=True),
656 Column(
657 "some_table_id",
658 Integer,
659 # ForeignKey("%s.some_table.id" % config.test_schema),
660 nullable=False,
661 ),
662 )
663
664 @classmethod
665 def insert_data(cls, connection):
666 some_table, some_table_schema = cls.tables(
667 "some_table", "%s.some_table" % config.test_schema
668 )
669 connection.execute(some_table_schema.insert(), {"id": 1})
670 connection.execute(some_table.insert(), {"id": 1, "some_table_id": 1})
671
672 def test_simple_join_both_tables(self, connection):
673 some_table, some_table_schema = self.tables(
674 "some_table", "%s.some_table" % config.test_schema
675 )
676
677 eq_(
678 connection.execute(
679 select(some_table, some_table_schema).join_from(
680 some_table,
681 some_table_schema,
682 some_table.c.some_table_id == some_table_schema.c.id,
683 )
684 ).first(),
685 (1, 1, 1),
686 )
687
688 def test_simple_join_whereclause_only(self, connection):
689 some_table, some_table_schema = self.tables(
690 "some_table", "%s.some_table" % config.test_schema
691 )
692
693 eq_(
694 connection.execute(
695 select(some_table)
696 .join_from(
697 some_table,
698 some_table_schema,
699 some_table.c.some_table_id == some_table_schema.c.id,
700 )
701 .where(some_table.c.id == 1)
702 ).first(),
703 (1, 1),
704 )
705
706 def test_subquery(self, connection):
707 some_table, some_table_schema = self.tables(
708 "some_table", "%s.some_table" % config.test_schema
709 )
710
711 subq = (
712 select(some_table)
713 .join_from(
714 some_table,
715 some_table_schema,
716 some_table.c.some_table_id == some_table_schema.c.id,
717 )
718 .where(some_table.c.id == 1)
719 .subquery()
720 )
721
722 eq_(
723 connection.execute(
724 select(some_table, subq.c.id)
725 .join_from(
726 some_table,
727 subq,
728 some_table.c.some_table_id == subq.c.id,
729 )
730 .where(some_table.c.id == 1)
731 ).first(),
732 (1, 1, 1),
733 )
734
735
736class JoinTest(fixtures.TablesTest):
737 __sparse_driver_backend__ = True
738
739 def _assert_result(self, select, result, params=()):
740 with config.db.connect() as conn:
741 eq_(conn.execute(select, params).fetchall(), result)
742
743 @classmethod
744 def define_tables(cls, metadata):
745 Table("a", metadata, Column("id", Integer, primary_key=True))
746 Table(
747 "b",
748 metadata,
749 Column("id", Integer, primary_key=True),
750 Column("a_id", ForeignKey("a.id"), nullable=False),
751 )
752
753 @classmethod
754 def insert_data(cls, connection):
755 connection.execute(
756 cls.tables.a.insert(),
757 [{"id": 1}, {"id": 2}, {"id": 3}, {"id": 4}, {"id": 5}],
758 )
759
760 connection.execute(
761 cls.tables.b.insert(),
762 [
763 {"id": 1, "a_id": 1},
764 {"id": 2, "a_id": 1},
765 {"id": 4, "a_id": 2},
766 {"id": 5, "a_id": 3},
767 ],
768 )
769
770 def test_inner_join_fk(self):
771 a, b = self.tables("a", "b")
772
773 stmt = select(a, b).select_from(a.join(b)).order_by(a.c.id, b.c.id)
774
775 self._assert_result(stmt, [(1, 1, 1), (1, 2, 1), (2, 4, 2), (3, 5, 3)])
776
777 def test_inner_join_true(self):
778 a, b = self.tables("a", "b")
779
780 stmt = (
781 select(a, b)
782 .select_from(a.join(b, true()))
783 .order_by(a.c.id, b.c.id)
784 )
785
786 self._assert_result(
787 stmt,
788 [
789 (a, b, c)
790 for (a,), (b, c) in itertools.product(
791 [(1,), (2,), (3,), (4,), (5,)],
792 [(1, 1), (2, 1), (4, 2), (5, 3)],
793 )
794 ],
795 )
796
797 def test_inner_join_false(self):
798 a, b = self.tables("a", "b")
799
800 stmt = (
801 select(a, b)
802 .select_from(a.join(b, false()))
803 .order_by(a.c.id, b.c.id)
804 )
805
806 self._assert_result(stmt, [])
807
808 def test_outer_join_false(self):
809 a, b = self.tables("a", "b")
810
811 stmt = (
812 select(a, b)
813 .select_from(a.outerjoin(b, false()))
814 .order_by(a.c.id, b.c.id)
815 )
816
817 self._assert_result(
818 stmt,
819 [
820 (1, None, None),
821 (2, None, None),
822 (3, None, None),
823 (4, None, None),
824 (5, None, None),
825 ],
826 )
827
828 def test_outer_join_fk(self):
829 a, b = self.tables("a", "b")
830
831 stmt = select(a, b).select_from(a.join(b)).order_by(a.c.id, b.c.id)
832
833 self._assert_result(stmt, [(1, 1, 1), (1, 2, 1), (2, 4, 2), (3, 5, 3)])
834
835
836class CompoundSelectTest(fixtures.TablesTest):
837 __sparse_driver_backend__ = True
838
839 @classmethod
840 def define_tables(cls, metadata):
841 Table(
842 "some_table",
843 metadata,
844 Column("id", Integer, primary_key=True),
845 Column("x", Integer),
846 Column("y", Integer),
847 )
848
849 @classmethod
850 def insert_data(cls, connection):
851 connection.execute(
852 cls.tables.some_table.insert(),
853 [
854 {"id": 1, "x": 1, "y": 2},
855 {"id": 2, "x": 2, "y": 3},
856 {"id": 3, "x": 3, "y": 4},
857 {"id": 4, "x": 4, "y": 5},
858 ],
859 )
860
861 def _assert_result(self, select, result, params=()):
862 with config.db.connect() as conn:
863 eq_(conn.execute(select, params).fetchall(), result)
864
865 def test_plain_union(self):
866 table = self.tables.some_table
867 s1 = select(table).where(table.c.id == 2)
868 s2 = select(table).where(table.c.id == 3)
869
870 u1 = union(s1, s2)
871 self._assert_result(
872 u1.order_by(u1.selected_columns.id), [(2, 2, 3), (3, 3, 4)]
873 )
874
875 def test_select_from_plain_union(self):
876 table = self.tables.some_table
877 s1 = select(table).where(table.c.id == 2)
878 s2 = select(table).where(table.c.id == 3)
879
880 u1 = union(s1, s2).alias().select()
881 self._assert_result(
882 u1.order_by(u1.selected_columns.id), [(2, 2, 3), (3, 3, 4)]
883 )
884
885 @testing.requires.order_by_col_from_union
886 @testing.requires.parens_in_union_contained_select_w_limit_offset
887 def test_limit_offset_selectable_in_unions(self):
888 table = self.tables.some_table
889 s1 = select(table).where(table.c.id == 2).limit(1).order_by(table.c.id)
890 s2 = select(table).where(table.c.id == 3).limit(1).order_by(table.c.id)
891
892 u1 = union(s1, s2).limit(2)
893 self._assert_result(
894 u1.order_by(u1.selected_columns.id), [(2, 2, 3), (3, 3, 4)]
895 )
896
897 @testing.requires.parens_in_union_contained_select_wo_limit_offset
898 def test_order_by_selectable_in_unions(self):
899 table = self.tables.some_table
900 s1 = select(table).where(table.c.id == 2).order_by(table.c.id)
901 s2 = select(table).where(table.c.id == 3).order_by(table.c.id)
902
903 u1 = union(s1, s2).limit(2)
904 self._assert_result(
905 u1.order_by(u1.selected_columns.id), [(2, 2, 3), (3, 3, 4)]
906 )
907
908 def test_distinct_selectable_in_unions(self):
909 table = self.tables.some_table
910 s1 = select(table).where(table.c.id == 2).distinct()
911 s2 = select(table).where(table.c.id == 3).distinct()
912
913 u1 = union(s1, s2).limit(2)
914 self._assert_result(
915 u1.order_by(u1.selected_columns.id), [(2, 2, 3), (3, 3, 4)]
916 )
917
918 @testing.requires.parens_in_union_contained_select_w_limit_offset
919 def test_limit_offset_in_unions_from_alias(self):
920 table = self.tables.some_table
921 s1 = select(table).where(table.c.id == 2).limit(1).order_by(table.c.id)
922 s2 = select(table).where(table.c.id == 3).limit(1).order_by(table.c.id)
923
924 # this necessarily has double parens
925 u1 = union(s1, s2).alias()
926 self._assert_result(
927 u1.select().limit(2).order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)]
928 )
929
930 def test_limit_offset_aliased_selectable_in_unions(self):
931 table = self.tables.some_table
932 s1 = (
933 select(table)
934 .where(table.c.id == 2)
935 .limit(1)
936 .order_by(table.c.id)
937 .alias()
938 .select()
939 )
940 s2 = (
941 select(table)
942 .where(table.c.id == 3)
943 .limit(1)
944 .order_by(table.c.id)
945 .alias()
946 .select()
947 )
948
949 u1 = union(s1, s2).limit(2)
950 self._assert_result(
951 u1.order_by(u1.selected_columns.id), [(2, 2, 3), (3, 3, 4)]
952 )
953
954
955class PostCompileParamsTest(
956 AssertsExecutionResults, AssertsCompiledSQL, fixtures.TablesTest
957):
958 __backend__ = True
959
960 __requires__ = ("standard_cursor_sql",)
961
962 @classmethod
963 def define_tables(cls, metadata):
964 Table(
965 "some_table",
966 metadata,
967 Column("id", Integer, primary_key=True),
968 Column("x", Integer),
969 Column("y", Integer),
970 Column("z", String(50)),
971 )
972
973 @classmethod
974 def insert_data(cls, connection):
975 connection.execute(
976 cls.tables.some_table.insert(),
977 [
978 {"id": 1, "x": 1, "y": 2, "z": "z1"},
979 {"id": 2, "x": 2, "y": 3, "z": "z2"},
980 {"id": 3, "x": 3, "y": 4, "z": "z3"},
981 {"id": 4, "x": 4, "y": 5, "z": "z4"},
982 ],
983 )
984
985 def test_compile(self):
986 table = self.tables.some_table
987
988 stmt = select(table.c.id).where(
989 table.c.x == bindparam("q", literal_execute=True)
990 )
991
992 self.assert_compile(
993 stmt,
994 "SELECT some_table.id FROM some_table "
995 "WHERE some_table.x = __[POSTCOMPILE_q]",
996 {},
997 )
998
999 def test_compile_literal_binds(self):
1000 table = self.tables.some_table
1001
1002 stmt = select(table.c.id).where(
1003 table.c.x == bindparam("q", 10, literal_execute=True)
1004 )
1005
1006 self.assert_compile(
1007 stmt,
1008 "SELECT some_table.id FROM some_table WHERE some_table.x = 10",
1009 {},
1010 literal_binds=True,
1011 )
1012
1013 def test_execute(self):
1014 table = self.tables.some_table
1015
1016 stmt = select(table.c.id).where(
1017 table.c.x == bindparam("q", literal_execute=True)
1018 )
1019
1020 with self.sql_execution_asserter() as asserter:
1021 with config.db.connect() as conn:
1022 conn.execute(stmt, dict(q=10))
1023
1024 asserter.assert_(
1025 CursorSQL(
1026 "SELECT some_table.id \nFROM some_table "
1027 "\nWHERE some_table.x = 10",
1028 () if config.db.dialect.positional else {},
1029 )
1030 )
1031
1032 def test_execute_expanding_plus_literal_execute(self):
1033 table = self.tables.some_table
1034
1035 stmt = select(table.c.id).where(
1036 table.c.x.in_(bindparam("q", expanding=True, literal_execute=True))
1037 )
1038
1039 with self.sql_execution_asserter() as asserter:
1040 with config.db.connect() as conn:
1041 conn.execute(stmt, dict(q=[5, 6, 7]))
1042
1043 asserter.assert_(
1044 CursorSQL(
1045 "SELECT some_table.id \nFROM some_table "
1046 "\nWHERE some_table.x IN (5, 6, 7)",
1047 () if config.db.dialect.positional else {},
1048 )
1049 )
1050
1051 @testing.requires.tuple_in
1052 def test_execute_tuple_expanding_plus_literal_execute(self):
1053 table = self.tables.some_table
1054
1055 stmt = select(table.c.id).where(
1056 tuple_(table.c.x, table.c.y).in_(
1057 bindparam("q", expanding=True, literal_execute=True)
1058 )
1059 )
1060
1061 with self.sql_execution_asserter() as asserter:
1062 with config.db.connect() as conn:
1063 conn.execute(stmt, dict(q=[(5, 10), (12, 18)]))
1064
1065 asserter.assert_(
1066 CursorSQL(
1067 "SELECT some_table.id \nFROM some_table "
1068 "\nWHERE (some_table.x, some_table.y) "
1069 "IN (%s(5, 10), (12, 18))"
1070 % ("VALUES " if config.db.dialect.tuple_in_values else ""),
1071 () if config.db.dialect.positional else {},
1072 )
1073 )
1074
1075 @testing.requires.tuple_in
1076 def test_execute_tuple_expanding_plus_literal_heterogeneous_execute(self):
1077 table = self.tables.some_table
1078
1079 stmt = select(table.c.id).where(
1080 tuple_(table.c.x, table.c.z).in_(
1081 bindparam("q", expanding=True, literal_execute=True)
1082 )
1083 )
1084
1085 with self.sql_execution_asserter() as asserter:
1086 with config.db.connect() as conn:
1087 conn.execute(stmt, dict(q=[(5, "z1"), (12, "z3")]))
1088
1089 asserter.assert_(
1090 CursorSQL(
1091 "SELECT some_table.id \nFROM some_table "
1092 "\nWHERE (some_table.x, some_table.z) "
1093 "IN (%s(5, 'z1'), (12, 'z3'))"
1094 % ("VALUES " if config.db.dialect.tuple_in_values else ""),
1095 () if config.db.dialect.positional else {},
1096 )
1097 )
1098
1099
1100class ExpandingBoundInTest(fixtures.TablesTest):
1101 __backend__ = True
1102
1103 @classmethod
1104 def define_tables(cls, metadata):
1105 Table(
1106 "some_table",
1107 metadata,
1108 Column("id", Integer, primary_key=True),
1109 Column("x", Integer),
1110 Column("y", Integer),
1111 Column("z", String(50)),
1112 )
1113
1114 @classmethod
1115 def insert_data(cls, connection):
1116 connection.execute(
1117 cls.tables.some_table.insert(),
1118 [
1119 {"id": 1, "x": 1, "y": 2, "z": "z1"},
1120 {"id": 2, "x": 2, "y": 3, "z": "z2"},
1121 {"id": 3, "x": 3, "y": 4, "z": "z3"},
1122 {"id": 4, "x": 4, "y": 5, "z": "z4"},
1123 ],
1124 )
1125
1126 def _assert_result(self, select, result, params=()):
1127 with config.db.connect() as conn:
1128 eq_(conn.execute(select, params).fetchall(), result)
1129
1130 def test_multiple_empty_sets_bindparam(self):
1131 # test that any anonymous aliasing used by the dialect
1132 # is fine with duplicates
1133 table = self.tables.some_table
1134 stmt = (
1135 select(table.c.id)
1136 .where(table.c.x.in_(bindparam("q")))
1137 .where(table.c.y.in_(bindparam("p")))
1138 .order_by(table.c.id)
1139 )
1140 self._assert_result(stmt, [], params={"q": [], "p": []})
1141
1142 def test_multiple_empty_sets_direct(self):
1143 # test that any anonymous aliasing used by the dialect
1144 # is fine with duplicates
1145 table = self.tables.some_table
1146 stmt = (
1147 select(table.c.id)
1148 .where(table.c.x.in_([]))
1149 .where(table.c.y.in_([]))
1150 .order_by(table.c.id)
1151 )
1152 self._assert_result(stmt, [])
1153
1154 @testing.requires.tuple_in_w_empty
1155 def test_empty_heterogeneous_tuples_bindparam(self):
1156 table = self.tables.some_table
1157 stmt = (
1158 select(table.c.id)
1159 .where(tuple_(table.c.x, table.c.z).in_(bindparam("q")))
1160 .order_by(table.c.id)
1161 )
1162 self._assert_result(stmt, [], params={"q": []})
1163
1164 @testing.requires.tuple_in_w_empty
1165 def test_empty_heterogeneous_tuples_direct(self):
1166 table = self.tables.some_table
1167
1168 def go(val, expected):
1169 stmt = (
1170 select(table.c.id)
1171 .where(tuple_(table.c.x, table.c.z).in_(val))
1172 .order_by(table.c.id)
1173 )
1174 self._assert_result(stmt, expected)
1175
1176 go([], [])
1177 go([(2, "z2"), (3, "z3"), (4, "z4")], [(2,), (3,), (4,)])
1178 go([], [])
1179
1180 @testing.requires.tuple_in_w_empty
1181 def test_empty_homogeneous_tuples_bindparam(self):
1182 table = self.tables.some_table
1183 stmt = (
1184 select(table.c.id)
1185 .where(tuple_(table.c.x, table.c.y).in_(bindparam("q")))
1186 .order_by(table.c.id)
1187 )
1188 self._assert_result(stmt, [], params={"q": []})
1189
1190 @testing.requires.tuple_in_w_empty
1191 def test_empty_homogeneous_tuples_direct(self):
1192 table = self.tables.some_table
1193
1194 def go(val, expected):
1195 stmt = (
1196 select(table.c.id)
1197 .where(tuple_(table.c.x, table.c.y).in_(val))
1198 .order_by(table.c.id)
1199 )
1200 self._assert_result(stmt, expected)