codekingpro/portable-devtools
114k
1# testing/suite/test_select.py
2# Copyright (C) 2005-2024 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 __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 __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 __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 = select(table.c.id).limit(1).scalar_subquery()
238
239 u = union(select(stmt), select(stmt)).subquery().select()
240
241 self._assert_result(
242 connection,
243 u,
244 [
245 (1,),
246 ],
247 )
248
249 @testing.requires.fetch_first
250 def test_simple_fetch(self, connection):
251 table = self.tables.some_table
252 self._assert_result(
253 connection,
254 select(table).order_by(table.c.id).fetch(2),
255 [(1, 1, 2), (2, 2, 3)],
256 )
257 self._assert_result(
258 connection,
259 select(table).order_by(table.c.id).fetch(3),
260 [(1, 1, 2), (2, 2, 3), (3, 3, 4)],
261 )
262
263 @testing.requires.offset
264 def test_simple_offset(self, connection):
265 table = self.tables.some_table
266 self._assert_result(
267 connection,
268 select(table).order_by(table.c.id).offset(2),
269 [(3, 3, 4), (4, 4, 5), (5, 4, 6)],
270 )
271 self._assert_result(
272 connection,
273 select(table).order_by(table.c.id).offset(3),
274 [(4, 4, 5), (5, 4, 6)],
275 )
276
277 @testing.combinations(
278 ([(2, 0), (2, 1), (3, 2)]),
279 ([(2, 1), (2, 0), (3, 2)]),
280 ([(3, 1), (2, 1), (3, 1)]),
281 argnames="cases",
282 )
283 @testing.requires.offset
284 def test_simple_limit_offset(self, connection, cases):
285 table = self.tables.some_table
286 connection = connection.execution_options(compiled_cache={})
287
288 assert_data = [(1, 1, 2), (2, 2, 3), (3, 3, 4), (4, 4, 5), (5, 4, 6)]
289
290 for limit, offset in cases:
291 expected = assert_data[offset : offset + limit]
292 self._assert_result(
293 connection,
294 select(table).order_by(table.c.id).limit(limit).offset(offset),
295 expected,
296 )
297
298 @testing.requires.fetch_first
299 def test_simple_fetch_offset(self, connection):
300 table = self.tables.some_table
301 self._assert_result(
302 connection,
303 select(table).order_by(table.c.id).fetch(2).offset(1),
304 [(2, 2, 3), (3, 3, 4)],
305 )
306
307 self._assert_result(
308 connection,
309 select(table).order_by(table.c.id).fetch(3).offset(2),
310 [(3, 3, 4), (4, 4, 5), (5, 4, 6)],
311 )
312
313 @testing.requires.fetch_no_order_by
314 def test_fetch_offset_no_order(self, connection):
315 table = self.tables.some_table
316 self._assert_result(
317 connection,
318 select(table).fetch(10),
319 [(1, 1, 2), (2, 2, 3), (3, 3, 4), (4, 4, 5), (5, 4, 6)],
320 set_=True,
321 )
322
323 @testing.requires.offset
324 def test_simple_offset_zero(self, connection):
325 table = self.tables.some_table
326 self._assert_result(
327 connection,
328 select(table).order_by(table.c.id).offset(0),
329 [(1, 1, 2), (2, 2, 3), (3, 3, 4), (4, 4, 5), (5, 4, 6)],
330 )
331
332 self._assert_result(
333 connection,
334 select(table).order_by(table.c.id).offset(1),
335 [(2, 2, 3), (3, 3, 4), (4, 4, 5), (5, 4, 6)],
336 )
337
338 @testing.requires.offset
339 def test_limit_offset_nobinds(self):
340 """test that 'literal binds' mode works - no bound params."""
341
342 table = self.tables.some_table
343 stmt = select(table).order_by(table.c.id).limit(2).offset(1)
344 sql = stmt.compile(
345 dialect=config.db.dialect, compile_kwargs={"literal_binds": True}
346 )
347 sql = str(sql)
348
349 self._assert_result_str(sql, [(2, 2, 3), (3, 3, 4)])
350
351 @testing.requires.fetch_first
352 def test_fetch_offset_nobinds(self):
353 """test that 'literal binds' mode works - no bound params."""
354
355 table = self.tables.some_table
356 stmt = select(table).order_by(table.c.id).fetch(2).offset(1)
357 sql = stmt.compile(
358 dialect=config.db.dialect, compile_kwargs={"literal_binds": True}
359 )
360 sql = str(sql)
361
362 self._assert_result_str(sql, [(2, 2, 3), (3, 3, 4)])
363
364 @testing.requires.bound_limit_offset
365 def test_bound_limit(self, connection):
366 table = self.tables.some_table
367 self._assert_result(
368 connection,
369 select(table).order_by(table.c.id).limit(bindparam("l")),
370 [(1, 1, 2), (2, 2, 3)],
371 params={"l": 2},
372 )
373
374 self._assert_result(
375 connection,
376 select(table).order_by(table.c.id).limit(bindparam("l")),
377 [(1, 1, 2), (2, 2, 3), (3, 3, 4)],
378 params={"l": 3},
379 )
380
381 @testing.requires.bound_limit_offset
382 def test_bound_offset(self, connection):
383 table = self.tables.some_table
384 self._assert_result(
385 connection,
386 select(table).order_by(table.c.id).offset(bindparam("o")),
387 [(3, 3, 4), (4, 4, 5), (5, 4, 6)],
388 params={"o": 2},
389 )
390
391 self._assert_result(
392 connection,
393 select(table).order_by(table.c.id).offset(bindparam("o")),
394 [(2, 2, 3), (3, 3, 4), (4, 4, 5), (5, 4, 6)],
395 params={"o": 1},
396 )
397
398 @testing.requires.bound_limit_offset
399 def test_bound_limit_offset(self, connection):
400 table = self.tables.some_table
401 self._assert_result(
402 connection,
403 select(table)
404 .order_by(table.c.id)
405 .limit(bindparam("l"))
406 .offset(bindparam("o")),
407 [(2, 2, 3), (3, 3, 4)],
408 params={"l": 2, "o": 1},
409 )
410
411 self._assert_result(
412 connection,
413 select(table)
414 .order_by(table.c.id)
415 .limit(bindparam("l"))
416 .offset(bindparam("o")),
417 [(3, 3, 4), (4, 4, 5), (5, 4, 6)],
418 params={"l": 3, "o": 2},
419 )
420
421 @testing.requires.fetch_first
422 def test_bound_fetch_offset(self, connection):
423 table = self.tables.some_table
424 self._assert_result(
425 connection,
426 select(table)
427 .order_by(table.c.id)
428 .fetch(bindparam("f"))
429 .offset(bindparam("o")),
430 [(2, 2, 3), (3, 3, 4)],
431 params={"f": 2, "o": 1},
432 )
433
434 self._assert_result(
435 connection,
436 select(table)
437 .order_by(table.c.id)
438 .fetch(bindparam("f"))
439 .offset(bindparam("o")),
440 [(3, 3, 4), (4, 4, 5), (5, 4, 6)],
441 params={"f": 3, "o": 2},
442 )
443
444 @testing.requires.sql_expression_limit_offset
445 def test_expr_offset(self, connection):
446 table = self.tables.some_table
447 self._assert_result(
448 connection,
449 select(table)
450 .order_by(table.c.id)
451 .offset(literal_column("1") + literal_column("2")),
452 [(4, 4, 5), (5, 4, 6)],
453 )
454
455 @testing.requires.sql_expression_limit_offset
456 def test_expr_limit(self, connection):
457 table = self.tables.some_table
458 self._assert_result(
459 connection,
460 select(table)
461 .order_by(table.c.id)
462 .limit(literal_column("1") + literal_column("2")),
463 [(1, 1, 2), (2, 2, 3), (3, 3, 4)],
464 )
465
466 @testing.requires.sql_expression_limit_offset
467 def test_expr_limit_offset(self, connection):
468 table = self.tables.some_table
469 self._assert_result(
470 connection,
471 select(table)
472 .order_by(table.c.id)
473 .limit(literal_column("1") + literal_column("1"))
474 .offset(literal_column("1") + literal_column("1")),
475 [(3, 3, 4), (4, 4, 5)],
476 )
477
478 @testing.requires.fetch_first
479 @testing.requires.fetch_expression
480 def test_expr_fetch_offset(self, connection):
481 table = self.tables.some_table
482 self._assert_result(
483 connection,
484 select(table)
485 .order_by(table.c.id)
486 .fetch(literal_column("1") + literal_column("1"))
487 .offset(literal_column("1") + literal_column("1")),
488 [(3, 3, 4), (4, 4, 5)],
489 )
490
491 @testing.requires.sql_expression_limit_offset
492 def test_simple_limit_expr_offset(self, connection):
493 table = self.tables.some_table
494 self._assert_result(
495 connection,
496 select(table)
497 .order_by(table.c.id)
498 .limit(2)
499 .offset(literal_column("1") + literal_column("1")),
500 [(3, 3, 4), (4, 4, 5)],
501 )
502
503 self._assert_result(
504 connection,
505 select(table)
506 .order_by(table.c.id)
507 .limit(3)
508 .offset(literal_column("1") + literal_column("1")),
509 [(3, 3, 4), (4, 4, 5), (5, 4, 6)],
510 )
511
512 @testing.requires.sql_expression_limit_offset
513 def test_expr_limit_simple_offset(self, connection):
514 table = self.tables.some_table
515 self._assert_result(
516 connection,
517 select(table)
518 .order_by(table.c.id)
519 .limit(literal_column("1") + literal_column("1"))
520 .offset(2),
521 [(3, 3, 4), (4, 4, 5)],
522 )
523
524 self._assert_result(
525 connection,
526 select(table)
527 .order_by(table.c.id)
528 .limit(literal_column("1") + literal_column("1"))
529 .offset(1),
530 [(2, 2, 3), (3, 3, 4)],
531 )
532
533 @testing.requires.fetch_ties
534 def test_simple_fetch_ties(self, connection):
535 table = self.tables.some_table
536 self._assert_result(
537 connection,
538 select(table).order_by(table.c.x.desc()).fetch(1, with_ties=True),
539 [(4, 4, 5), (5, 4, 6)],
540 set_=True,
541 )
542
543 self._assert_result(
544 connection,
545 select(table).order_by(table.c.x.desc()).fetch(3, with_ties=True),
546 [(3, 3, 4), (4, 4, 5), (5, 4, 6)],
547 set_=True,
548 )
549
550 @testing.requires.fetch_ties
551 @testing.requires.fetch_offset_with_options
552 def test_fetch_offset_ties(self, connection):
553 table = self.tables.some_table
554 fa = connection.execute(
555 select(table)
556 .order_by(table.c.x)
557 .fetch(2, with_ties=True)
558 .offset(2)
559 ).fetchall()
560 eq_(fa[0], (3, 3, 4))
561 eq_(set(fa), {(3, 3, 4), (4, 4, 5), (5, 4, 6)})
562
563 @testing.requires.fetch_ties
564 @testing.requires.fetch_offset_with_options
565 def test_fetch_offset_ties_exact_number(self, connection):
566 table = self.tables.some_table
567 self._assert_result(
568 connection,
569 select(table)
570 .order_by(table.c.x)
571 .fetch(2, with_ties=True)
572 .offset(1),
573 [(2, 2, 3), (3, 3, 4)],
574 )
575
576 self._assert_result(
577 connection,
578 select(table)
579 .order_by(table.c.x)
580 .fetch(3, with_ties=True)
581 .offset(3),
582 [(4, 4, 5), (5, 4, 6)],
583 )
584
585 @testing.requires.fetch_percent
586 def test_simple_fetch_percent(self, connection):
587 table = self.tables.some_table
588 self._assert_result(
589 connection,
590 select(table).order_by(table.c.id).fetch(20, percent=True),
591 [(1, 1, 2)],
592 )
593
594 @testing.requires.fetch_percent
595 @testing.requires.fetch_offset_with_options
596 def test_fetch_offset_percent(self, connection):
597 table = self.tables.some_table
598 self._assert_result(
599 connection,
600 select(table)
601 .order_by(table.c.id)
602 .fetch(40, percent=True)
603 .offset(1),
604 [(2, 2, 3), (3, 3, 4)],
605 )
606
607 @testing.requires.fetch_ties
608 @testing.requires.fetch_percent
609 def test_simple_fetch_percent_ties(self, connection):
610 table = self.tables.some_table
611 self._assert_result(
612 connection,
613 select(table)
614 .order_by(table.c.x.desc())
615 .fetch(20, percent=True, with_ties=True),
616 [(4, 4, 5), (5, 4, 6)],
617 set_=True,
618 )
619
620 @testing.requires.fetch_ties
621 @testing.requires.fetch_percent
622 @testing.requires.fetch_offset_with_options
623 def test_fetch_offset_percent_ties(self, connection):
624 table = self.tables.some_table
625 fa = connection.execute(
626 select(table)
627 .order_by(table.c.x)
628 .fetch(40, percent=True, with_ties=True)
629 .offset(2)
630 ).fetchall()
631 eq_(fa[0], (3, 3, 4))
632 eq_(set(fa), {(3, 3, 4), (4, 4, 5), (5, 4, 6)})
633
634
635class SameNamedSchemaTableTest(fixtures.TablesTest):
636 """tests for #7471"""
637
638 __backend__ = True
639
640 __requires__ = ("schemas",)
641
642 @classmethod
643 def define_tables(cls, metadata):
644 Table(
645 "some_table",
646 metadata,
647 Column("id", Integer, primary_key=True),
648 schema=config.test_schema,
649 )
650 Table(
651 "some_table",
652 metadata,
653 Column("id", Integer, primary_key=True),
654 Column(
655 "some_table_id",
656 Integer,
657 # ForeignKey("%s.some_table.id" % config.test_schema),
658 nullable=False,
659 ),
660 )
661
662 @classmethod
663 def insert_data(cls, connection):
664 some_table, some_table_schema = cls.tables(
665 "some_table", "%s.some_table" % config.test_schema
666 )
667 connection.execute(some_table_schema.insert(), {"id": 1})
668 connection.execute(some_table.insert(), {"id": 1, "some_table_id": 1})
669
670 def test_simple_join_both_tables(self, connection):
671 some_table, some_table_schema = self.tables(
672 "some_table", "%s.some_table" % config.test_schema
673 )
674
675 eq_(
676 connection.execute(
677 select(some_table, some_table_schema).join_from(
678 some_table,
679 some_table_schema,
680 some_table.c.some_table_id == some_table_schema.c.id,
681 )
682 ).first(),
683 (1, 1, 1),
684 )
685
686 def test_simple_join_whereclause_only(self, connection):
687 some_table, some_table_schema = self.tables(
688 "some_table", "%s.some_table" % config.test_schema
689 )
690
691 eq_(
692 connection.execute(
693 select(some_table)
694 .join_from(
695 some_table,
696 some_table_schema,
697 some_table.c.some_table_id == some_table_schema.c.id,
698 )
699 .where(some_table.c.id == 1)
700 ).first(),
701 (1, 1),
702 )
703
704 def test_subquery(self, connection):
705 some_table, some_table_schema = self.tables(
706 "some_table", "%s.some_table" % config.test_schema
707 )
708
709 subq = (
710 select(some_table)
711 .join_from(
712 some_table,
713 some_table_schema,
714 some_table.c.some_table_id == some_table_schema.c.id,
715 )
716 .where(some_table.c.id == 1)
717 .subquery()
718 )
719
720 eq_(
721 connection.execute(
722 select(some_table, subq.c.id)
723 .join_from(
724 some_table,
725 subq,
726 some_table.c.some_table_id == subq.c.id,
727 )
728 .where(some_table.c.id == 1)
729 ).first(),
730 (1, 1, 1),
731 )
732
733
734class JoinTest(fixtures.TablesTest):
735 __backend__ = True
736
737 def _assert_result(self, select, result, params=()):
738 with config.db.connect() as conn:
739 eq_(conn.execute(select, params).fetchall(), result)
740
741 @classmethod
742 def define_tables(cls, metadata):
743 Table("a", metadata, Column("id", Integer, primary_key=True))
744 Table(
745 "b",
746 metadata,
747 Column("id", Integer, primary_key=True),
748 Column("a_id", ForeignKey("a.id"), nullable=False),
749 )
750
751 @classmethod
752 def insert_data(cls, connection):
753 connection.execute(
754 cls.tables.a.insert(),
755 [{"id": 1}, {"id": 2}, {"id": 3}, {"id": 4}, {"id": 5}],
756 )
757
758 connection.execute(
759 cls.tables.b.insert(),
760 [
761 {"id": 1, "a_id": 1},
762 {"id": 2, "a_id": 1},
763 {"id": 4, "a_id": 2},
764 {"id": 5, "a_id": 3},
765 ],
766 )
767
768 def test_inner_join_fk(self):
769 a, b = self.tables("a", "b")
770
771 stmt = select(a, b).select_from(a.join(b)).order_by(a.c.id, b.c.id)
772
773 self._assert_result(stmt, [(1, 1, 1), (1, 2, 1), (2, 4, 2), (3, 5, 3)])
774
775 def test_inner_join_true(self):
776 a, b = self.tables("a", "b")
777
778 stmt = (
779 select(a, b)
780 .select_from(a.join(b, true()))
781 .order_by(a.c.id, b.c.id)
782 )
783
784 self._assert_result(
785 stmt,
786 [
787 (a, b, c)
788 for (a,), (b, c) in itertools.product(
789 [(1,), (2,), (3,), (4,), (5,)],
790 [(1, 1), (2, 1), (4, 2), (5, 3)],
791 )
792 ],
793 )
794
795 def test_inner_join_false(self):
796 a, b = self.tables("a", "b")
797
798 stmt = (
799 select(a, b)
800 .select_from(a.join(b, false()))
801 .order_by(a.c.id, b.c.id)
802 )
803
804 self._assert_result(stmt, [])
805
806 def test_outer_join_false(self):
807 a, b = self.tables("a", "b")
808
809 stmt = (
810 select(a, b)
811 .select_from(a.outerjoin(b, false()))
812 .order_by(a.c.id, b.c.id)
813 )
814
815 self._assert_result(
816 stmt,
817 [
818 (1, None, None),
819 (2, None, None),
820 (3, None, None),
821 (4, None, None),
822 (5, None, None),
823 ],
824 )
825
826 def test_outer_join_fk(self):
827 a, b = self.tables("a", "b")
828
829 stmt = select(a, b).select_from(a.join(b)).order_by(a.c.id, b.c.id)
830
831 self._assert_result(stmt, [(1, 1, 1), (1, 2, 1), (2, 4, 2), (3, 5, 3)])
832
833
834class CompoundSelectTest(fixtures.TablesTest):
835 __backend__ = True
836
837 @classmethod
838 def define_tables(cls, metadata):
839 Table(
840 "some_table",
841 metadata,
842 Column("id", Integer, primary_key=True),
843 Column("x", Integer),
844 Column("y", Integer),
845 )
846
847 @classmethod
848 def insert_data(cls, connection):
849 connection.execute(
850 cls.tables.some_table.insert(),
851 [
852 {"id": 1, "x": 1, "y": 2},
853 {"id": 2, "x": 2, "y": 3},
854 {"id": 3, "x": 3, "y": 4},
855 {"id": 4, "x": 4, "y": 5},
856 ],
857 )
858
859 def _assert_result(self, select, result, params=()):
860 with config.db.connect() as conn:
861 eq_(conn.execute(select, params).fetchall(), result)
862
863 def test_plain_union(self):
864 table = self.tables.some_table
865 s1 = select(table).where(table.c.id == 2)
866 s2 = select(table).where(table.c.id == 3)
867
868 u1 = union(s1, s2)
869 self._assert_result(
870 u1.order_by(u1.selected_columns.id), [(2, 2, 3), (3, 3, 4)]
871 )
872
873 def test_select_from_plain_union(self):
874 table = self.tables.some_table
875 s1 = select(table).where(table.c.id == 2)
876 s2 = select(table).where(table.c.id == 3)
877
878 u1 = union(s1, s2).alias().select()
879 self._assert_result(
880 u1.order_by(u1.selected_columns.id), [(2, 2, 3), (3, 3, 4)]
881 )
882
883 @testing.requires.order_by_col_from_union
884 @testing.requires.parens_in_union_contained_select_w_limit_offset
885 def test_limit_offset_selectable_in_unions(self):
886 table = self.tables.some_table
887 s1 = select(table).where(table.c.id == 2).limit(1).order_by(table.c.id)
888 s2 = select(table).where(table.c.id == 3).limit(1).order_by(table.c.id)
889
890 u1 = union(s1, s2).limit(2)
891 self._assert_result(
892 u1.order_by(u1.selected_columns.id), [(2, 2, 3), (3, 3, 4)]
893 )
894
895 @testing.requires.parens_in_union_contained_select_wo_limit_offset
896 def test_order_by_selectable_in_unions(self):
897 table = self.tables.some_table
898 s1 = select(table).where(table.c.id == 2).order_by(table.c.id)
899 s2 = select(table).where(table.c.id == 3).order_by(table.c.id)
900
901 u1 = union(s1, s2).limit(2)
902 self._assert_result(
903 u1.order_by(u1.selected_columns.id), [(2, 2, 3), (3, 3, 4)]
904 )
905
906 def test_distinct_selectable_in_unions(self):
907 table = self.tables.some_table
908 s1 = select(table).where(table.c.id == 2).distinct()
909 s2 = select(table).where(table.c.id == 3).distinct()
910
911 u1 = union(s1, s2).limit(2)
912 self._assert_result(
913 u1.order_by(u1.selected_columns.id), [(2, 2, 3), (3, 3, 4)]
914 )
915
916 @testing.requires.parens_in_union_contained_select_w_limit_offset
917 def test_limit_offset_in_unions_from_alias(self):
918 table = self.tables.some_table
919 s1 = select(table).where(table.c.id == 2).limit(1).order_by(table.c.id)
920 s2 = select(table).where(table.c.id == 3).limit(1).order_by(table.c.id)
921
922 # this necessarily has double parens
923 u1 = union(s1, s2).alias()
924 self._assert_result(
925 u1.select().limit(2).order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)]
926 )
927
928 def test_limit_offset_aliased_selectable_in_unions(self):
929 table = self.tables.some_table
930 s1 = (
931 select(table)
932 .where(table.c.id == 2)
933 .limit(1)
934 .order_by(table.c.id)
935 .alias()
936 .select()
937 )
938 s2 = (
939 select(table)
940 .where(table.c.id == 3)
941 .limit(1)
942 .order_by(table.c.id)
943 .alias()
944 .select()
945 )
946
947 u1 = union(s1, s2).limit(2)
948 self._assert_result(
949 u1.order_by(u1.selected_columns.id), [(2, 2, 3), (3, 3, 4)]
950 )
951
952
953class PostCompileParamsTest(
954 AssertsExecutionResults, AssertsCompiledSQL, fixtures.TablesTest
955):
956 __backend__ = True
957
958 __requires__ = ("standard_cursor_sql",)
959
960 @classmethod
961 def define_tables(cls, metadata):
962 Table(
963 "some_table",
964 metadata,
965 Column("id", Integer, primary_key=True),
966 Column("x", Integer),
967 Column("y", Integer),
968 Column("z", String(50)),
969 )
970
971 @classmethod
972 def insert_data(cls, connection):
973 connection.execute(
974 cls.tables.some_table.insert(),
975 [
976 {"id": 1, "x": 1, "y": 2, "z": "z1"},
977 {"id": 2, "x": 2, "y": 3, "z": "z2"},
978 {"id": 3, "x": 3, "y": 4, "z": "z3"},
979 {"id": 4, "x": 4, "y": 5, "z": "z4"},
980 ],
981 )
982
983 def test_compile(self):
984 table = self.tables.some_table
985
986 stmt = select(table.c.id).where(
987 table.c.x == bindparam("q", literal_execute=True)
988 )
989
990 self.assert_compile(
991 stmt,
992 "SELECT some_table.id FROM some_table "
993 "WHERE some_table.x = __[POSTCOMPILE_q]",
994 {},
995 )
996
997 def test_compile_literal_binds(self):
998 table = self.tables.some_table
999
1000 stmt = select(table.c.id).where(
1001 table.c.x == bindparam("q", 10, literal_execute=True)
1002 )
1003
1004 self.assert_compile(
1005 stmt,
1006 "SELECT some_table.id FROM some_table WHERE some_table.x = 10",
1007 {},
1008 literal_binds=True,
1009 )
1010
1011 def test_execute(self):
1012 table = self.tables.some_table
1013
1014 stmt = select(table.c.id).where(
1015 table.c.x == bindparam("q", literal_execute=True)
1016 )
1017
1018 with self.sql_execution_asserter() as asserter:
1019 with config.db.connect() as conn:
1020 conn.execute(stmt, dict(q=10))
1021
1022 asserter.assert_(
1023 CursorSQL(
1024 "SELECT some_table.id \nFROM some_table "
1025 "\nWHERE some_table.x = 10",
1026 () if config.db.dialect.positional else {},
1027 )
1028 )
1029
1030 def test_execute_expanding_plus_literal_execute(self):
1031 table = self.tables.some_table
1032
1033 stmt = select(table.c.id).where(
1034 table.c.x.in_(bindparam("q", expanding=True, literal_execute=True))
1035 )
1036
1037 with self.sql_execution_asserter() as asserter:
1038 with config.db.connect() as conn:
1039 conn.execute(stmt, dict(q=[5, 6, 7]))
1040
1041 asserter.assert_(
1042 CursorSQL(
1043 "SELECT some_table.id \nFROM some_table "
1044 "\nWHERE some_table.x IN (5, 6, 7)",
1045 () if config.db.dialect.positional else {},
1046 )
1047 )
1048
1049 @testing.requires.tuple_in
1050 def test_execute_tuple_expanding_plus_literal_execute(self):
1051 table = self.tables.some_table
1052
1053 stmt = select(table.c.id).where(
1054 tuple_(table.c.x, table.c.y).in_(
1055 bindparam("q", expanding=True, literal_execute=True)
1056 )
1057 )
1058
1059 with self.sql_execution_asserter() as asserter:
1060 with config.db.connect() as conn:
1061 conn.execute(stmt, dict(q=[(5, 10), (12, 18)]))
1062
1063 asserter.assert_(
1064 CursorSQL(
1065 "SELECT some_table.id \nFROM some_table "
1066 "\nWHERE (some_table.x, some_table.y) "
1067 "IN (%s(5, 10), (12, 18))"
1068 % ("VALUES " if config.db.dialect.tuple_in_values else ""),
1069 () if config.db.dialect.positional else {},
1070 )
1071 )
1072
1073 @testing.requires.tuple_in
1074 def test_execute_tuple_expanding_plus_literal_heterogeneous_execute(self):
1075 table = self.tables.some_table
1076
1077 stmt = select(table.c.id).where(
1078 tuple_(table.c.x, table.c.z).in_(
1079 bindparam("q", expanding=True, literal_execute=True)
1080 )
1081 )
1082
1083 with self.sql_execution_asserter() as asserter:
1084 with config.db.connect() as conn:
1085 conn.execute(stmt, dict(q=[(5, "z1"), (12, "z3")]))
1086
1087 asserter.assert_(
1088 CursorSQL(
1089 "SELECT some_table.id \nFROM some_table "
1090 "\nWHERE (some_table.x, some_table.z) "
1091 "IN (%s(5, 'z1'), (12, 'z3'))"
1092 % ("VALUES " if config.db.dialect.tuple_in_values else ""),
1093 () if config.db.dialect.positional else {},
1094 )
1095 )
1096
1097
1098class ExpandingBoundInTest(fixtures.TablesTest):
1099 __backend__ = True
1100
1101 @classmethod
1102 def define_tables(cls, metadata):
1103 Table(
1104 "some_table",
1105 metadata,
1106 Column("id", Integer, primary_key=True),
1107 Column("x", Integer),
1108 Column("y", Integer),
1109 Column("z", String(50)),
1110 )
1111
1112 @classmethod
1113 def insert_data(cls, connection):
1114 connection.execute(
1115 cls.tables.some_table.insert(),
1116 [
1117 {"id": 1, "x": 1, "y": 2, "z": "z1"},
1118 {"id": 2, "x": 2, "y": 3, "z": "z2"},
1119 {"id": 3, "x": 3, "y": 4, "z": "z3"},
1120 {"id": 4, "x": 4, "y": 5, "z": "z4"},
1121 ],
1122 )
1123
1124 def _assert_result(self, select, result, params=()):
1125 with config.db.connect() as conn:
1126 eq_(conn.execute(select, params).fetchall(), result)
1127
1128 def test_multiple_empty_sets_bindparam(self):
1129 # test that any anonymous aliasing used by the dialect
1130 # is fine with duplicates
1131 table = self.tables.some_table
1132 stmt = (
1133 select(table.c.id)
1134 .where(table.c.x.in_(bindparam("q")))
1135 .where(table.c.y.in_(bindparam("p")))
1136 .order_by(table.c.id)
1137 )
1138 self._assert_result(stmt, [], params={"q": [], "p": []})
1139
1140 def test_multiple_empty_sets_direct(self):
1141 # test that any anonymous aliasing used by the dialect
1142 # is fine with duplicates
1143 table = self.tables.some_table
1144 stmt = (
1145 select(table.c.id)
1146 .where(table.c.x.in_([]))
1147 .where(table.c.y.in_([]))
1148 .order_by(table.c.id)
1149 )
1150 self._assert_result(stmt, [])
1151
1152 @testing.requires.tuple_in_w_empty
1153 def test_empty_heterogeneous_tuples_bindparam(self):
1154 table = self.tables.some_table
1155 stmt = (
1156 select(table.c.id)
1157 .where(tuple_(table.c.x, table.c.z).in_(bindparam("q")))
1158 .order_by(table.c.id)
1159 )
1160 self._assert_result(stmt, [], params={"q": []})
1161
1162 @testing.requires.tuple_in_w_empty
1163 def test_empty_heterogeneous_tuples_direct(self):
1164 table = self.tables.some_table
1165
1166 def go(val, expected):
1167 stmt = (
1168 select(table.c.id)
1169 .where(tuple_(table.c.x, table.c.z).in_(val))
1170 .order_by(table.c.id)
1171 )
1172 self._assert_result(stmt, expected)
1173
1174 go([], [])
1175 go([(2, "z2"), (3, "z3"), (4, "z4")], [(2,), (3,), (4,)])
1176 go([], [])
1177
1178 @testing.requires.tuple_in_w_empty
1179 def test_empty_homogeneous_tuples_bindparam(self):
1180 table = self.tables.some_table
1181 stmt = (
1182 select(table.c.id)
1183 .where(tuple_(table.c.x, table.c.y).in_(bindparam("q")))
1184 .order_by(table.c.id)
1185 )
1186 self._assert_result(stmt, [], params={"q": []})
1187
1188 @testing.requires.tuple_in_w_empty
1189 def test_empty_homogeneous_tuples_direct(self):
1190 table = self.tables.some_table
1191
1192 def go(val, expected):
1193 stmt = (
1194 select(table.c.id)
1195 .where(tuple_(table.c.x, table.c.y).in_(val))
1196 .order_by(table.c.id)
1197 )
1198 self._assert_result(stmt, expected)
1199
1200 go([], [])