1from __future__ import annotations
2
3import datetime
4import itertools
5import math
6from collections.abc import Sequence
7from functools import cached_property
8from typing import TYPE_CHECKING, Any, Self
9
10from plain.postgres.dialect import (
11 OPERATORS,
12 PATTERN_ESC,
13 PATTERN_OPS,
14 lookup_cast,
15 prep_for_like_query,
16 regex_lookup,
17 year_lookup_bounds_for_date_field,
18 year_lookup_bounds_for_datetime_field,
19)
20from plain.postgres.exceptions import EmptyResultSet, FullResultSet
21from plain.postgres.expressions import Expression, Func, ResolvableExpression, Value
22from plain.postgres.fields import (
23 BooleanField,
24 DateTimeField,
25 Field,
26 IntegerField,
27 UUIDField,
28)
29from plain.postgres.query_utils import RegisterLookupMixin
30from plain.utils.datastructures import OrderedSet
31from plain.utils.hashable import make_hashable
32
33if TYPE_CHECKING:
34 from plain.postgres.connection import DatabaseConnection
35 from plain.postgres.sql.compiler import SQLCompiler
36
37
38class Lookup(Expression):
39 lookup_name: str | None = None
40 prepare_rhs: bool = True
41 can_use_none_as_rhs: bool = False
42 lhs: Any
43 rhs: Any
44
45 def __init__(self, lhs: Any, rhs: Any):
46 self.lhs, self.rhs = lhs, rhs
47 self.rhs = self.get_prep_lookup()
48 self.lhs = self.get_prep_lhs()
49 if hasattr(self.lhs, "get_bilateral_transforms"):
50 bilateral_transforms = self.lhs.get_bilateral_transforms()
51 else:
52 bilateral_transforms = []
53 if bilateral_transforms:
54 # Warn the user as soon as possible if they are trying to apply
55 # a bilateral transformation on a nested QuerySet: that won't work.
56 from plain.postgres.sql.query import Query # avoid circular import
57
58 if isinstance(rhs, Query):
59 raise NotImplementedError(
60 "Bilateral transformations on nested querysets are not implemented."
61 )
62 self.bilateral_transforms = bilateral_transforms
63
64 def apply_bilateral_transforms(self, value: Any) -> Any:
65 for transform in self.bilateral_transforms:
66 value = transform(value)
67 return value
68
69 def __repr__(self) -> str:
70 return f"{self.__class__.__name__}({self.lhs!r}, {self.rhs!r})"
71
72 def batch_process_rhs(
73 self, compiler: SQLCompiler, connection: DatabaseConnection, rhs: Any = None
74 ) -> tuple[list[str], list[Any]]:
75 if rhs is None:
76 rhs = self.rhs
77 if self.bilateral_transforms:
78 sqls: list[str] = []
79 sqls_params: list[Any] = []
80 for p in rhs:
81 value = Value(p, output_field=self.lhs.output_field)
82 value = self.apply_bilateral_transforms(value)
83 value = value.resolve_expression(compiler.query)
84 sql, sql_params = compiler.compile(value)
85 sqls.append(sql)
86 sqls_params.extend(sql_params)
87 else:
88 _, params = self.get_db_prep_lookup(rhs, connection)
89 sqls = ["%s"] * len(params)
90 sqls_params = list(params)
91 return sqls, sqls_params
92
93 def get_source_expressions(self) -> list[Any]:
94 if self.rhs_is_direct_value():
95 return [self.lhs]
96 return [self.lhs, self.rhs]
97
98 def set_source_expressions(self, exprs: Sequence[Any]) -> None:
99 exprs_list = list(exprs)
100 if len(exprs_list) == 1:
101 self.lhs = exprs_list[0]
102 else:
103 self.lhs, self.rhs = exprs_list
104
105 def get_prep_lookup(self) -> Any:
106 if not self.prepare_rhs or isinstance(self.rhs, ResolvableExpression):
107 return self.rhs
108 if output_field := getattr(self.lhs, "output_field", None):
109 if get_prep_value := getattr(output_field, "get_prep_value", None):
110 return get_prep_value(self.rhs)
111 elif self.rhs_is_direct_value():
112 return Value(self.rhs)
113 return self.rhs
114
115 def get_prep_lhs(self) -> Any:
116 if isinstance(self.lhs, ResolvableExpression):
117 return self.lhs
118 return Value(self.lhs)
119
120 def get_db_prep_lookup(
121 self, value: Any, connection: DatabaseConnection
122 ) -> tuple[str, list[Any]]:
123 return ("%s", [value])
124
125 def process_lhs(
126 self, compiler: SQLCompiler, connection: DatabaseConnection, lhs: Any = None
127 ) -> tuple[str, list[Any]]:
128 lhs = lhs or self.lhs
129 if isinstance(lhs, ResolvableExpression):
130 lhs = lhs.resolve_expression(compiler.query)
131 sql, params = compiler.compile(lhs)
132 if isinstance(lhs, Lookup):
133 # Wrapped in parentheses to respect operator precedence.
134 sql = f"({sql})"
135 return sql, list(params)
136
137 def process_rhs(
138 self, compiler: SQLCompiler, connection: DatabaseConnection
139 ) -> tuple[str, list[Any]] | tuple[list[str], list[Any]]:
140 value = self.rhs
141 if self.bilateral_transforms:
142 if self.rhs_is_direct_value():
143 # Do not call get_db_prep_lookup here as the value will be
144 # transformed before being used for lookup
145 value = Value(value, output_field=self.lhs.output_field)
146 value = self.apply_bilateral_transforms(value)
147 value = value.resolve_expression(compiler.query)
148 if hasattr(value, "as_sql"):
149 sql, params = compiler.compile(value)
150 # Ensure expression is wrapped in parentheses to respect operator
151 # precedence but avoid double wrapping.
152 if sql and sql[0] != "(":
153 sql = f"({sql})"
154 return sql, list(params)
155 else:
156 return self.get_db_prep_lookup(value, connection)
157
158 def rhs_is_direct_value(self) -> bool:
159 return not hasattr(self.rhs, "as_sql")
160
161 def get_group_by_cols(self) -> list[Any]:
162 cols = []
163 for source in self.get_source_expressions():
164 cols.extend(source.get_group_by_cols())
165 return cols
166
167 @cached_property
168 def output_field(self) -> BooleanField:
169 return BooleanField()
170
171 @property
172 def identity(self) -> tuple[type[Lookup], Any, Any]:
173 return self.__class__, self.lhs, self.rhs
174
175 def __eq__(self, other: object) -> bool:
176 if not isinstance(other, Lookup):
177 return NotImplemented
178 return self.identity == other.identity
179
180 def __hash__(self) -> int:
181 return hash(make_hashable(self.identity))
182
183 def resolve_expression(
184 self,
185 query: Any = None,
186 allow_joins: bool = True,
187 reuse: Any = None,
188 summarize: bool = False,
189 for_save: bool = False,
190 ) -> Self:
191 c = self.copy()
192 c.is_summary = summarize
193 c.lhs = self.lhs.resolve_expression(
194 query, allow_joins, reuse, summarize, for_save
195 )
196 if isinstance(self.rhs, ResolvableExpression):
197 c.rhs = self.rhs.resolve_expression(
198 query, allow_joins, reuse, summarize, for_save
199 )
200 return c
201
202 def select_format(
203 self, compiler: SQLCompiler, sql: str, params: Sequence[Any]
204 ) -> tuple[str, Sequence[Any]]:
205 # Boolean expressions work directly in SELECT
206 return sql, params
207
208
209class Transform(RegisterLookupMixin, Func):
210 """
211 RegisterLookupMixin() is first so that get_lookup() and get_transform()
212 first examine self and then check output_field.
213 """
214
215 lookup_name: str | None = None
216 bilateral: bool = False
217 arity: int = 1
218
219 @property
220 def lhs(self) -> Any:
221 return self.get_source_expressions()[0]
222
223 def get_bilateral_transforms(self) -> list[type[Transform]]:
224 if hasattr(self.lhs, "get_bilateral_transforms"):
225 bilateral_transforms = self.lhs.get_bilateral_transforms()
226 else:
227 bilateral_transforms = []
228 if self.bilateral:
229 bilateral_transforms.append(self.__class__)
230 return bilateral_transforms
231
232
233class BuiltinLookup(Lookup):
234 def process_lhs(
235 self, compiler: SQLCompiler, connection: DatabaseConnection, lhs: Any = None
236 ) -> tuple[str, list[Any]]:
237 assert self.lookup_name is not None, (
238 "lookup_name must be set on Lookup subclass"
239 )
240 lhs_sql, params = super().process_lhs(compiler, connection, lhs)
241 lhs_sql = lookup_cast(self.lookup_name, self.lhs.output_field) % lhs_sql
242 return lhs_sql, list(params)
243
244 def as_sql(
245 self, compiler: SQLCompiler, connection: DatabaseConnection
246 ) -> tuple[str, list[Any]]:
247 lhs_sql, params = self.process_lhs(compiler, connection)
248 rhs_sql, rhs_params = self.process_rhs(compiler, connection)
249 params.extend(rhs_params)
250 rhs_sql = self.get_rhs_op(connection, rhs_sql)
251 return f"{lhs_sql} {rhs_sql}", params
252
253 def get_rhs_op(self, connection: DatabaseConnection, rhs: str | list[str]) -> str:
254 assert self.lookup_name is not None, (
255 "lookup_name must be set on Lookup subclass"
256 )
257 return OPERATORS[self.lookup_name] % rhs
258
259
260class FieldGetDbPrepValueMixin(Lookup):
261 """
262 Some lookups require Field.get_db_prep_value() to be called on their
263 inputs.
264 """
265
266 get_db_prep_lookup_value_is_iterable: bool = False
267 lhs: Any
268 rhs: Any
269
270 def get_db_prep_lookup(
271 self, value: Any, connection: DatabaseConnection
272 ) -> tuple[str, list[Any]]:
273 from plain.postgres.fields.related import RelatedField
274 from plain.postgres.fields.reverse_related import ForeignObjectRel
275
276 # A relation prepares lookup values against its target column's field
277 # (e.g. the remote `id`); every other field prepares its own values.
278 # Narrowing instead of getattr() keeps `.target_field` greppable and
279 # type-checked, so removing it fails loudly rather than silently here.
280 output_field = self.lhs.output_field
281 if isinstance(output_field, RelatedField | ForeignObjectRel):
282 prep_field = output_field.target_field
283 else:
284 prep_field = output_field
285 return (
286 "%s",
287 [prep_field.get_db_prep_value(v, connection, prepared=True) for v in value]
288 if self.get_db_prep_lookup_value_is_iterable
289 else [prep_field.get_db_prep_value(value, connection, prepared=True)],
290 )
291
292
293class FieldGetDbPrepValueIterableMixin(FieldGetDbPrepValueMixin):
294 """
295 Some lookups require Field.get_db_prep_value() to be called on each value
296 in an iterable.
297 """
298
299 get_db_prep_lookup_value_is_iterable: bool = True
300 prepare_rhs: bool
301
302 def get_prep_lookup(self) -> Any:
303 if isinstance(self.rhs, ResolvableExpression):
304 return self.rhs
305 prepared_values = []
306 for rhs_value in self.rhs:
307 if isinstance(rhs_value, ResolvableExpression):
308 # An expression will be handled by the database but can coexist
309 # alongside real values.
310 pass
311 elif (
312 self.prepare_rhs
313 and (output_field := getattr(self.lhs, "output_field", None))
314 and (get_prep_value := getattr(output_field, "get_prep_value", None))
315 ):
316 rhs_value = get_prep_value(rhs_value)
317 prepared_values.append(rhs_value)
318 return prepared_values
319
320 def process_rhs(
321 self, compiler: SQLCompiler, connection: DatabaseConnection
322 ) -> tuple[str, list[Any]] | tuple[list[str], list[Any]]:
323 if self.rhs_is_direct_value():
324 # rhs should be an iterable of values. Use batch_process_rhs()
325 # to prepare/transform those values.
326 return self.batch_process_rhs(compiler, connection)
327 else:
328 return super().process_rhs(compiler, connection)
329
330 def resolve_expression_parameter(
331 self,
332 compiler: SQLCompiler,
333 connection: DatabaseConnection,
334 sql: str,
335 param: Any,
336 ) -> tuple[str, list[Any]]:
337 params: list[Any] = [param]
338 if isinstance(param, ResolvableExpression):
339 param = param.resolve_expression(compiler.query)
340 if hasattr(param, "as_sql"):
341 sql, compiled_params = compiler.compile(param)
342 params = list(compiled_params)
343 return sql, params
344
345 def batch_process_rhs(
346 self, compiler: SQLCompiler, connection: DatabaseConnection, rhs: Any = None
347 ) -> tuple[list[str], list[Any]]:
348 pre_processed = super().batch_process_rhs(compiler, connection, rhs)
349 # The params list may contain expressions which compile to a
350 # sql/param pair. Zip them to get sql and param pairs that refer to the
351 # same argument and attempt to replace them with the result of
352 # compiling the param step.
353 sql, params = zip(
354 *(
355 self.resolve_expression_parameter(compiler, connection, sql, param)
356 for sql, param in zip(*pre_processed)
357 )
358 )
359 params_list = list(itertools.chain.from_iterable(params))
360 return list(sql), params_list
361
362
363class OperatorLookup(Lookup):
364 """Lookup defined by a SQL operator."""
365
366 operator: str | None = None
367
368 def as_sql(
369 self, compiler: SQLCompiler, connection: DatabaseConnection
370 ) -> tuple[str, tuple[Any, ...]]:
371 lhs, lhs_params = self.process_lhs(compiler, connection)
372 rhs, rhs_params = self.process_rhs(compiler, connection)
373 params = tuple(lhs_params) + tuple(rhs_params)
374 return f"{lhs} {self.operator} {rhs}", params
375
376
377@Field.register_lookup
378class Exact(FieldGetDbPrepValueMixin, BuiltinLookup):
379 lookup_name: str = "exact"
380
381 def get_prep_lookup(self) -> Any:
382 from plain.postgres.sql.query import Query # avoid circular import
383
384 if isinstance(self.rhs, Query):
385 if self.rhs.has_limit_one():
386 if not self.rhs.has_select_fields:
387 self.rhs.clear_select_clause()
388 self.rhs.add_fields(["id"])
389 else:
390 raise ValueError(
391 "The QuerySet value for an exact lookup must be limited to "
392 "one result using slicing."
393 )
394 return super().get_prep_lookup()
395
396 def as_sql(
397 self, compiler: SQLCompiler, connection: DatabaseConnection
398 ) -> tuple[str, list[Any]]:
399 # Avoid comparison against direct rhs if lhs is a boolean value. That
400 # turns "boolfield__exact=True" into "WHERE boolean_field" instead of
401 # "WHERE boolean_field = True" when allowed.
402 if isinstance(self.rhs, bool) and getattr(self.lhs, "conditional", False):
403 lhs_sql, params = self.process_lhs(compiler, connection)
404 template = "%s" if self.rhs else "NOT %s"
405 return template % lhs_sql, params
406 return super().as_sql(compiler, connection)
407
408
409@Field.register_lookup
410class IExact(BuiltinLookup):
411 lookup_name: str = "iexact"
412 prepare_rhs: bool = False
413
414
415@Field.register_lookup
416class GreaterThan(FieldGetDbPrepValueMixin, BuiltinLookup):
417 lookup_name: str = "gt"
418
419
420@Field.register_lookup
421class GreaterThanOrEqual(FieldGetDbPrepValueMixin, BuiltinLookup):
422 lookup_name: str = "gte"
423
424
425@Field.register_lookup
426class LessThan(FieldGetDbPrepValueMixin, BuiltinLookup):
427 lookup_name: str = "lt"
428
429
430@Field.register_lookup
431class LessThanOrEqual(FieldGetDbPrepValueMixin, BuiltinLookup):
432 lookup_name: str = "lte"
433
434
435class IntegerFieldOverflow:
436 underflow_exception: type[Exception] = EmptyResultSet
437 overflow_exception: type[Exception] = EmptyResultSet
438 lhs: Any
439 rhs: Any
440
441 def process_rhs(
442 self, compiler: SQLCompiler, connection: DatabaseConnection
443 ) -> tuple[str, list[Any]]:
444 rhs = self.rhs
445 if isinstance(rhs, int):
446 min_value, max_value = self.lhs.output_field.integer_range
447 if min_value is not None and rhs < min_value:
448 raise self.underflow_exception
449 if max_value is not None and rhs > max_value:
450 raise self.overflow_exception
451 return super().process_rhs(compiler, connection) # ty: ignore[unresolved-attribute]
452
453
454class IntegerFieldFloatRounding:
455 """
456 Allow floats to work as query values for IntegerField. Without this, the
457 decimal portion of the float would always be discarded.
458 """
459
460 rhs: Any
461
462 def get_prep_lookup(self) -> Any:
463 if isinstance(self.rhs, float):
464 self.rhs = math.ceil(self.rhs)
465 return super().get_prep_lookup() # ty: ignore[unresolved-attribute]
466
467
468@IntegerField.register_lookup
469class IntegerFieldExact(IntegerFieldOverflow, Exact):
470 pass
471
472
473@IntegerField.register_lookup
474class IntegerGreaterThan(IntegerFieldOverflow, GreaterThan):
475 underflow_exception = FullResultSet
476
477
478@IntegerField.register_lookup
479class IntegerGreaterThanOrEqual(
480 IntegerFieldOverflow, IntegerFieldFloatRounding, GreaterThanOrEqual
481):
482 underflow_exception = FullResultSet
483
484
485@IntegerField.register_lookup
486class IntegerLessThan(IntegerFieldOverflow, IntegerFieldFloatRounding, LessThan):
487 overflow_exception = FullResultSet
488
489
490@IntegerField.register_lookup
491class IntegerLessThanOrEqual(IntegerFieldOverflow, LessThanOrEqual):
492 overflow_exception = FullResultSet
493
494
495@Field.register_lookup
496class In(FieldGetDbPrepValueIterableMixin, BuiltinLookup):
497 lookup_name: str = "in"
498
499 def get_prep_lookup(self) -> Any:
500 from plain.postgres.sql.query import Query # avoid circular import
501
502 if isinstance(self.rhs, Query):
503 self.rhs.clear_ordering(clear_default=True)
504 if not self.rhs.has_select_fields:
505 self.rhs.clear_select_clause()
506 self.rhs.add_fields(["id"])
507 return super().get_prep_lookup()
508
509 def process_rhs(
510 self, compiler: SQLCompiler, connection: DatabaseConnection
511 ) -> tuple[str, list[Any]] | tuple[list[str], list[Any]]:
512 if self.rhs_is_direct_value():
513 # Remove None from the list as NULL is never equal to anything.
514 try:
515 rhs = OrderedSet(self.rhs)
516 rhs.discard(None)
517 except TypeError: # Unhashable items in self.rhs
518 rhs = [r for r in self.rhs if r is not None]
519
520 if not rhs:
521 raise EmptyResultSet
522
523 # rhs should be an iterable; use batch_process_rhs() to
524 # prepare/transform those values.
525 sqls, sqls_params = self.batch_process_rhs(compiler, connection, rhs)
526 placeholder = "(" + ", ".join(sqls) + ")"
527 return (placeholder, sqls_params)
528 return super().process_rhs(compiler, connection)
529
530 def get_rhs_op(self, connection: DatabaseConnection, rhs: str | list[str]) -> str:
531 return f"IN {rhs}"
532
533 # PostgreSQL has no limit on IN clause size, so no need to override as_sql()
534
535
536class PatternLookup(BuiltinLookup):
537 param_pattern: str = "%%%s%%"
538 prepare_rhs: bool = False
539 bilateral_transforms: list[Any]
540
541 def get_rhs_op(self, connection: DatabaseConnection, rhs: str | list[str]) -> str:
542 # Assume we are in startswith. We need to produce SQL like:
543 # col LIKE %s, ['thevalue%']
544 # For python values we can (and should) do that directly in Python,
545 # but if the value is for example reference to other column, then
546 # we need to add the % pattern match to the lookup by something like
547 # col LIKE othercol || '%%'
548 # So, for Python values we don't need any special pattern, but for
549 # SQL reference values or SQL transformations we need the correct
550 # pattern added.
551 if hasattr(self.rhs, "as_sql") or self.bilateral_transforms:
552 assert self.lookup_name is not None, (
553 "lookup_name must be set on Lookup subclass"
554 )
555 pattern = PATTERN_OPS[self.lookup_name].format(PATTERN_ESC)
556 return pattern.format(rhs)
557 else:
558 return super().get_rhs_op(connection, rhs)
559
560 def process_rhs(
561 self, compiler: SQLCompiler, connection: DatabaseConnection
562 ) -> tuple[str, list[Any]] | tuple[list[str], list[Any]]:
563 rhs, params = super().process_rhs(compiler, connection)
564 if isinstance(rhs, str):
565 if self.rhs_is_direct_value() and params and not self.bilateral_transforms:
566 params[0] = self.param_pattern % prep_for_like_query(params[0])
567 return rhs, params
568 else:
569 return rhs, params
570
571
572@Field.register_lookup
573class Contains(PatternLookup):
574 lookup_name: str = "contains"
575
576
577@Field.register_lookup
578class IContains(Contains):
579 lookup_name: str = "icontains"
580
581
582@Field.register_lookup
583class StartsWith(PatternLookup):
584 lookup_name: str = "startswith"
585 param_pattern: str = "%s%%"
586
587
588@Field.register_lookup
589class IStartsWith(StartsWith):
590 lookup_name: str = "istartswith"
591
592
593@Field.register_lookup
594class EndsWith(PatternLookup):
595 lookup_name: str = "endswith"
596 param_pattern: str = "%%%s"
597
598
599@Field.register_lookup
600class IEndsWith(EndsWith):
601 lookup_name: str = "iendswith"
602
603
604@Field.register_lookup
605class Range(FieldGetDbPrepValueIterableMixin, BuiltinLookup):
606 lookup_name: str = "range"
607
608 def get_rhs_op(self, connection: DatabaseConnection, rhs: str | list[str]) -> str:
609 # Range lookup always receives a list of two elements from process_rhs
610 assert isinstance(rhs, list), f"Range lookup expects list, got {type(rhs)}"
611 return f"BETWEEN {rhs[0]} AND {rhs[1]}"
612
613
614@Field.register_lookup
615class IsNull(BuiltinLookup):
616 lookup_name: str = "isnull"
617 prepare_rhs: bool = False
618
619 def as_sql(
620 self, compiler: SQLCompiler, connection: DatabaseConnection
621 ) -> tuple[str, list[Any]]:
622 if not isinstance(self.rhs, bool):
623 raise TypeError(
624 "The QuerySet value for an isnull lookup must be True or False."
625 )
626 sql, params = self.process_lhs(compiler, connection)
627 if self.rhs:
628 return f"{sql} IS NULL", params
629 else:
630 return f"{sql} IS NOT NULL", params
631
632
633@Field.register_lookup
634class Regex(BuiltinLookup):
635 lookup_name: str = "regex"
636 prepare_rhs: bool = False
637
638 def as_sql(
639 self, compiler: SQLCompiler, connection: DatabaseConnection
640 ) -> tuple[str, list[Any]]:
641 if self.lookup_name in OPERATORS:
642 return super().as_sql(compiler, connection)
643 else:
644 lhs, lhs_params = self.process_lhs(compiler, connection)
645 rhs, rhs_params = self.process_rhs(compiler, connection)
646 sql_template = regex_lookup(self.lookup_name)
647 return sql_template % (lhs, rhs), lhs_params + rhs_params
648
649
650@Field.register_lookup
651class IRegex(Regex):
652 lookup_name: str = "iregex"
653
654
655class YearLookup(Lookup):
656 def year_lookup_bounds(
657 self, connection: DatabaseConnection, year: int
658 ) -> list[datetime.date] | list[datetime.datetime]:
659 from plain.postgres.functions import ExtractIsoYear
660
661 iso_year = isinstance(self.lhs, ExtractIsoYear)
662 output_field = self.lhs.lhs.output_field
663 if isinstance(output_field, DateTimeField):
664 bounds = year_lookup_bounds_for_datetime_field(year, iso_year=iso_year)
665 else:
666 bounds = year_lookup_bounds_for_date_field(year, iso_year=iso_year)
667 return bounds
668
669 def as_sql(
670 self, compiler: SQLCompiler, connection: DatabaseConnection
671 ) -> tuple[str, Sequence[Any]]:
672 # Avoid the extract operation if the rhs is a direct value to allow
673 # indexes to be used.
674 if self.rhs_is_direct_value():
675 # Skip the extract part by directly using the originating field,
676 # that is self.lhs.lhs.
677 lhs_sql, params = self.process_lhs(compiler, connection, self.lhs.lhs)
678 rhs_sql, _ = self.process_rhs(compiler, connection)
679 # rhs_sql should be a string for year lookups
680 assert isinstance(rhs_sql, str), f"Expected str, got {type(rhs_sql)}"
681 rhs_sql = self.get_direct_rhs_sql(connection, rhs_sql)
682 start, finish = self.year_lookup_bounds(connection, self.rhs)
683 params.extend(self.get_bound_params(start, finish))
684 return f"{lhs_sql} {rhs_sql}", params
685 return super().as_sql(compiler, connection)
686
687 def get_direct_rhs_sql(self, connection: DatabaseConnection, rhs: str) -> str:
688 assert self.lookup_name is not None, (
689 "lookup_name must be set on Lookup subclass"
690 )
691 return OPERATORS[self.lookup_name] % rhs
692
693 def get_bound_params(self, start: Any, finish: Any) -> tuple[Any, ...]:
694 """Return bound parameters for the year lookup."""
695 raise NotImplementedError("Subclasses must implement get_bound_params()")
696
697
698class YearExact(YearLookup, Exact):
699 def get_direct_rhs_sql(self, connection: DatabaseConnection, rhs: str) -> str:
700 return "BETWEEN %s AND %s"
701
702 def get_bound_params(self, start: Any, finish: Any) -> tuple[Any, Any]:
703 return (start, finish)
704
705
706class YearGt(YearLookup, GreaterThan):
707 def get_bound_params(self, start: Any, finish: Any) -> tuple[Any]:
708 return (finish,)
709
710
711class YearGte(YearLookup, GreaterThanOrEqual):
712 def get_bound_params(self, start: Any, finish: Any) -> tuple[Any]:
713 return (start,)
714
715
716class YearLt(YearLookup, LessThan):
717 def get_bound_params(self, start: Any, finish: Any) -> tuple[Any]:
718 return (start,)
719
720
721class YearLte(YearLookup, LessThanOrEqual):
722 def get_bound_params(self, start: Any, finish: Any) -> tuple[Any]:
723 return (finish,)
724
725
726# UUID lookups - PostgreSQL has native UUID support so these inherit directly
727# from their base classes without any special processing.
728
729
730@UUIDField.register_lookup
731class UUIDIExact(IExact):
732 pass
733
734
735@UUIDField.register_lookup
736class UUIDContains(Contains):
737 pass
738
739
740@UUIDField.register_lookup
741class UUIDIContains(IContains):
742 pass
743
744
745@UUIDField.register_lookup
746class UUIDStartsWith(StartsWith):
747 pass
748
749
750@UUIDField.register_lookup
751class UUIDIStartsWith(IStartsWith):
752 pass
753
754
755@UUIDField.register_lookup
756class UUIDEndsWith(EndsWith):
757 pass
758
759
760@UUIDField.register_lookup
761class UUIDIEndsWith(IEndsWith):
762 pass