1from __future__ import annotations
2
3import copy
4import datetime
5import functools
6import inspect
7from collections import defaultdict
8from decimal import Decimal
9from functools import cached_property
10from types import NoneType
11from typing import TYPE_CHECKING, Any, Protocol, Self, cast, runtime_checkable
12from uuid import UUID
13
14import psycopg
15
16from plain.postgres import fields
17from plain.postgres.constants import LOOKUP_SEP
18from plain.postgres.dialect import (
19 CURRENT_ROW,
20 FOLLOWING,
21 PRECEDING,
22 UNBOUNDED_FOLLOWING,
23 UNBOUNDED_PRECEDING,
24 combine_expression,
25 quote_name,
26 subtract_temporals,
27 window_frame_range_start_end,
28 window_frame_rows_start_end,
29)
30from plain.postgres.exceptions import EmptyResultSet, FieldError, FullResultSet
31from plain.postgres.query_utils import Q
32from plain.utils.deconstruct import deconstructible
33from plain.utils.hashable import make_hashable
34
35if TYPE_CHECKING:
36 from collections.abc import Callable, Iterable, Sequence
37
38 from plain.postgres.connection import DatabaseConnection
39 from plain.postgres.fields import Field
40 from plain.postgres.lookups import Lookup, Transform
41 from plain.postgres.query import QuerySet
42 from plain.postgres.sql.compiler import SQLCompilable, SQLCompiler
43 from plain.postgres.sql.query import Query
44
45__all__ = [
46 # Core expression classes
47 "F",
48 "Value",
49 "Case",
50 "When",
51 "Subquery",
52 "Exists",
53 "OuterRef",
54 "Window",
55 "ExpressionWrapper",
56 "RawSQL",
57 "OrderBy",
58 # Base classes (for extension)
59 "Func",
60 "Expression",
61 "Combinable",
62 # Window frame specs
63 "RowRange",
64 "ValueRange",
65]
66
67
68@runtime_checkable
69class ResolvableExpression(Protocol):
70 """Protocol for expressions that can be resolved in query context."""
71
72 def resolve_expression(
73 self,
74 query: Any = None,
75 allow_joins: bool = True,
76 reuse: Any = None,
77 summarize: bool = False,
78 for_save: bool = False,
79 ) -> Any: ...
80
81
82@runtime_checkable
83class ReplaceableExpression(Protocol):
84 """Protocol for expressions that support expression replacement."""
85
86 def replace_expressions(self, replacements: dict[Any, Any]) -> Self: ...
87
88
89class Combinable:
90 """
91 Provide the ability to combine one or two objects with
92 some connector. For example F('foo') + F('bar').
93 """
94
95 # Arithmetic connectors
96 ADD = "+"
97 SUB = "-"
98 MUL = "*"
99 DIV = "/"
100 POW = "^"
101 # The following is a quoted % operator - it is quoted because it can be
102 # used in strings that also have parameter substitution.
103 MOD = "%%"
104
105 # Bitwise operators - note that these are generated by .bitand()
106 # and .bitor(), the '&' and '|' are reserved for boolean operator
107 # usage.
108 BITAND = "&"
109 BITOR = "|"
110 BITLEFTSHIFT = "<<"
111 BITRIGHTSHIFT = ">>"
112 BITXOR = "#"
113
114 def _combine(
115 self, other: Any, connector: str, reversed: bool
116 ) -> CombinedExpression:
117 if not isinstance(other, ResolvableExpression):
118 # everything must be resolvable to an expression
119 other = Value(other)
120
121 if reversed:
122 return CombinedExpression(other, connector, self)
123 return CombinedExpression(self, connector, other)
124
125 #############
126 # OPERATORS #
127 #############
128
129 def __neg__(self) -> CombinedExpression:
130 return self._combine(-1, self.MUL, False)
131
132 def __add__(self, other: Any) -> CombinedExpression:
133 return self._combine(other, self.ADD, False)
134
135 def __sub__(self, other: Any) -> CombinedExpression:
136 return self._combine(other, self.SUB, False)
137
138 def __mul__(self, other: Any) -> CombinedExpression:
139 return self._combine(other, self.MUL, False)
140
141 def __truediv__(self, other: Any) -> CombinedExpression:
142 return self._combine(other, self.DIV, False)
143
144 def __mod__(self, other: Any) -> CombinedExpression:
145 return self._combine(other, self.MOD, False)
146
147 def __pow__(self, other: Any) -> CombinedExpression:
148 return self._combine(other, self.POW, False)
149
150 def __and__(self, other: Any) -> Q:
151 if getattr(self, "conditional", False) and getattr(other, "conditional", False):
152 return Q(self) & Q(other)
153 raise NotImplementedError(
154 "Use .bitand(), .bitor(), and .bitxor() for bitwise logical operations."
155 )
156
157 def bitand(self, other: Any) -> CombinedExpression:
158 return self._combine(other, self.BITAND, False)
159
160 def bitleftshift(self, other: Any) -> CombinedExpression:
161 return self._combine(other, self.BITLEFTSHIFT, False)
162
163 def bitrightshift(self, other: Any) -> CombinedExpression:
164 return self._combine(other, self.BITRIGHTSHIFT, False)
165
166 def __xor__(self, other: Any) -> None:
167 raise NotImplementedError(
168 "Use .bitand(), .bitor(), and .bitxor() for bitwise logical operations."
169 )
170
171 def bitxor(self, other: Any) -> CombinedExpression:
172 return self._combine(other, self.BITXOR, False)
173
174 def __or__(self, other: Any) -> Q:
175 if getattr(self, "conditional", False) and getattr(other, "conditional", False):
176 return Q(self) | Q(other)
177 raise NotImplementedError(
178 "Use .bitand(), .bitor(), and .bitxor() for bitwise logical operations."
179 )
180
181 def bitor(self, other: Any) -> CombinedExpression:
182 return self._combine(other, self.BITOR, False)
183
184 def __radd__(self, other: Any) -> CombinedExpression:
185 return self._combine(other, self.ADD, True)
186
187 def __rsub__(self, other: Any) -> CombinedExpression:
188 return self._combine(other, self.SUB, True)
189
190 def __rmul__(self, other: Any) -> CombinedExpression:
191 return self._combine(other, self.MUL, True)
192
193 def __rtruediv__(self, other: Any) -> CombinedExpression:
194 return self._combine(other, self.DIV, True)
195
196 def __rmod__(self, other: Any) -> CombinedExpression:
197 return self._combine(other, self.MOD, True)
198
199 def __rpow__(self, other: Any) -> CombinedExpression:
200 return self._combine(other, self.POW, True)
201
202 def __rand__(self, other: Any) -> None:
203 raise NotImplementedError(
204 "Use .bitand(), .bitor(), and .bitxor() for bitwise logical operations."
205 )
206
207 def __ror__(self, other: Any) -> None:
208 raise NotImplementedError(
209 "Use .bitand(), .bitor(), and .bitxor() for bitwise logical operations."
210 )
211
212 def __rxor__(self, other: Any) -> None:
213 raise NotImplementedError(
214 "Use .bitand(), .bitor(), and .bitxor() for bitwise logical operations."
215 )
216
217 def __invert__(self) -> NegatedExpression:
218 return NegatedExpression(self)
219
220
221class BaseExpression:
222 """Base class for all query expressions."""
223
224 empty_result_set_value = NotImplemented
225 # aggregate specific fields
226 is_summary = False
227 _output_field_resolved_to_none = False
228 # Can the expression be used in a WHERE clause?
229 filterable = True
230 # Can the expression can be used as a source expression in Window?
231 window_compatible = False
232
233 def __init__(self, output_field: Field | None = None):
234 if output_field is not None:
235 self.output_field = output_field
236
237 def __getstate__(self) -> dict[str, Any]:
238 state = self.__dict__.copy()
239 state.pop("convert_value", None)
240 return state
241
242 def get_db_converters(
243 self, connection: DatabaseConnection
244 ) -> list[Callable[..., Any]]:
245 converters = []
246 if self.convert_value is not self._convert_value_noop:
247 converters.append(self.convert_value)
248 converters.extend(self.output_field.get_db_converters(connection))
249 return converters
250
251 def get_source_expressions(self) -> list[Any]:
252 return []
253
254 def set_source_expressions(self, exprs: Sequence[Any]) -> None:
255 assert not exprs
256
257 def _parse_expressions(self, *expressions: Any) -> list[Any]:
258 return [
259 arg
260 if isinstance(arg, ResolvableExpression)
261 else (F(arg) if isinstance(arg, str) else Value(arg))
262 for arg in expressions
263 ]
264
265 def as_sql(
266 self, compiler: SQLCompiler, connection: DatabaseConnection
267 ) -> tuple[str, Sequence[Any]]:
268 """
269 Return a (sql, params) tuple to be included in the current query.
270
271 Arguments:
272 * compiler: the query compiler responsible for generating the query.
273 Must have a compile method, returning a (sql, [params]) tuple.
274 Calling compiler(value) will return a quoted `value`.
275
276 * connection: the database connection used for the current query.
277
278 Return: (sql, params)
279 Where `sql` is a string containing ordered sql parameters to be
280 replaced with the elements of the list `params`.
281 """
282 raise NotImplementedError("Subclasses must implement as_sql()")
283
284 @cached_property
285 def contains_aggregate(self) -> bool:
286 return any(
287 expr and expr.contains_aggregate for expr in self.get_source_expressions()
288 )
289
290 @cached_property
291 def contains_over_clause(self) -> bool:
292 return any(
293 expr and expr.contains_over_clause for expr in self.get_source_expressions()
294 )
295
296 @cached_property
297 def contains_column_references(self) -> bool:
298 return any(
299 expr and expr.contains_column_references
300 for expr in self.get_source_expressions()
301 )
302
303 def resolve_expression(
304 self,
305 query: Any = None,
306 allow_joins: bool = True,
307 reuse: Any = None,
308 summarize: bool = False,
309 for_save: bool = False,
310 ) -> Self:
311 """
312 Provide the chance to do any preprocessing or validation before being
313 added to the query.
314
315 Arguments:
316 * query: the backend query implementation
317 * allow_joins: boolean allowing or denying use of joins
318 in this query
319 * reuse: a set of reusable joins for multijoins
320 * summarize: a terminal aggregate clause
321 * for_save: whether this expression about to be used in a save or update
322
323 Return: an Expression to be added to the query.
324 """
325 c = self.copy()
326 c.is_summary = summarize
327 c.set_source_expressions(
328 [
329 expr.resolve_expression(query, allow_joins, reuse, summarize)
330 if expr
331 else None
332 for expr in c.get_source_expressions()
333 ]
334 )
335 return c
336
337 @property
338 def conditional(self) -> bool:
339 output_field = getattr(self, "output_field", None)
340 return isinstance(output_field, fields.BooleanField)
341
342 @property
343 def field(self) -> Field:
344 return self.output_field
345
346 @cached_property
347 def output_field(self) -> Field:
348 """Return the output type of this expressions."""
349 output_field = self._resolve_output_field()
350 if output_field is None:
351 self._output_field_resolved_to_none = True
352 raise FieldError("Cannot resolve expression type, unknown output_field")
353 return output_field
354
355 @cached_property
356 def _output_field_or_none(self) -> Field | None:
357 """
358 Return the output field of this expression, or None if
359 _resolve_output_field() didn't return an output type.
360 """
361 try:
362 return self.output_field
363 except FieldError:
364 if not self._output_field_resolved_to_none:
365 raise
366 return None
367
368 def _resolve_output_field(self) -> Field | None:
369 """
370 Attempt to infer the output type of the expression.
371
372 As a guess, if the output fields of all source fields match then simply
373 infer the same type here.
374
375 If a source's output field resolves to None, exclude it from this check.
376 If all sources are None, then an error is raised higher up the stack in
377 the output_field property.
378 """
379 # This guess is mostly a bad idea, but there is quite a lot of code
380 # (especially 3rd party Func subclasses) that depend on it, we'd need a
381 # deprecation path to fix it.
382 sources_iter = (
383 source for source in self.get_source_fields() if source is not None
384 )
385 for output_field in sources_iter:
386 for source in sources_iter:
387 if not isinstance(output_field, source.__class__):
388 raise FieldError(
389 f"Expression contains mixed types: {output_field.__class__.__name__}, {source.__class__.__name__}. You must "
390 "set output_field."
391 )
392 return output_field
393 return None
394
395 @staticmethod
396 def _convert_value_noop(
397 value: Any, expression: Any, connection: DatabaseConnection
398 ) -> Any:
399 return value
400
401 @cached_property
402 def convert_value(self) -> Callable[[Any, Any, Any], Any]:
403 """
404 Expressions provide their own converters because users have the option
405 of manually specifying the output_field which may be a different type
406 from the one the database returns.
407 """
408 field = self.output_field
409 if isinstance(field, fields.FloatField):
410 return (
411 lambda value, expression, connection: None
412 if value is None
413 else float(value)
414 )
415 elif isinstance(field, fields.IntegerField | fields.PrimaryKeyField):
416 return (
417 lambda value, expression, connection: None
418 if value is None
419 else int(value)
420 )
421 elif isinstance(field, fields.DecimalField):
422 return (
423 lambda value, expression, connection: None
424 if value is None
425 else Decimal(value)
426 )
427 return self._convert_value_noop
428
429 def get_lookup(self, lookup: str) -> type[Lookup] | None:
430 return self.output_field.get_lookup(lookup)
431
432 def get_transform(self, name: str) -> Callable[..., Transform] | None:
433 return self.output_field.get_transform(name)
434
435 def relabeled_clone(self, change_map: dict[str, str]) -> Self:
436 clone = self.copy()
437 clone.set_source_expressions(
438 [
439 e.relabeled_clone(change_map) if e is not None else None
440 for e in self.get_source_expressions()
441 ]
442 )
443 return clone
444
445 def replace_expressions(self, replacements: dict[BaseExpression, Any]) -> Self:
446 if replacement := replacements.get(self):
447 return replacement
448 clone = self.copy()
449 source_expressions = clone.get_source_expressions()
450 clone.set_source_expressions(
451 [
452 expr.replace_expressions(replacements) if expr else None
453 for expr in source_expressions
454 ]
455 )
456 return clone
457
458 def get_refs(self) -> set[str]:
459 refs = set()
460 for expr in self.get_source_expressions():
461 refs |= expr.get_refs()
462 return refs
463
464 def copy(self) -> Self:
465 return copy.copy(self)
466
467 def prefix_references(self, prefix: str) -> Self:
468 clone = self.copy()
469 clone.set_source_expressions(
470 [
471 F(f"{prefix}{expr.name}")
472 if isinstance(expr, F)
473 else expr.prefix_references(prefix)
474 for expr in self.get_source_expressions()
475 ]
476 )
477 return clone
478
479 def get_group_by_cols(self) -> list[BaseExpression]:
480 if not self.contains_aggregate:
481 return [self]
482 cols: list[BaseExpression] = []
483 for source in self.get_source_expressions():
484 cols.extend(source.get_group_by_cols())
485 return cols
486
487 def get_source_fields(self) -> list[Field | None]:
488 """Return the underlying field types used by this aggregate."""
489 return [e._output_field_or_none for e in self.get_source_expressions()]
490
491 def asc(self, **kwargs: Any) -> OrderBy:
492 return OrderBy(self, **kwargs)
493
494 def desc(self, **kwargs: Any) -> OrderBy:
495 return OrderBy(self, descending=True, **kwargs)
496
497 def reverse_ordering(self) -> Self:
498 return self
499
500 def flatten(self) -> Iterable[Any]:
501 """
502 Recursively yield this expression and all subexpressions, in
503 depth-first order.
504 """
505 yield self
506 for expr in self.get_source_expressions():
507 if expr:
508 if hasattr(expr, "flatten"):
509 yield from expr.flatten()
510 else:
511 yield expr
512
513 def select_format(
514 self, compiler: SQLCompiler, sql: str, params: Sequence[Any]
515 ) -> tuple[str, Sequence[Any]]:
516 """Custom format for select clauses."""
517 if output_field := getattr(self, "output_field", None):
518 if select_format := getattr(output_field, "select_format", None):
519 return select_format(compiler, sql, params)
520 return sql, params
521
522
523@deconstructible
524class Expression(BaseExpression, Combinable):
525 """An expression that can be combined with other expressions."""
526
527 # Set by @deconstructible decorator in __new__
528 _constructor_args: tuple[tuple[Any, ...], dict[str, Any]]
529
530 @cached_property
531 def identity(self) -> tuple[Any, ...]:
532 constructor_signature = inspect.signature(self.__init__)
533 args, kwargs = self._constructor_args
534 signature = constructor_signature.bind_partial(*args, **kwargs)
535 signature.apply_defaults()
536 arguments = signature.arguments.items()
537 identity: list[Any] = [self.__class__]
538 for arg, value in arguments:
539 if isinstance(value, fields.Field):
540 if value.name and value.model:
541 value = (value.model.model_options.label, value.name)
542 else:
543 value = type(value)
544 else:
545 value = make_hashable(value)
546 identity.append((arg, value))
547 return tuple(identity)
548
549 def __eq__(self, other: object) -> bool:
550 if not isinstance(other, Expression):
551 return NotImplemented
552 return other.identity == self.identity
553
554 def __hash__(self) -> int:
555 return hash(self.identity)
556
557
558# Type inference for CombinedExpression.output_field.
559# Missing items will result in FieldError, by design.
560#
561# The current approach for NULL is based on lowest common denominator behavior
562# i.e. if one of the supported databases is raising an error (rather than
563# return NULL) for `val <op> NULL`, then Plain raises FieldError.
564
565_connector_combinations = [
566 # Numeric operations - operands of same type.
567 {
568 connector: [
569 (fields.IntegerField, fields.IntegerField, fields.IntegerField),
570 (fields.FloatField, fields.FloatField, fields.FloatField),
571 (fields.DecimalField, fields.DecimalField, fields.DecimalField),
572 ]
573 for connector in (
574 Combinable.ADD,
575 Combinable.SUB,
576 Combinable.MUL,
577 Combinable.DIV,
578 Combinable.MOD,
579 Combinable.POW,
580 )
581 },
582 # Numeric operations - operands of different type.
583 {
584 connector: [
585 (fields.IntegerField, fields.DecimalField, fields.DecimalField),
586 (fields.DecimalField, fields.IntegerField, fields.DecimalField),
587 (fields.IntegerField, fields.FloatField, fields.FloatField),
588 (fields.FloatField, fields.IntegerField, fields.FloatField),
589 ]
590 for connector in (
591 Combinable.ADD,
592 Combinable.SUB,
593 Combinable.MUL,
594 Combinable.DIV,
595 Combinable.MOD,
596 )
597 },
598 # Bitwise operators.
599 {
600 connector: [
601 (fields.IntegerField, fields.IntegerField, fields.IntegerField),
602 ]
603 for connector in (
604 Combinable.BITAND,
605 Combinable.BITOR,
606 Combinable.BITLEFTSHIFT,
607 Combinable.BITRIGHTSHIFT,
608 Combinable.BITXOR,
609 )
610 },
611 # Numeric with NULL.
612 {
613 connector: [
614 (field_type, NoneType, field_type),
615 (NoneType, field_type, field_type),
616 ]
617 for connector in (
618 Combinable.ADD,
619 Combinable.SUB,
620 Combinable.MUL,
621 Combinable.DIV,
622 Combinable.MOD,
623 Combinable.POW,
624 )
625 for field_type in (fields.IntegerField, fields.DecimalField, fields.FloatField)
626 },
627 # Date/DateTimeField/DurationField/TimeField.
628 {
629 Combinable.ADD: [
630 # Date/DateTimeField.
631 (fields.DateField, fields.DurationField, fields.DateTimeField),
632 (fields.DateTimeField, fields.DurationField, fields.DateTimeField),
633 (fields.DurationField, fields.DateField, fields.DateTimeField),
634 (fields.DurationField, fields.DateTimeField, fields.DateTimeField),
635 # DurationField.
636 (fields.DurationField, fields.DurationField, fields.DurationField),
637 # TimeField.
638 (fields.TimeField, fields.DurationField, fields.TimeField),
639 (fields.DurationField, fields.TimeField, fields.TimeField),
640 ],
641 },
642 {
643 Combinable.SUB: [
644 # Date/DateTimeField.
645 (fields.DateField, fields.DurationField, fields.DateTimeField),
646 (fields.DateTimeField, fields.DurationField, fields.DateTimeField),
647 (fields.DateField, fields.DateField, fields.DurationField),
648 (fields.DateField, fields.DateTimeField, fields.DurationField),
649 (fields.DateTimeField, fields.DateField, fields.DurationField),
650 (fields.DateTimeField, fields.DateTimeField, fields.DurationField),
651 # DurationField.
652 (fields.DurationField, fields.DurationField, fields.DurationField),
653 # TimeField.
654 (fields.TimeField, fields.DurationField, fields.TimeField),
655 (fields.TimeField, fields.TimeField, fields.DurationField),
656 ],
657 },
658]
659
660_connector_combinators = defaultdict(list)
661
662
663def register_combinable_fields(
664 lhs: type[Field] | type[None],
665 connector: str,
666 rhs: type[Field] | type[None],
667 result: type[Field],
668) -> None:
669 """
670 Register combinable types:
671 lhs <connector> rhs -> result
672 e.g.
673 register_combinable_fields(
674 IntegerField, Combinable.ADD, FloatField, FloatField
675 )
676 """
677 _connector_combinators[connector].append((lhs, rhs, result))
678
679
680for d in _connector_combinations:
681 for connector, field_types in d.items():
682 for lhs, rhs, result in field_types:
683 register_combinable_fields(lhs, connector, rhs, result)
684
685
686@functools.lru_cache(maxsize=128)
687def _resolve_combined_type(
688 connector: str, lhs_type: type[Field], rhs_type: type[Field]
689) -> type[Field] | None:
690 combinators = _connector_combinators.get(connector, ())
691 for combinator_lhs_type, combinator_rhs_type, combined_type in combinators:
692 if issubclass(lhs_type, combinator_lhs_type) and issubclass(
693 rhs_type, combinator_rhs_type
694 ):
695 return combined_type
696 return None
697
698
699class CombinedExpression(Expression):
700 def __init__(
701 self, lhs: Any, connector: str, rhs: Any, output_field: Field | None = None
702 ):
703 super().__init__(output_field=output_field)
704 self.connector = connector
705 self.lhs = lhs
706 self.rhs = rhs
707
708 def __repr__(self) -> str:
709 return f"<{self.__class__.__name__}: {self}>"
710
711 def __str__(self) -> str:
712 return f"{self.lhs} {self.connector} {self.rhs}"
713
714 def get_source_expressions(self) -> list[Any]:
715 return [self.lhs, self.rhs]
716
717 def set_source_expressions(self, exprs: Sequence[Any]) -> None:
718 self.lhs, self.rhs = exprs
719
720 def _resolve_output_field(self) -> Field | None:
721 # We avoid using super() here for reasons given in
722 # Expression._resolve_output_field()
723 combined_type = _resolve_combined_type(
724 self.connector,
725 type(self.lhs._output_field_or_none),
726 type(self.rhs._output_field_or_none),
727 )
728 if combined_type is None:
729 raise FieldError(
730 f"Cannot infer type of {self.connector!r} expression involving these "
731 f"types: {self.lhs.output_field.__class__.__name__}, "
732 f"{self.rhs.output_field.__class__.__name__}. You must set "
733 f"output_field."
734 )
735 return combined_type()
736
737 def as_sql(
738 self, compiler: SQLCompiler, connection: DatabaseConnection
739 ) -> tuple[str, list[Any]]:
740 expressions = []
741 expression_params = []
742 sql, params = compiler.compile(self.lhs)
743 expressions.append(sql)
744 expression_params.extend(params)
745 sql, params = compiler.compile(self.rhs)
746 expressions.append(sql)
747 expression_params.extend(params)
748 # order of precedence
749 expression_wrapper = "(%s)"
750 sql = combine_expression(self.connector, expressions)
751 return expression_wrapper % sql, expression_params
752
753 def resolve_expression(
754 self,
755 query: Any = None,
756 allow_joins: bool = True,
757 reuse: Any = None,
758 summarize: bool = False,
759 for_save: bool = False,
760 ) -> CombinedExpression | TemporalSubtraction:
761 lhs = self.lhs.resolve_expression(
762 query, allow_joins, reuse, summarize, for_save
763 )
764 rhs = self.rhs.resolve_expression(
765 query, allow_joins, reuse, summarize, for_save
766 )
767 if not isinstance(self, TemporalSubtraction):
768 try:
769 lhs_field = lhs.output_field
770 except (AttributeError, FieldError):
771 lhs_field = None
772 try:
773 rhs_field = rhs.output_field
774 except (AttributeError, FieldError):
775 rhs_field = None
776 is_temporal = isinstance(
777 lhs_field, fields.DateField | fields.DateTimeField | fields.TimeField
778 )
779 same_type = (
780 lhs_field is not None
781 and rhs_field is not None
782 and type(lhs_field) is type(rhs_field)
783 )
784 if self.connector == self.SUB and is_temporal and same_type:
785 return TemporalSubtraction(self.lhs, self.rhs).resolve_expression(
786 query,
787 allow_joins,
788 reuse,
789 summarize,
790 for_save,
791 )
792 c = self.copy()
793 c.is_summary = summarize
794 c.lhs = lhs
795 c.rhs = rhs
796 return c
797
798
799class TemporalSubtraction(CombinedExpression):
800 output_field = fields.DurationField()
801
802 def __init__(self, lhs: Any, rhs: Any):
803 super().__init__(lhs, self.SUB, rhs)
804
805 def as_sql(
806 self, compiler: SQLCompiler, connection: DatabaseConnection
807 ) -> tuple[str, list[Any]]:
808 lhs = compiler.compile(self.lhs)
809 rhs = compiler.compile(self.rhs)
810 sql, params = subtract_temporals(self.lhs.output_field, lhs, rhs)
811 return sql, list(params)
812
813
814@deconstructible(path="plain.postgres.F")
815class F(Combinable):
816 """An object capable of resolving references to existing query objects."""
817
818 def __init__(self, name: str):
819 """
820 Arguments:
821 * name: the name of the field this expression references
822 """
823 self.name = name
824
825 def __repr__(self) -> str:
826 return f"{self.__class__.__name__}({self.name})"
827
828 def resolve_expression(
829 self,
830 query: Any = None,
831 allow_joins: bool = True,
832 reuse: Any = None,
833 summarize: bool = False,
834 for_save: bool = False,
835 ) -> Any:
836 return query.resolve_ref(self.name, allow_joins, reuse, summarize)
837
838 def replace_expressions(self, replacements: dict[Any, Any]) -> F:
839 return replacements.get(self, self)
840
841 def asc(self, **kwargs: Any) -> OrderBy:
842 return OrderBy(self, **kwargs)
843
844 def desc(self, **kwargs: Any) -> OrderBy:
845 return OrderBy(self, descending=True, **kwargs)
846
847 def __eq__(self, other: object) -> bool:
848 if not isinstance(other, F):
849 return NotImplemented
850 return self.__class__ == other.__class__ and self.name == other.name
851
852 def __hash__(self) -> int:
853 return hash(self.name)
854
855 def copy(self) -> Self:
856 return copy.copy(self)
857
858
859class ResolvedOuterRef(F):
860 """
861 An object that contains a reference to an outer query.
862
863 In this case, the reference to the outer query has been resolved because
864 the inner query has been used as a subquery.
865 """
866
867 contains_aggregate = False
868 contains_over_clause = False
869
870 def as_sql(self, *args: Any, **kwargs: Any) -> None:
871 raise ValueError(
872 "This queryset contains a reference to an outer query and may "
873 "only be used in a subquery."
874 )
875
876 def resolve_expression(self, *args: Any, **kwargs: Any) -> Any:
877 col = super().resolve_expression(*args, **kwargs)
878 if col.contains_over_clause:
879 raise psycopg.NotSupportedError(
880 f"Referencing outer query window expression is not supported: "
881 f"{self.name}."
882 )
883 # FIXME: Rename possibly_multivalued to multivalued and fix detection
884 # for non-multivalued JOINs (e.g. foreign key fields). This should take
885 # into account only many-to-many and one-to-many relationships.
886 col.possibly_multivalued = LOOKUP_SEP in self.name
887 return col
888
889 def relabeled_clone(self, relabels: dict[str, str]) -> ResolvedOuterRef:
890 return self
891
892 def get_group_by_cols(self) -> list[Any]:
893 return []
894
895
896class OuterRef(F):
897 contains_aggregate = False
898
899 def resolve_expression(self, *args: Any, **kwargs: Any) -> ResolvedOuterRef | F:
900 if isinstance(self.name, self.__class__):
901 return self.name
902 return ResolvedOuterRef(self.name)
903
904 def relabeled_clone(self, relabels: dict[str, str]) -> OuterRef:
905 return self
906
907
908@deconstructible(path="plain.postgres.expressions.Func")
909class Func(Expression):
910 """An SQL function call."""
911
912 function: str | None = None
913 template: str = "%(function)s(%(expressions)s)"
914 arg_joiner: str = ", "
915 arity: int | None = None # The number of arguments the function accepts.
916
917 def __init__(
918 self, *expressions: Any, output_field: Field | None = None, **extra: Any
919 ):
920 if self.arity is not None and len(expressions) != self.arity:
921 raise TypeError(
922 "'{}' takes exactly {} {} ({} given)".format(
923 self.__class__.__name__,
924 self.arity,
925 "argument" if self.arity == 1 else "arguments",
926 len(expressions),
927 )
928 )
929 super().__init__(output_field=output_field)
930 self.source_expressions: list[Any] = self._parse_expressions(*expressions)
931 self.extra = extra
932
933 def __repr__(self) -> str:
934 args = self.arg_joiner.join(str(arg) for arg in self.source_expressions)
935 extra = {**self.extra, **self._get_repr_options()}
936 if extra:
937 extra = ", ".join(
938 str(key) + "=" + str(val) for key, val in sorted(extra.items())
939 )
940 return f"{self.__class__.__name__}({args}, {extra})"
941 return f"{self.__class__.__name__}({args})"
942
943 def _get_repr_options(self) -> dict[str, Any]:
944 """Return a dict of extra __init__() options to include in the repr."""
945 return {}
946
947 def get_source_expressions(self) -> list[Any]:
948 return self.source_expressions
949
950 def set_source_expressions(self, exprs: Sequence[Any]) -> None:
951 self.source_expressions = list(exprs)
952
953 def resolve_expression(
954 self,
955 query: Any = None,
956 allow_joins: bool = True,
957 reuse: Any = None,
958 summarize: bool = False,
959 for_save: bool = False,
960 ) -> Self:
961 c = self.copy()
962 c.is_summary = summarize
963 for pos, arg in enumerate(c.source_expressions):
964 c.source_expressions[pos] = arg.resolve_expression(
965 query, allow_joins, reuse, summarize, for_save
966 )
967 return c
968
969 def as_sql(
970 self,
971 compiler: SQLCompiler,
972 connection: DatabaseConnection,
973 function: str | None = None,
974 template: str | None = None,
975 arg_joiner: str | None = None,
976 **extra_context: Any,
977 ) -> tuple[str, list[Any]]:
978 sql_parts = []
979 params = []
980 for arg in self.source_expressions:
981 try:
982 arg_sql, arg_params = compiler.compile(arg)
983 except EmptyResultSet:
984 empty_result_set_value = getattr(
985 arg, "empty_result_set_value", NotImplemented
986 )
987 if empty_result_set_value is NotImplemented:
988 raise
989 arg_sql, arg_params = compiler.compile(Value(empty_result_set_value))
990 except FullResultSet:
991 arg_sql, arg_params = compiler.compile(Value(True))
992 sql_parts.append(arg_sql)
993 params.extend(arg_params)
994 data = {**self.extra, **extra_context}
995 # Use the first supplied value in this order: the parameter to this
996 # method, a value supplied in __init__()'s **extra (the value in
997 # `data`), or the value defined on the class.
998 if function is not None:
999 data["function"] = function
1000 else:
1001 data.setdefault("function", self.function)
1002 # `data` is typed dict[str, Any], so the override values come back as
1003 # Any; they are always strings, so cast to keep the `or` fallback typed.
1004 resolved_template = template or cast(str, data.get("template", self.template))
1005 resolved_joiner = arg_joiner or cast(
1006 str, data.get("arg_joiner", self.arg_joiner)
1007 )
1008 data["expressions"] = data["field"] = resolved_joiner.join(sql_parts)
1009 return resolved_template % data, params
1010
1011 def copy(self) -> Self:
1012 clone = super().copy()
1013 clone.source_expressions = self.source_expressions[:]
1014 clone.extra = self.extra.copy()
1015 return clone
1016
1017
1018@deconstructible(path="plain.postgres.expressions.Value")
1019class Value(Expression):
1020 """Represent a wrapped value as a node within an expression."""
1021
1022 # Provide a default value for `for_save` in order to allow unresolved
1023 # instances to be compiled until a decision is taken in #25425.
1024 for_save = False
1025
1026 def __init__(self, value: Any, output_field: Field | None = None):
1027 """
1028 Arguments:
1029 * value: the value this expression represents. The value will be
1030 added into the sql parameter list and properly quoted.
1031
1032 * output_field: an instance of the model field type that this
1033 expression will return, such as IntegerField() or TextField().
1034 """
1035 super().__init__(output_field=output_field)
1036 self.value = value
1037
1038 def __repr__(self) -> str:
1039 return f"{self.__class__.__name__}({self.value!r})"
1040
1041 def as_sql(
1042 self, compiler: SQLCompiler, connection: DatabaseConnection
1043 ) -> tuple[str, list[Any]]:
1044 val = self.value
1045 output_field = self._output_field_or_none
1046 if output_field is not None:
1047 if self.for_save:
1048 val = output_field.get_db_prep_save(val, connection=connection)
1049 else:
1050 val = output_field.get_db_prep_value(val, connection=connection)
1051 if hasattr(output_field, "get_placeholder"):
1052 return output_field.get_placeholder(val, compiler, connection), [val] # ty: ignore[call-non-callable]
1053 if val is None:
1054 return "NULL", []
1055 return "%s", [val]
1056
1057 def resolve_expression(
1058 self,
1059 query: Any = None,
1060 allow_joins: bool = True,
1061 reuse: Any = None,
1062 summarize: bool = False,
1063 for_save: bool = False,
1064 ) -> Value:
1065 c = super().resolve_expression(query, allow_joins, reuse, summarize, for_save)
1066 c.for_save = for_save
1067 return c
1068
1069 def get_group_by_cols(self) -> list[Any]:
1070 return []
1071
1072 def _resolve_output_field(self) -> Field | None:
1073 if isinstance(self.value, str):
1074 return fields.TextField()
1075 if isinstance(self.value, bool):
1076 return fields.BooleanField()
1077 if isinstance(self.value, int):
1078 return fields.IntegerField()
1079 if isinstance(self.value, float):
1080 return fields.FloatField()
1081 if isinstance(self.value, datetime.datetime):
1082 return fields.DateTimeField()
1083 if isinstance(self.value, datetime.date):
1084 return fields.DateField()
1085 if isinstance(self.value, datetime.time):
1086 return fields.TimeField()
1087 if isinstance(self.value, datetime.timedelta):
1088 return fields.DurationField()
1089 if isinstance(self.value, Decimal):
1090 return fields.DecimalField()
1091 if isinstance(self.value, bytes):
1092 return fields.BinaryField()
1093 if isinstance(self.value, UUID):
1094 return fields.UUIDField()
1095
1096 @property
1097 def empty_result_set_value(self) -> Any:
1098 return self.value
1099
1100
1101class RawSQL(Expression):
1102 def __init__(
1103 self, sql: str, params: Sequence[Any], output_field: Field | None = None
1104 ):
1105 if output_field is None:
1106 output_field = fields.Field()
1107 self.sql, self.params = sql, params
1108 super().__init__(output_field=output_field)
1109
1110 def __repr__(self) -> str:
1111 return f"{self.__class__.__name__}({self.sql}, {self.params})"
1112
1113 def as_sql(
1114 self, compiler: SQLCompiler, connection: DatabaseConnection
1115 ) -> tuple[str, Sequence[Any]]:
1116 return f"({self.sql})", self.params
1117
1118 def get_group_by_cols(self) -> list[BaseExpression]:
1119 return [self]
1120
1121
1122class Star(Expression):
1123 def __repr__(self) -> str:
1124 return "'*'"
1125
1126 def as_sql(
1127 self, compiler: SQLCompiler, connection: DatabaseConnection
1128 ) -> tuple[str, list[Any]]:
1129 return "*", []
1130
1131
1132class Col(Expression):
1133 contains_column_references = True
1134 possibly_multivalued = False
1135
1136 def __init__(
1137 self, alias: str | None, target: Any, output_field: Field | None = None
1138 ):
1139 if output_field is None:
1140 output_field = target
1141 super().__init__(output_field=output_field)
1142 self.alias, self.target = alias, target
1143
1144 def __repr__(self) -> str:
1145 alias, target = self.alias, self.target
1146 identifiers = (alias, str(target)) if alias else (str(target),)
1147 return "{}({})".format(self.__class__.__name__, ", ".join(identifiers))
1148
1149 def as_sql(
1150 self, compiler: SQLCompiler, connection: DatabaseConnection
1151 ) -> tuple[str, list[Any]]:
1152 alias, column = self.alias, self.target.column
1153 identifiers = (alias, column) if alias else (column,)
1154 sql = ".".join(map(compiler.quote_name_unless_alias, identifiers))
1155 return sql, []
1156
1157 def relabeled_clone(self, change_map: dict[str, str]) -> Self:
1158 if self.alias is None:
1159 return self
1160 return self.__class__(
1161 change_map.get(self.alias, self.alias), self.target, self.output_field
1162 )
1163
1164 def get_group_by_cols(self) -> list[BaseExpression]:
1165 return [self]
1166
1167 def get_db_converters(
1168 self, connection: DatabaseConnection
1169 ) -> list[Callable[..., Any]]:
1170 if self.target == self.output_field:
1171 return self.output_field.get_db_converters(connection)
1172 return self.output_field.get_db_converters(
1173 connection
1174 ) + self.target.get_db_converters(connection)
1175
1176
1177class Ref(Expression):
1178 """
1179 Reference to column alias of the query. For example, Ref('sum_cost') in
1180 qs.annotate(sum_cost=Sum('cost')) query.
1181 """
1182
1183 def __init__(self, refs: str, source: Any):
1184 super().__init__()
1185 self.refs, self.source = refs, source
1186
1187 def __repr__(self) -> str:
1188 return f"{self.__class__.__name__}({self.refs}, {self.source})"
1189
1190 def get_source_expressions(self) -> list[Any]:
1191 return [self.source]
1192
1193 def set_source_expressions(self, exprs: Sequence[Any]) -> None:
1194 (self.source,) = exprs
1195
1196 def resolve_expression(
1197 self,
1198 query: Any = None,
1199 allow_joins: bool = True,
1200 reuse: Any = None,
1201 summarize: bool = False,
1202 for_save: bool = False,
1203 ) -> Ref:
1204 # The sub-expression `source` has already been resolved, as this is
1205 # just a reference to the name of `source`.
1206 return self
1207
1208 def get_refs(self) -> set[str]:
1209 return {self.refs}
1210
1211 def relabeled_clone(self, change_map: dict[str, str]) -> Self:
1212 return self
1213
1214 def as_sql(
1215 self, compiler: SQLCompiler, connection: DatabaseConnection
1216 ) -> tuple[str, list[Any]]:
1217 return quote_name(self.refs), []
1218
1219 def get_group_by_cols(self) -> list[BaseExpression]:
1220 return [self]
1221
1222
1223class ExpressionList(Func):
1224 """
1225 An expression containing multiple expressions. Can be used to provide a
1226 list of expressions as an argument to another expression, like a partition
1227 clause.
1228 """
1229
1230 template = "%(expressions)s"
1231
1232 def __init__(self, *expressions: Any, **extra: Any):
1233 if not expressions:
1234 raise ValueError(
1235 f"{self.__class__.__name__} requires at least one expression."
1236 )
1237 super().__init__(*expressions, **extra)
1238
1239 def __str__(self) -> str:
1240 return self.arg_joiner.join(str(arg) for arg in self.source_expressions)
1241
1242
1243class OrderByList(Func):
1244 template = "ORDER BY %(expressions)s"
1245
1246 def __init__(self, *expressions: Any, **extra: Any):
1247 expressions_tuple = tuple(
1248 (
1249 OrderBy(F(expr[1:]), descending=True)
1250 if isinstance(expr, str) and expr[0] == "-"
1251 else expr
1252 )
1253 for expr in expressions
1254 )
1255 super().__init__(*expressions_tuple, **extra)
1256
1257 def as_sql(self, *args: Any, **kwargs: Any) -> tuple[str, list[Any]]:
1258 if not self.source_expressions:
1259 return "", []
1260 sql, params = super().as_sql(*args, **kwargs)
1261 return sql, list(params)
1262
1263 def get_group_by_cols(self) -> list[Any]:
1264 group_by_cols = []
1265 for order_by in self.get_source_expressions():
1266 group_by_cols.extend(order_by.get_group_by_cols())
1267 return group_by_cols
1268
1269
1270@deconstructible(path="plain.postgres.expressions.ExpressionWrapper")
1271class ExpressionWrapper(Expression):
1272 """
1273 An expression that can wrap another expression so that it can provide
1274 extra context to the inner expression, such as the output_field.
1275 """
1276
1277 def __init__(self, expression: Any, output_field: Field):
1278 super().__init__(output_field=output_field)
1279 self.expression = expression
1280
1281 def set_source_expressions(self, exprs: Sequence[Any]) -> None:
1282 self.expression = exprs[0]
1283
1284 def get_source_expressions(self) -> list[Any]:
1285 return [self.expression]
1286
1287 def get_group_by_cols(self) -> list[Any]:
1288 if isinstance(self.expression, Expression):
1289 expression = self.expression.copy()
1290 expression.output_field = self.output_field
1291 return expression.get_group_by_cols()
1292 # For non-expressions e.g. an SQL WHERE clause, the entire
1293 # `expression` must be included in the GROUP BY clause.
1294 return super().get_group_by_cols()
1295
1296 def as_sql(
1297 self, compiler: SQLCompiler, connection: DatabaseConnection
1298 ) -> tuple[str, Sequence[Any]]:
1299 return compiler.compile(self.expression)
1300
1301 def __repr__(self) -> str:
1302 return f"{self.__class__.__name__}({self.expression})"
1303
1304
1305class NegatedExpression(ExpressionWrapper):
1306 """The logical negation of a conditional expression."""
1307
1308 def __init__(self, expression: Any):
1309 super().__init__(expression, output_field=fields.BooleanField())
1310
1311 def __invert__(self) -> Any:
1312 return self.expression.copy()
1313
1314 def as_sql(
1315 self, compiler: SQLCompiler, connection: DatabaseConnection
1316 ) -> tuple[str, Sequence[Any]]:
1317 try:
1318 sql, params = super().as_sql(compiler, connection)
1319 except EmptyResultSet:
1320 return compiler.compile(Value(True))
1321 return f"NOT {sql}", params
1322
1323 def resolve_expression(
1324 self,
1325 query: Any = None,
1326 allow_joins: bool = True,
1327 reuse: Any = None,
1328 summarize: bool = False,
1329 for_save: bool = False,
1330 ) -> NegatedExpression:
1331 resolved = super().resolve_expression(
1332 query, allow_joins, reuse, summarize, for_save
1333 )
1334 if not getattr(resolved.expression, "conditional", False):
1335 raise TypeError("Cannot negate non-conditional expressions.")
1336 return resolved
1337
1338 def select_format(
1339 self, compiler: SQLCompiler, sql: str, params: Sequence[Any]
1340 ) -> tuple[str, Sequence[Any]]:
1341 # Boolean expressions work directly in SELECT
1342 return sql, params
1343
1344
1345@deconstructible(path="plain.postgres.expressions.When")
1346class When(Expression):
1347 template = "WHEN %(condition)s THEN %(result)s"
1348 # This isn't a complete conditional expression, must be used in Case().
1349 conditional = False
1350 condition: SQLCompilable
1351
1352 def __init__(
1353 self, condition: Q | Expression | None = None, then: Any = None, **lookups: Any
1354 ):
1355 lookups_dict: dict[str, Any] | None = lookups or None
1356 if lookups_dict:
1357 if condition is None:
1358 condition, lookups_dict = Q(**lookups_dict), None
1359 elif getattr(condition, "conditional", False):
1360 condition, lookups_dict = Q(condition, **lookups_dict), None
1361 if (
1362 condition is None
1363 or not getattr(condition, "conditional", False)
1364 or lookups_dict
1365 ):
1366 raise TypeError(
1367 "When() supports a Q object, a boolean expression, or lookups "
1368 "as a condition."
1369 )
1370 if isinstance(condition, Q) and not condition:
1371 raise ValueError("An empty Q() can't be used as a When() condition.")
1372 super().__init__(output_field=None)
1373 self.condition = condition # ty: ignore[invalid-assignment]
1374 self.result = self._parse_expressions(then)[0]
1375
1376 def __str__(self) -> str:
1377 return f"WHEN {self.condition!r} THEN {self.result!r}"
1378
1379 def __repr__(self) -> str:
1380 return f"<{self.__class__.__name__}: {self}>"
1381
1382 def get_source_expressions(self) -> list[Any]:
1383 return [self.condition, self.result]
1384
1385 def set_source_expressions(self, exprs: Sequence[Any]) -> None:
1386 self.condition, self.result = exprs
1387
1388 def get_source_fields(self) -> list[Field | None]:
1389 # We're only interested in the fields of the result expressions.
1390 return [self.result._output_field_or_none]
1391
1392 def resolve_expression(
1393 self,
1394 query: Any = None,
1395 allow_joins: bool = True,
1396 reuse: Any = None,
1397 summarize: bool = False,
1398 for_save: bool = False,
1399 ) -> When:
1400 c = self.copy()
1401 c.is_summary = summarize
1402 if isinstance(c.condition, ResolvableExpression):
1403 c.condition = c.condition.resolve_expression(
1404 query, allow_joins, reuse, summarize, False
1405 )
1406 c.result = c.result.resolve_expression(
1407 query, allow_joins, reuse, summarize, for_save
1408 )
1409 return c
1410
1411 def as_sql(
1412 self,
1413 compiler: SQLCompiler,
1414 connection: DatabaseConnection,
1415 template: str | None = None,
1416 **extra_context: Any,
1417 ) -> tuple[str, tuple[Any, ...]]:
1418 template_params = extra_context
1419 sql_params = []
1420 # After resolve_expression, condition is WhereNode | resolved Expression (both SQLCompilable)
1421 condition_sql, condition_params = compiler.compile(self.condition)
1422 template_params["condition"] = condition_sql
1423 result_sql, result_params = compiler.compile(self.result)
1424 template_params["result"] = result_sql
1425 template = template or self.template
1426 return template % template_params, (
1427 *sql_params,
1428 *condition_params,
1429 *result_params,
1430 )
1431
1432 def get_group_by_cols(self) -> list[Any]:
1433 # This is not a complete expression and cannot be used in GROUP BY.
1434 cols = []
1435 for source in self.get_source_expressions():
1436 cols.extend(source.get_group_by_cols())
1437 return cols
1438
1439
1440@deconstructible(path="plain.postgres.expressions.Case")
1441class Case(Expression):
1442 """
1443 An SQL searched CASE expression:
1444
1445 CASE
1446 WHEN n > 0
1447 THEN 'positive'
1448 WHEN n < 0
1449 THEN 'negative'
1450 ELSE 'zero'
1451 END
1452 """
1453
1454 template = "CASE %(cases)s ELSE %(default)s END"
1455 case_joiner = " "
1456
1457 def __init__(
1458 self,
1459 *cases: When,
1460 default: Any = None,
1461 output_field: Field | None = None,
1462 **extra: Any,
1463 ):
1464 if not all(isinstance(case, When) for case in cases):
1465 raise TypeError("Positional arguments must all be When objects.")
1466 super().__init__(output_field)
1467 self.cases = list(cases)
1468 self.default = self._parse_expressions(default)[0]
1469 self.extra = extra
1470
1471 def __str__(self) -> str:
1472 return "CASE {}, ELSE {!r}".format(
1473 ", ".join(str(c) for c in self.cases),
1474 self.default,
1475 )
1476
1477 def __repr__(self) -> str:
1478 return f"<{self.__class__.__name__}: {self}>"
1479
1480 def get_source_expressions(self) -> list[Any]:
1481 return self.cases + [self.default]
1482
1483 def set_source_expressions(self, exprs: Sequence[Any]) -> None:
1484 *self.cases, self.default = exprs
1485
1486 def resolve_expression(
1487 self,
1488 query: Any = None,
1489 allow_joins: bool = True,
1490 reuse: Any = None,
1491 summarize: bool = False,
1492 for_save: bool = False,
1493 ) -> Case:
1494 c = self.copy()
1495 c.is_summary = summarize
1496 for pos, case in enumerate(c.cases):
1497 c.cases[pos] = case.resolve_expression(
1498 query, allow_joins, reuse, summarize, for_save
1499 )
1500 c.default = c.default.resolve_expression(
1501 query, allow_joins, reuse, summarize, for_save
1502 )
1503 return c
1504
1505 def copy(self) -> Self:
1506 c = super().copy()
1507 c.cases = c.cases[:]
1508 return c
1509
1510 def as_sql(
1511 self,
1512 compiler: SQLCompiler,
1513 connection: DatabaseConnection,
1514 template: str | None = None,
1515 case_joiner: str | None = None,
1516 **extra_context: Any,
1517 ) -> tuple[str, list[Any]]:
1518 if not self.cases:
1519 sql, params = compiler.compile(self.default)
1520 return sql, list(params)
1521 template_params = {**self.extra, **extra_context}
1522 case_parts = []
1523 sql_params = []
1524 default_sql, default_params = compiler.compile(self.default)
1525 for case in self.cases:
1526 try:
1527 case_sql, case_params = compiler.compile(case)
1528 except EmptyResultSet:
1529 continue
1530 except FullResultSet:
1531 default_sql, default_params = compiler.compile(case.result)
1532 break
1533 case_parts.append(case_sql)
1534 sql_params.extend(case_params)
1535 if not case_parts:
1536 return default_sql, list(default_params)
1537 case_joiner = case_joiner or self.case_joiner
1538 template_params["cases"] = case_joiner.join(case_parts)
1539 template_params["default"] = default_sql
1540 sql_params.extend(default_params)
1541 template = template or template_params.get("template", self.template)
1542 sql = template % template_params
1543 if self._output_field_or_none is not None:
1544 sql = connection.unification_cast_sql(self.output_field) % sql
1545 return sql, sql_params
1546
1547 def get_group_by_cols(self) -> list[Any]:
1548 if not self.cases:
1549 return self.default.get_group_by_cols()
1550 return super().get_group_by_cols()
1551
1552
1553class Subquery(BaseExpression, Combinable):
1554 """
1555 An explicit subquery. It may contain OuterRef() references to the outer
1556 query which will be resolved when it is applied to that query.
1557 """
1558
1559 template = "(%(subquery)s)"
1560 contains_aggregate = False
1561 empty_result_set_value = None
1562
1563 def __init__(
1564 self,
1565 query: QuerySet[Any] | Query,
1566 output_field: Field | None = None,
1567 **extra: Any,
1568 ):
1569 # Import here to avoid circular import
1570 from plain.postgres.sql.query import Query
1571
1572 # Allow the usage of both QuerySet and sql.Query objects.
1573 if isinstance(query, Query):
1574 # It's already a Query object, use it directly
1575 sql_query = query
1576 else:
1577 # It's a QuerySet, extract the sql.Query
1578 sql_query = query.sql_query
1579 self.query = sql_query.clone()
1580 self.query.subquery = True
1581 self.extra = extra
1582 super().__init__(output_field)
1583
1584 def get_source_expressions(self) -> list[Any]:
1585 return [self.query]
1586
1587 def set_source_expressions(self, exprs: Sequence[Any]) -> None:
1588 self.query = exprs[0]
1589
1590 def _resolve_output_field(self) -> Field | None:
1591 return self.query.output_field
1592
1593 def copy(self) -> Self:
1594 clone = super().copy()
1595 clone.query = clone.query.clone()
1596 return clone
1597
1598 @property
1599 def external_aliases(self) -> dict[str, bool]:
1600 return self.query.external_aliases
1601
1602 def get_external_cols(self) -> list[Any]:
1603 return self.query.get_external_cols()
1604
1605 def as_sql(
1606 self,
1607 compiler: SQLCompiler,
1608 connection: DatabaseConnection,
1609 template: str | None = None,
1610 **extra_context: Any,
1611 ) -> tuple[str, tuple[Any, ...]]:
1612 template_params = {**self.extra, **extra_context}
1613 subquery_sql, sql_params = self.query.as_sql(compiler, connection)
1614 template_params["subquery"] = subquery_sql[1:-1]
1615
1616 template = template or template_params.get("template", self.template)
1617 sql = template % template_params
1618 return sql, sql_params
1619
1620 def get_group_by_cols(self) -> list[Any]:
1621 return self.query.get_group_by_cols(wrapper=self)
1622
1623
1624class Exists(Subquery):
1625 template = "EXISTS(%(subquery)s)"
1626 output_field = fields.BooleanField()
1627 empty_result_set_value = False
1628
1629 def __init__(self, query: QuerySet[Any] | Query, **kwargs: Any):
1630 super().__init__(query, **kwargs)
1631 self.query = self.query.exists()
1632
1633 def select_format(
1634 self, compiler: SQLCompiler, sql: str, params: Sequence[Any]
1635 ) -> tuple[str, Sequence[Any]]:
1636 # Boolean expressions work directly in SELECT
1637 return sql, params
1638
1639
1640@deconstructible(path="plain.postgres.expressions.OrderBy")
1641class OrderBy(Expression):
1642 template = "%(expression)s %(ordering)s"
1643 conditional = False
1644
1645 def __init__(
1646 self,
1647 expression: Any,
1648 descending: bool = False,
1649 nulls_first: bool | None = None,
1650 nulls_last: bool | None = None,
1651 ):
1652 if nulls_first and nulls_last:
1653 raise ValueError("nulls_first and nulls_last are mutually exclusive")
1654 if nulls_first is False or nulls_last is False:
1655 raise ValueError("nulls_first and nulls_last values must be True or None.")
1656 self.nulls_first = nulls_first
1657 self.nulls_last = nulls_last
1658 self.descending = descending
1659 if not isinstance(expression, ResolvableExpression):
1660 raise ValueError("expression must be an expression type")
1661 self.expression = expression
1662
1663 def __repr__(self) -> str:
1664 return f"{self.__class__.__name__}({self.expression}, descending={self.descending})"
1665
1666 def set_source_expressions(self, exprs: Sequence[Any]) -> None:
1667 self.expression = exprs[0]
1668
1669 def get_source_expressions(self) -> list[Any]:
1670 return [self.expression]
1671
1672 def as_sql(
1673 self,
1674 compiler: SQLCompiler,
1675 connection: DatabaseConnection,
1676 template: str | None = None,
1677 **extra_context: Any,
1678 ) -> tuple[str, tuple[Any, ...]]:
1679 template = template or self.template
1680 # Handle NULLS FIRST/LAST modifiers
1681 if self.nulls_last:
1682 template = f"{template} NULLS LAST"
1683 elif self.nulls_first:
1684 template = f"{template} NULLS FIRST"
1685 expression_sql, params = compiler.compile(self.expression)
1686 placeholders = {
1687 "expression": expression_sql,
1688 "ordering": "DESC" if self.descending else "ASC",
1689 **extra_context,
1690 }
1691 params *= template.count("%(expression)s")
1692 return (template % placeholders).rstrip(), params
1693
1694 def get_group_by_cols(self) -> list[Any]:
1695 cols = []
1696 for source in self.get_source_expressions():
1697 cols.extend(source.get_group_by_cols())
1698 return cols
1699
1700 def reverse_ordering(self) -> OrderBy:
1701 self.descending = not self.descending
1702 if self.nulls_first:
1703 self.nulls_last = True
1704 self.nulls_first = None
1705 elif self.nulls_last:
1706 self.nulls_first = True
1707 self.nulls_last = None
1708 return self
1709
1710 def asc(self) -> None: # ty: ignore[invalid-method-override]
1711 self.descending = False
1712
1713 def desc(self) -> None: # ty: ignore[invalid-method-override]
1714 self.descending = True
1715
1716
1717class Window(Expression):
1718 template = "%(expression)s OVER (%(window)s)"
1719 # Although the main expression may either be an aggregate or an
1720 # expression with an aggregate function, the GROUP BY that will
1721 # be introduced in the query as a result is not desired.
1722 contains_aggregate = False
1723 contains_over_clause = True
1724 partition_by: ExpressionList | None
1725 order_by: OrderByList | None
1726
1727 def __init__(
1728 self,
1729 expression: Any,
1730 partition_by: Any = None,
1731 order_by: Any = None,
1732 frame: Any = None,
1733 output_field: Field | None = None,
1734 ):
1735 self.partition_by = partition_by
1736 self.order_by = order_by
1737 self.frame = frame
1738
1739 if not getattr(expression, "window_compatible", False):
1740 raise ValueError(
1741 f"Expression '{expression.__class__.__name__}' isn't compatible with OVER clauses."
1742 )
1743
1744 if self.partition_by is not None:
1745 partition_by_values = (
1746 self.partition_by
1747 if isinstance(self.partition_by, tuple | list)
1748 else (self.partition_by,)
1749 )
1750 self.partition_by = ExpressionList(*partition_by_values)
1751
1752 if self.order_by is not None:
1753 if isinstance(self.order_by, list | tuple):
1754 self.order_by = OrderByList(*self.order_by)
1755 elif isinstance(self.order_by, BaseExpression | str):
1756 self.order_by = OrderByList(self.order_by)
1757 else:
1758 raise ValueError(
1759 "Window.order_by must be either a string reference to a "
1760 "field, an expression, or a list or tuple of them."
1761 )
1762 super().__init__(output_field=output_field)
1763 self.source_expression = self._parse_expressions(expression)[0]
1764
1765 def _resolve_output_field(self) -> Field | None:
1766 return self.source_expression.output_field
1767
1768 def get_source_expressions(self) -> list[Any]:
1769 return [self.source_expression, self.partition_by, self.order_by, self.frame]
1770
1771 def set_source_expressions(self, exprs: Sequence[Any]) -> None:
1772 self.source_expression, self.partition_by, self.order_by, self.frame = exprs
1773
1774 def as_sql(
1775 self,
1776 compiler: SQLCompiler,
1777 connection: DatabaseConnection,
1778 template: str | None = None,
1779 ) -> tuple[str, tuple[Any, ...]]:
1780 expr_sql, params = compiler.compile(self.source_expression)
1781 window_sql, window_params = [], ()
1782
1783 if self.partition_by is not None:
1784 sql_expr, sql_params = self.partition_by.as_sql(
1785 compiler=compiler,
1786 connection=connection,
1787 template="PARTITION BY %(expressions)s",
1788 )
1789 window_sql.append(sql_expr)
1790 window_params += tuple(sql_params)
1791
1792 if self.order_by is not None:
1793 order_sql, order_params = compiler.compile(self.order_by)
1794 window_sql.append(order_sql)
1795 window_params += tuple(order_params)
1796
1797 if self.frame:
1798 frame_sql, frame_params = compiler.compile(self.frame)
1799 window_sql.append(frame_sql)
1800 window_params += tuple(frame_params)
1801
1802 template = template or self.template
1803
1804 return (
1805 template % {"expression": expr_sql, "window": " ".join(window_sql).strip()},
1806 (*params, *window_params),
1807 )
1808
1809 def __str__(self) -> str:
1810 return "{} OVER ({}{}{})".format(
1811 str(self.source_expression),
1812 "PARTITION BY " + str(self.partition_by) if self.partition_by else "",
1813 str(self.order_by or ""),
1814 str(self.frame or ""),
1815 )
1816
1817 def __repr__(self) -> str:
1818 return f"<{self.__class__.__name__}: {self}>"
1819
1820 def get_group_by_cols(self) -> list[Any]:
1821 group_by_cols = []
1822 if self.partition_by:
1823 group_by_cols.extend(self.partition_by.get_group_by_cols())
1824 if self.order_by is not None:
1825 group_by_cols.extend(self.order_by.get_group_by_cols())
1826 return group_by_cols
1827
1828
1829class WindowFrame(Expression):
1830 """
1831 Model the frame clause in window expressions. There are two types of frame
1832 clauses which are subclasses, however, all processing and validation (by no
1833 means intended to be complete) is done here. Thus, providing an end for a
1834 frame is optional (the default is UNBOUNDED FOLLOWING, which is the last
1835 row in the frame).
1836 """
1837
1838 template = "%(frame_type)s BETWEEN %(start)s AND %(end)s"
1839 frame_type: str
1840
1841 def __init__(self, start: int | None = None, end: int | None = None):
1842 self.start = Value(start)
1843 self.end = Value(end)
1844
1845 def set_source_expressions(self, exprs: Sequence[Any]) -> None:
1846 self.start, self.end = exprs
1847
1848 def get_source_expressions(self) -> list[Any]:
1849 return [self.start, self.end]
1850
1851 def as_sql(
1852 self, compiler: SQLCompiler, connection: DatabaseConnection
1853 ) -> tuple[str, list[Any]]:
1854 start, end = self.window_frame_start_end(
1855 connection, self.start.value, self.end.value
1856 )
1857 return (
1858 self.template
1859 % {
1860 "frame_type": self.frame_type,
1861 "start": start,
1862 "end": end,
1863 },
1864 [],
1865 )
1866
1867 def __repr__(self) -> str:
1868 return f"<{self.__class__.__name__}: {self}>"
1869
1870 def get_group_by_cols(self) -> list[Any]:
1871 return []
1872
1873 def __str__(self) -> str:
1874 if self.start.value is not None and self.start.value < 0:
1875 start = f"{abs(self.start.value)} {PRECEDING}"
1876 elif self.start.value is not None and self.start.value == 0:
1877 start = CURRENT_ROW
1878 else:
1879 start = UNBOUNDED_PRECEDING
1880
1881 if self.end.value is not None and self.end.value > 0:
1882 end = f"{self.end.value} {FOLLOWING}"
1883 elif self.end.value is not None and self.end.value == 0:
1884 end = CURRENT_ROW
1885 else:
1886 end = UNBOUNDED_FOLLOWING
1887 return self.template % {
1888 "frame_type": self.frame_type,
1889 "start": start,
1890 "end": end,
1891 }
1892
1893 def window_frame_start_end(
1894 self, connection: DatabaseConnection, start: int | None, end: int | None
1895 ) -> tuple[str, str]:
1896 """Return the window frame start and end for the given connection."""
1897 raise NotImplementedError("Subclasses must implement window_frame_start_end()")
1898
1899
1900class RowRange(WindowFrame):
1901 frame_type = "ROWS"
1902
1903 def window_frame_start_end(
1904 self, connection: DatabaseConnection, start: int | None, end: int | None
1905 ) -> tuple[str, str]:
1906 return window_frame_rows_start_end(start, end)
1907
1908
1909class ValueRange(WindowFrame):
1910 frame_type = "RANGE"
1911
1912 def window_frame_start_end(
1913 self, connection: DatabaseConnection, start: int | None, end: int | None
1914 ) -> tuple[str, str]:
1915 return window_frame_range_start_end(start, end)