codekingpro/portable-devtools
114k
1# dialects/postgresql/ranges.py
2# Copyright (C) 2013-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
8from __future__ import annotations
9
10import dataclasses
11from datetime import date
12from datetime import datetime
13from datetime import timedelta
14from decimal import Decimal
15from typing import Any
16from typing import cast
17from typing import Generic
18from typing import List
19from typing import Optional
20from typing import overload
21from typing import Sequence
22from typing import Tuple
23from typing import Type
24from typing import TYPE_CHECKING
25from typing import TypeVar
26from typing import Union
27
28from .operators import ADJACENT_TO
29from .operators import CONTAINED_BY
30from .operators import CONTAINS
31from .operators import NOT_EXTEND_LEFT_OF
32from .operators import NOT_EXTEND_RIGHT_OF
33from .operators import OVERLAP
34from .operators import STRICTLY_LEFT_OF
35from .operators import STRICTLY_RIGHT_OF
36from ... import types as sqltypes
37from ...sql import operators
38from ...sql.type_api import TypeEngine
39from ...util import py310
40from ...util.typing import Literal
41
42if TYPE_CHECKING:
43 from ...sql.elements import ColumnElement
44 from ...sql.type_api import _TE
45 from ...sql.type_api import TypeEngineMixin
46
47_T = TypeVar("_T", bound=Any)
48
49_BoundsType = Literal["()", "[)", "(]", "[]"]
50
51if py310:
52 dc_slots = {"slots": True}
53 dc_kwonly = {"kw_only": True}
54else:
55 dc_slots = {}
56 dc_kwonly = {}
57
58
59@dataclasses.dataclass(frozen=True, **dc_slots)
60class Range(Generic[_T]):
61 """Represent a PostgreSQL range.
62
63 E.g.::
64
65 r = Range(10, 50, bounds="()")
66
67 The calling style is similar to that of psycopg and psycopg2, in part
68 to allow easier migration from previous SQLAlchemy versions that used
69 these objects directly.
70
71 :param lower: Lower bound value, or None
72 :param upper: Upper bound value, or None
73 :param bounds: keyword-only, optional string value that is one of
74 ``"()"``, ``"[)"``, ``"(]"``, ``"[]"``. Defaults to ``"[)"``.
75 :param empty: keyword-only, optional bool indicating this is an "empty"
76 range
77
78 .. versionadded:: 2.0
79
80 """
81
82 lower: Optional[_T] = None
83 """the lower bound"""
84
85 upper: Optional[_T] = None
86 """the upper bound"""
87
88 if TYPE_CHECKING:
89 bounds: _BoundsType = dataclasses.field(default="[)")
90 empty: bool = dataclasses.field(default=False)
91 else:
92 bounds: _BoundsType = dataclasses.field(default="[)", **dc_kwonly)
93 empty: bool = dataclasses.field(default=False, **dc_kwonly)
94
95 if not py310:
96
97 def __init__(
98 self,
99 lower: Optional[_T] = None,
100 upper: Optional[_T] = None,
101 *,
102 bounds: _BoundsType = "[)",
103 empty: bool = False,
104 ):
105 # no __slots__ either so we can update dict
106 self.__dict__.update(
107 {
108 "lower": lower,
109 "upper": upper,
110 "bounds": bounds,
111 "empty": empty,
112 }
113 )
114
115 def __bool__(self) -> bool:
116 return not self.empty
117
118 @property
119 def isempty(self) -> bool:
120 "A synonym for the 'empty' attribute."
121
122 return self.empty
123
124 @property
125 def is_empty(self) -> bool:
126 "A synonym for the 'empty' attribute."
127
128 return self.empty
129
130 @property
131 def lower_inc(self) -> bool:
132 """Return True if the lower bound is inclusive."""
133
134 return self.bounds[0] == "["
135
136 @property
137 def lower_inf(self) -> bool:
138 """Return True if this range is non-empty and lower bound is
139 infinite."""
140
141 return not self.empty and self.lower is None
142
143 @property
144 def upper_inc(self) -> bool:
145 """Return True if the upper bound is inclusive."""
146
147 return self.bounds[1] == "]"
148
149 @property
150 def upper_inf(self) -> bool:
151 """Return True if this range is non-empty and the upper bound is
152 infinite."""
153
154 return not self.empty and self.upper is None
155
156 @property
157 def __sa_type_engine__(self) -> AbstractSingleRange[_T]:
158 return AbstractSingleRange()
159
160 def _contains_value(self, value: _T) -> bool:
161 """Return True if this range contains the given value."""
162
163 if self.empty:
164 return False
165
166 if self.lower is None:
167 return self.upper is None or (
168 value < self.upper
169 if self.bounds[1] == ")"
170 else value <= self.upper
171 )
172
173 if self.upper is None:
174 return ( # type: ignore
175 value > self.lower
176 if self.bounds[0] == "("
177 else value >= self.lower
178 )
179
180 return ( # type: ignore
181 value > self.lower
182 if self.bounds[0] == "("
183 else value >= self.lower
184 ) and (
185 value < self.upper
186 if self.bounds[1] == ")"
187 else value <= self.upper
188 )
189
190 def _get_discrete_step(self) -> Any:
191 "Determine the “step” for this range, if it is a discrete one."
192
193 # See
194 # https://www.postgresql.org/docs/current/rangetypes.html#RANGETYPES-DISCRETE
195 # for the rationale
196
197 if isinstance(self.lower, int) or isinstance(self.upper, int):
198 return 1
199 elif isinstance(self.lower, datetime) or isinstance(
200 self.upper, datetime
201 ):
202 # This is required, because a `isinstance(datetime.now(), date)`
203 # is True
204 return None
205 elif isinstance(self.lower, date) or isinstance(self.upper, date):
206 return timedelta(days=1)
207 else:
208 return None
209
210 def _compare_edges(
211 self,
212 value1: Optional[_T],
213 bound1: str,
214 value2: Optional[_T],
215 bound2: str,
216 only_values: bool = False,
217 ) -> int:
218 """Compare two range bounds.
219
220 Return -1, 0 or 1 respectively when `value1` is less than,
221 equal to or greater than `value2`.
222
223 When `only_value` is ``True``, do not consider the *inclusivity*
224 of the edges, just their values.
225 """
226
227 value1_is_lower_bound = bound1 in {"[", "("}
228 value2_is_lower_bound = bound2 in {"[", "("}
229
230 # Infinite edges are equal when they are on the same side,
231 # otherwise a lower edge is considered less than the upper end
232 if value1 is value2 is None:
233 if value1_is_lower_bound == value2_is_lower_bound:
234 return 0
235 else:
236 return -1 if value1_is_lower_bound else 1
237 elif value1 is None:
238 return -1 if value1_is_lower_bound else 1
239 elif value2 is None:
240 return 1 if value2_is_lower_bound else -1
241
242 # Short path for trivial case
243 if bound1 == bound2 and value1 == value2:
244 return 0
245
246 value1_inc = bound1 in {"[", "]"}
247 value2_inc = bound2 in {"[", "]"}
248 step = self._get_discrete_step()
249
250 if step is not None:
251 # "Normalize" the two edges as '[)', to simplify successive
252 # logic when the range is discrete: otherwise we would need
253 # to handle the comparison between ``(0`` and ``[1`` that
254 # are equal when dealing with integers while for floats the
255 # former is lesser than the latter
256
257 if value1_is_lower_bound:
258 if not value1_inc:
259 value1 += step
260 value1_inc = True
261 else:
262 if value1_inc:
263 value1 += step
264 value1_inc = False
265 if value2_is_lower_bound:
266 if not value2_inc:
267 value2 += step
268 value2_inc = True
269 else:
270 if value2_inc:
271 value2 += step
272 value2_inc = False
273
274 if value1 < value2: # type: ignore
275 return -1
276 elif value1 > value2: # type: ignore
277 return 1
278 elif only_values:
279 return 0
280 else:
281 # Neither one is infinite but are equal, so we
282 # need to consider the respective inclusive/exclusive
283 # flag
284
285 if value1_inc and value2_inc:
286 return 0
287 elif not value1_inc and not value2_inc:
288 if value1_is_lower_bound == value2_is_lower_bound:
289 return 0
290 else:
291 return 1 if value1_is_lower_bound else -1
292 elif not value1_inc:
293 return 1 if value1_is_lower_bound else -1
294 elif not value2_inc:
295 return -1 if value2_is_lower_bound else 1
296 else:
297 return 0
298
299 def __eq__(self, other: Any) -> bool:
300 """Compare this range to the `other` taking into account
301 bounds inclusivity, returning ``True`` if they are equal.
302 """
303
304 if not isinstance(other, Range):
305 return NotImplemented
306
307 if self.empty and other.empty:
308 return True
309 elif self.empty != other.empty:
310 return False
311
312 slower = self.lower
313 slower_b = self.bounds[0]
314 olower = other.lower
315 olower_b = other.bounds[0]
316 supper = self.upper
317 supper_b = self.bounds[1]
318 oupper = other.upper
319 oupper_b = other.bounds[1]
320
321 return (
322 self._compare_edges(slower, slower_b, olower, olower_b) == 0
323 and self._compare_edges(supper, supper_b, oupper, oupper_b) == 0
324 )
325
326 def contained_by(self, other: Range[_T]) -> bool:
327 "Determine whether this range is a contained by `other`."
328
329 # Any range contains the empty one
330 if self.empty:
331 return True
332
333 # An empty range does not contain any range except the empty one
334 if other.empty:
335 return False
336
337 slower = self.lower
338 slower_b = self.bounds[0]
339 olower = other.lower
340 olower_b = other.bounds[0]
341
342 if self._compare_edges(slower, slower_b, olower, olower_b) < 0:
343 return False
344
345 supper = self.upper
346 supper_b = self.bounds[1]
347 oupper = other.upper
348 oupper_b = other.bounds[1]
349
350 if self._compare_edges(supper, supper_b, oupper, oupper_b) > 0:
351 return False
352
353 return True
354
355 def contains(self, value: Union[_T, Range[_T]]) -> bool:
356 "Determine whether this range contains `value`."
357
358 if isinstance(value, Range):
359 return value.contained_by(self)
360 else:
361 return self._contains_value(value)
362
363 def overlaps(self, other: Range[_T]) -> bool:
364 "Determine whether this range overlaps with `other`."
365
366 # Empty ranges never overlap with any other range
367 if self.empty or other.empty:
368 return False
369
370 slower = self.lower
371 slower_b = self.bounds[0]
372 supper = self.upper
373 supper_b = self.bounds[1]
374 olower = other.lower
375 olower_b = other.bounds[0]
376 oupper = other.upper
377 oupper_b = other.bounds[1]
378
379 # Check whether this lower bound is contained in the other range
380 if (
381 self._compare_edges(slower, slower_b, olower, olower_b) >= 0
382 and self._compare_edges(slower, slower_b, oupper, oupper_b) <= 0
383 ):
384 return True
385
386 # Check whether other lower bound is contained in this range
387 if (
388 self._compare_edges(olower, olower_b, slower, slower_b) >= 0
389 and self._compare_edges(olower, olower_b, supper, supper_b) <= 0
390 ):
391 return True
392
393 return False
394
395 def strictly_left_of(self, other: Range[_T]) -> bool:
396 "Determine whether this range is completely to the left of `other`."
397
398 # Empty ranges are neither to left nor to the right of any other range
399 if self.empty or other.empty:
400 return False
401
402 supper = self.upper
403 supper_b = self.bounds[1]
404 olower = other.lower
405 olower_b = other.bounds[0]
406
407 # Check whether this upper edge is less than other's lower end
408 return self._compare_edges(supper, supper_b, olower, olower_b) < 0
409
410 __lshift__ = strictly_left_of
411
412 def strictly_right_of(self, other: Range[_T]) -> bool:
413 "Determine whether this range is completely to the right of `other`."
414
415 # Empty ranges are neither to left nor to the right of any other range
416 if self.empty or other.empty:
417 return False
418
419 slower = self.lower
420 slower_b = self.bounds[0]
421 oupper = other.upper
422 oupper_b = other.bounds[1]
423
424 # Check whether this lower edge is greater than other's upper end
425 return self._compare_edges(slower, slower_b, oupper, oupper_b) > 0
426
427 __rshift__ = strictly_right_of
428
429 def not_extend_left_of(self, other: Range[_T]) -> bool:
430 "Determine whether this does not extend to the left of `other`."
431
432 # Empty ranges are neither to left nor to the right of any other range
433 if self.empty or other.empty:
434 return False
435
436 slower = self.lower
437 slower_b = self.bounds[0]
438 olower = other.lower
439 olower_b = other.bounds[0]
440
441 # Check whether this lower edge is not less than other's lower end
442 return self._compare_edges(slower, slower_b, olower, olower_b) >= 0
443
444 def not_extend_right_of(self, other: Range[_T]) -> bool:
445 "Determine whether this does not extend to the right of `other`."
446
447 # Empty ranges are neither to left nor to the right of any other range
448 if self.empty or other.empty:
449 return False
450
451 supper = self.upper
452 supper_b = self.bounds[1]
453 oupper = other.upper
454 oupper_b = other.bounds[1]
455
456 # Check whether this upper edge is not greater than other's upper end
457 return self._compare_edges(supper, supper_b, oupper, oupper_b) <= 0
458
459 def _upper_edge_adjacent_to_lower(
460 self,
461 value1: Optional[_T],
462 bound1: str,
463 value2: Optional[_T],
464 bound2: str,
465 ) -> bool:
466 """Determine whether an upper bound is immediately successive to a
467 lower bound."""
468
469 # Since we need a peculiar way to handle the bounds inclusivity,
470 # just do a comparison by value here
471 res = self._compare_edges(value1, bound1, value2, bound2, True)
472 if res == -1:
473 step = self._get_discrete_step()
474 if step is None:
475 return False
476 if bound1 == "]":
477 if bound2 == "[":
478 return value1 == value2 - step # type: ignore
479 else:
480 return value1 == value2
481 else:
482 if bound2 == "[":
483 return value1 == value2
484 else:
485 return value1 == value2 - step # type: ignore
486 elif res == 0:
487 # Cover cases like [0,0] -|- [1,] and [0,2) -|- (1,3]
488 if (
489 bound1 == "]"
490 and bound2 == "["
491 or bound1 == ")"
492 and bound2 == "("
493 ):
494 step = self._get_discrete_step()
495 if step is not None:
496 return True
497 return (
498 bound1 == ")"
499 and bound2 == "["
500 or bound1 == "]"
501 and bound2 == "("
502 )
503 else:
504 return False
505
506 def adjacent_to(self, other: Range[_T]) -> bool:
507 "Determine whether this range is adjacent to the `other`."
508
509 # Empty ranges are not adjacent to any other range
510 if self.empty or other.empty:
511 return False
512
513 slower = self.lower
514 slower_b = self.bounds[0]
515 supper = self.upper
516 supper_b = self.bounds[1]
517 olower = other.lower
518 olower_b = other.bounds[0]
519 oupper = other.upper
520 oupper_b = other.bounds[1]
521
522 return self._upper_edge_adjacent_to_lower(
523 supper, supper_b, olower, olower_b
524 ) or self._upper_edge_adjacent_to_lower(
525 oupper, oupper_b, slower, slower_b
526 )
527
528 def union(self, other: Range[_T]) -> Range[_T]:
529 """Compute the union of this range with the `other`.
530
531 This raises a ``ValueError`` exception if the two ranges are
532 "disjunct", that is neither adjacent nor overlapping.
533 """
534
535 # Empty ranges are "additive identities"
536 if self.empty:
537 return other
538 if other.empty:
539 return self
540
541 if not self.overlaps(other) and not self.adjacent_to(other):
542 raise ValueError(
543 "Adding non-overlapping and non-adjacent"
544 " ranges is not implemented"
545 )
546
547 slower = self.lower
548 slower_b = self.bounds[0]
549 supper = self.upper
550 supper_b = self.bounds[1]
551 olower = other.lower
552 olower_b = other.bounds[0]
553 oupper = other.upper
554 oupper_b = other.bounds[1]
555
556 if self._compare_edges(slower, slower_b, olower, olower_b) < 0:
557 rlower = slower
558 rlower_b = slower_b
559 else:
560 rlower = olower
561 rlower_b = olower_b
562
563 if self._compare_edges(supper, supper_b, oupper, oupper_b) > 0:
564 rupper = supper
565 rupper_b = supper_b
566 else:
567 rupper = oupper
568 rupper_b = oupper_b
569
570 return Range(
571 rlower, rupper, bounds=cast(_BoundsType, rlower_b + rupper_b)
572 )
573
574 def __add__(self, other: Range[_T]) -> Range[_T]:
575 return self.union(other)
576
577 def difference(self, other: Range[_T]) -> Range[_T]:
578 """Compute the difference between this range and the `other`.
579
580 This raises a ``ValueError`` exception if the two ranges are
581 "disjunct", that is neither adjacent nor overlapping.
582 """
583
584 # Subtracting an empty range is a no-op
585 if self.empty or other.empty:
586 return self
587
588 slower = self.lower
589 slower_b = self.bounds[0]
590 supper = self.upper
591 supper_b = self.bounds[1]
592 olower = other.lower
593 olower_b = other.bounds[0]
594 oupper = other.upper
595 oupper_b = other.bounds[1]
596
597 sl_vs_ol = self._compare_edges(slower, slower_b, olower, olower_b)
598 su_vs_ou = self._compare_edges(supper, supper_b, oupper, oupper_b)
599 if sl_vs_ol < 0 and su_vs_ou > 0:
600 raise ValueError(
601 "Subtracting a strictly inner range is not implemented"
602 )
603
604 sl_vs_ou = self._compare_edges(slower, slower_b, oupper, oupper_b)
605 su_vs_ol = self._compare_edges(supper, supper_b, olower, olower_b)
606
607 # If the ranges do not overlap, result is simply the first
608 if sl_vs_ou > 0 or su_vs_ol < 0:
609 return self
610
611 # If this range is completely contained by the other, result is empty
612 if sl_vs_ol >= 0 and su_vs_ou <= 0:
613 return Range(None, None, empty=True)
614
615 # If this range extends to the left of the other and ends in its
616 # middle
617 if sl_vs_ol <= 0 and su_vs_ol >= 0 and su_vs_ou <= 0:
618 rupper_b = ")" if olower_b == "[" else "]"
619 if (
620 slower_b != "["
621 and rupper_b != "]"
622 and self._compare_edges(slower, slower_b, olower, rupper_b)
623 == 0
624 ):
625 return Range(None, None, empty=True)
626 else:
627 return Range(
628 slower,
629 olower,
630 bounds=cast(_BoundsType, slower_b + rupper_b),
631 )
632
633 # If this range starts in the middle of the other and extends to its
634 # right
635 if sl_vs_ol >= 0 and su_vs_ou >= 0 and sl_vs_ou <= 0:
636 rlower_b = "(" if oupper_b == "]" else "["
637 if (
638 rlower_b != "["
639 and supper_b != "]"
640 and self._compare_edges(oupper, rlower_b, supper, supper_b)
641 == 0
642 ):
643 return Range(None, None, empty=True)
644 else:
645 return Range(
646 oupper,
647 supper,
648 bounds=cast(_BoundsType, rlower_b + supper_b),
649 )
650
651 assert False, f"Unhandled case computing {self} - {other}"
652
653 def __sub__(self, other: Range[_T]) -> Range[_T]:
654 return self.difference(other)
655
656 def intersection(self, other: Range[_T]) -> Range[_T]:
657 """Compute the intersection of this range with the `other`.
658
659 .. versionadded:: 2.0.10
660
661 """
662 if self.empty or other.empty or not self.overlaps(other):
663 return Range(None, None, empty=True)
664
665 slower = self.lower
666 slower_b = self.bounds[0]
667 supper = self.upper
668 supper_b = self.bounds[1]
669 olower = other.lower
670 olower_b = other.bounds[0]
671 oupper = other.upper
672 oupper_b = other.bounds[1]
673
674 if self._compare_edges(slower, slower_b, olower, olower_b) < 0:
675 rlower = olower
676 rlower_b = olower_b
677 else:
678 rlower = slower
679 rlower_b = slower_b
680
681 if self._compare_edges(supper, supper_b, oupper, oupper_b) > 0:
682 rupper = oupper
683 rupper_b = oupper_b
684 else:
685 rupper = supper
686 rupper_b = supper_b
687
688 return Range(
689 rlower,
690 rupper,
691 bounds=cast(_BoundsType, rlower_b + rupper_b),
692 )
693
694 def __mul__(self, other: Range[_T]) -> Range[_T]:
695 return self.intersection(other)
696
697 def __str__(self) -> str:
698 return self._stringify()
699
700 def _stringify(self) -> str:
701 if self.empty:
702 return "empty"
703
704 l, r = self.lower, self.upper
705 l = "" if l is None else l # type: ignore
706 r = "" if r is None else r # type: ignore
707
708 b0, b1 = cast("Tuple[str, str]", self.bounds)
709
710 return f"{b0}{l},{r}{b1}"
711
712
713class MultiRange(List[Range[_T]]):
714 """Represents a multirange sequence.
715
716 This list subclass is an utility to allow automatic type inference of
717 the proper multi-range SQL type depending on the single range values.
718 This is useful when operating on literal multi-ranges::
719
720 import sqlalchemy as sa
721 from sqlalchemy.dialects.postgresql import MultiRange, Range
722
723 value = literal(MultiRange([Range(2, 4)]))
724
725 select(tbl).where(tbl.c.value.op("@")(MultiRange([Range(-3, 7)])))
726
727 .. versionadded:: 2.0.26
728
729 .. seealso::
730
731 - :ref:`postgresql_multirange_list_use`.
732 """
733
734 @property
735 def __sa_type_engine__(self) -> AbstractMultiRange[_T]:
736 return AbstractMultiRange()
737
738
739class AbstractRange(sqltypes.TypeEngine[_T]):
740 """Base class for single and multi Range SQL types."""
741
742 render_bind_cast = True
743
744 __abstract__ = True
745
746 @overload
747 def adapt(self, cls: Type[_TE], **kw: Any) -> _TE: ...
748
749 @overload
750 def adapt(
751 self, cls: Type[TypeEngineMixin], **kw: Any
752 ) -> TypeEngine[Any]: ...
753
754 def adapt(
755 self,
756 cls: Type[Union[TypeEngine[Any], TypeEngineMixin]],
757 **kw: Any,
758 ) -> TypeEngine[Any]:
759 """Dynamically adapt a range type to an abstract impl.
760
761 For example ``INT4RANGE().adapt(_Psycopg2NumericRange)`` should
762 produce a type that will have ``_Psycopg2NumericRange`` behaviors
763 and also render as ``INT4RANGE`` in SQL and DDL.
764
765 """
766 if (
767 issubclass(cls, (AbstractSingleRangeImpl, AbstractMultiRangeImpl))
768 and cls is not self.__class__
769 ):
770 # two ways to do this are: 1. create a new type on the fly
771 # or 2. have AbstractRangeImpl(visit_name) constructor and a
772 # visit_abstract_range_impl() method in the PG compiler.
773 # I'm choosing #1 as the resulting type object
774 # will then make use of the same mechanics
775 # as if we had made all these sub-types explicitly, and will
776 # also look more obvious under pdb etc.
777 # The adapt() operation here is cached per type-class-per-dialect,
778 # so is not much of a performance concern
779 visit_name = self.__visit_name__
780 return type( # type: ignore
781 f"{visit_name}RangeImpl",
782 (cls, self.__class__),
783 {"__visit_name__": visit_name},
784 )()
785 else:
786 return super().adapt(cls)
787
788 class comparator_factory(TypeEngine.Comparator[Range[Any]]):
789 """Define comparison operations for range types."""
790
791 def contains(self, other: Any, **kw: Any) -> ColumnElement[bool]:
792 """Boolean expression. Returns true if the right hand operand,
793 which can be an element or a range, is contained within the
794 column.
795
796 kwargs may be ignored by this operator but are required for API
797 conformance.
798 """
799 return self.expr.operate(CONTAINS, other)
800
801 def contained_by(self, other: Any) -> ColumnElement[bool]:
802 """Boolean expression. Returns true if the column is contained
803 within the right hand operand.
804 """
805 return self.expr.operate(CONTAINED_BY, other)
806
807 def overlaps(self, other: Any) -> ColumnElement[bool]:
808 """Boolean expression. Returns true if the column overlaps
809 (has points in common with) the right hand operand.
810 """
811 return self.expr.operate(OVERLAP, other)
812
813 def strictly_left_of(self, other: Any) -> ColumnElement[bool]:
814 """Boolean expression. Returns true if the column is strictly
815 left of the right hand operand.
816 """
817 return self.expr.operate(STRICTLY_LEFT_OF, other)
818
819 __lshift__ = strictly_left_of
820
821 def strictly_right_of(self, other: Any) -> ColumnElement[bool]:
822 """Boolean expression. Returns true if the column is strictly
823 right of the right hand operand.
824 """
825 return self.expr.operate(STRICTLY_RIGHT_OF, other)
826
827 __rshift__ = strictly_right_of
828
829 def not_extend_right_of(self, other: Any) -> ColumnElement[bool]:
830 """Boolean expression. Returns true if the range in the column
831 does not extend right of the range in the operand.
832 """
833 return self.expr.operate(NOT_EXTEND_RIGHT_OF, other)
834
835 def not_extend_left_of(self, other: Any) -> ColumnElement[bool]:
836 """Boolean expression. Returns true if the range in the column
837 does not extend left of the range in the operand.
838 """
839 return self.expr.operate(NOT_EXTEND_LEFT_OF, other)
840
841 def adjacent_to(self, other: Any) -> ColumnElement[bool]:
842 """Boolean expression. Returns true if the range in the column
843 is adjacent to the range in the operand.
844 """
845 return self.expr.operate(ADJACENT_TO, other)
846
847 def union(self, other: Any) -> ColumnElement[bool]:
848 """Range expression. Returns the union of the two ranges.
849 Will raise an exception if the resulting range is not
850 contiguous.
851 """
852 return self.expr.operate(operators.add, other)
853
854 def difference(self, other: Any) -> ColumnElement[bool]:
855 """Range expression. Returns the union of the two ranges.
856 Will raise an exception if the resulting range is not
857 contiguous.
858 """
859 return self.expr.operate(operators.sub, other)
860
861 def intersection(self, other: Any) -> ColumnElement[Range[_T]]:
862 """Range expression. Returns the intersection of the two ranges.
863 Will raise an exception if the resulting range is not
864 contiguous.
865 """
866 return self.expr.operate(operators.mul, other)
867
868
869class AbstractSingleRange(AbstractRange[Range[_T]]):
870 """Base for PostgreSQL RANGE types.
871
872 These are types that return a single :class:`_postgresql.Range` object.
873
874 .. seealso::
875
876 `PostgreSQL range functions <https://www.postgresql.org/docs/current/static/functions-range.html>`_
877
878 """ # noqa: E501
879
880 __abstract__ = True
881
882 def _resolve_for_literal(self, value: Range[Any]) -> Any:
883 spec = value.lower if value.lower is not None else value.upper
884
885 if isinstance(spec, int):
886 # pg is unreasonably picky here: the query
887 # "select 1::INTEGER <@ '[1, 4)'::INT8RANGE" raises
888 # "operator does not exist: integer <@ int8range" as of pg 16
889 if _is_int32(value):
890 return INT4RANGE()
891 else:
892 return INT8RANGE()
893 elif isinstance(spec, (Decimal, float)):
894 return NUMRANGE()
895 elif isinstance(spec, datetime):
896 return TSRANGE() if not spec.tzinfo else TSTZRANGE()
897 elif isinstance(spec, date):
898 return DATERANGE()
899 else:
900 # empty Range, SQL datatype can't be determined here
901 return sqltypes.NULLTYPE
902
903
904class AbstractSingleRangeImpl(AbstractSingleRange[_T]):
905 """Marker for AbstractSingleRange that will apply a subclass-specific
906 adaptation"""
907
908
909class AbstractMultiRange(AbstractRange[Sequence[Range[_T]]]):
910 """Base for PostgreSQL MULTIRANGE types.
911
912 these are types that return a sequence of :class:`_postgresql.Range`
913 objects.
914
915 """
916
917 __abstract__ = True
918
919 def _resolve_for_literal(self, value: Sequence[Range[Any]]) -> Any:
920 if not value:
921 # empty MultiRange, SQL datatype can't be determined here
922 return sqltypes.NULLTYPE
923 first = value[0]
924 spec = first.lower if first.lower is not None else first.upper
925
926 if isinstance(spec, int):
927 # pg is unreasonably picky here: the query
928 # "select 1::INTEGER <@ '{[1, 4),[6,19)}'::INT8MULTIRANGE" raises
929 # "operator does not exist: integer <@ int8multirange" as of pg 16
930 if all(_is_int32(r) for r in value):
931 return INT4MULTIRANGE()
932 else:
933 return INT8MULTIRANGE()
934 elif isinstance(spec, (Decimal, float)):
935 return NUMMULTIRANGE()
936 elif isinstance(spec, datetime):
937 return TSMULTIRANGE() if not spec.tzinfo else TSTZMULTIRANGE()
938 elif isinstance(spec, date):
939 return DATEMULTIRANGE()
940 else:
941 # empty Range, SQL datatype can't be determined here
942 return sqltypes.NULLTYPE
943
944
945class AbstractMultiRangeImpl(AbstractMultiRange[_T]):
946 """Marker for AbstractMultiRange that will apply a subclass-specific
947 adaptation"""
948
949
950class INT4RANGE(AbstractSingleRange[int]):
951 """Represent the PostgreSQL INT4RANGE type."""
952
953 __visit_name__ = "INT4RANGE"
954
955
956class INT8RANGE(AbstractSingleRange[int]):
957 """Represent the PostgreSQL INT8RANGE type."""
958
959 __visit_name__ = "INT8RANGE"
960
961
962class NUMRANGE(AbstractSingleRange[Decimal]):
963 """Represent the PostgreSQL NUMRANGE type."""
964
965 __visit_name__ = "NUMRANGE"
966
967
968class DATERANGE(AbstractSingleRange[date]):
969 """Represent the PostgreSQL DATERANGE type."""
970
971 __visit_name__ = "DATERANGE"
972
973
974class TSRANGE(AbstractSingleRange[datetime]):
975 """Represent the PostgreSQL TSRANGE type."""
976
977 __visit_name__ = "TSRANGE"
978
979
980class TSTZRANGE(AbstractSingleRange[datetime]):
981 """Represent the PostgreSQL TSTZRANGE type."""
982
983 __visit_name__ = "TSTZRANGE"
984
985
986class INT4MULTIRANGE(AbstractMultiRange[int]):
987 """Represent the PostgreSQL INT4MULTIRANGE type."""
988
989 __visit_name__ = "INT4MULTIRANGE"
990
991
992class INT8MULTIRANGE(AbstractMultiRange[int]):
993 """Represent the PostgreSQL INT8MULTIRANGE type."""
994
995 __visit_name__ = "INT8MULTIRANGE"
996
997
998class NUMMULTIRANGE(AbstractMultiRange[Decimal]):
999 """Represent the PostgreSQL NUMMULTIRANGE type."""
1000
1001 __visit_name__ = "NUMMULTIRANGE"
1002
1003
1004class DATEMULTIRANGE(AbstractMultiRange[date]):
1005 """Represent the PostgreSQL DATEMULTIRANGE type."""
1006
1007 __visit_name__ = "DATEMULTIRANGE"
1008
1009
1010class TSMULTIRANGE(AbstractMultiRange[datetime]):
1011 """Represent the PostgreSQL TSRANGE type."""
1012
1013 __visit_name__ = "TSMULTIRANGE"
1014
1015
1016class TSTZMULTIRANGE(AbstractMultiRange[datetime]):
1017 """Represent the PostgreSQL TSTZRANGE type."""
1018
1019 __visit_name__ = "TSTZMULTIRANGE"
1020
1021
1022_max_int_32 = 2**31 - 1
1023_min_int_32 = -(2**31)
1024
1025
1026def _is_int32(r: Range[int]) -> bool:
1027 return (r.lower is None or _min_int_32 <= r.lower <= _max_int_32) and (
1028 r.upper is None or _min_int_32 <= r.upper <= _max_int_32
1029 )
1030