Coverage for /usr/local/lib/python3.12/site-packages/prefect/server/events/filters.py: 55%
350 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 02:04 +0000
1from __future__ import annotations
3import sys
4from collections.abc import Iterable
5from datetime import datetime, timedelta
6from typing import TYPE_CHECKING, Optional, Sequence, Union
7from uuid import UUID
8from zoneinfo import ZoneInfo
10import sqlalchemy as sa
11from pydantic import Field, PrivateAttr
12from sqlalchemy import orm
13from sqlalchemy.sql import Select
15import prefect.types._datetime
16from prefect.server.database import PrefectDBInterface, db_injector
17from prefect.server.schemas.filters import (
18 PrefectFilterBaseModel,
19 PrefectOperatorFilterBaseModel,
20)
21from prefect.server.utilities.schemas.bases import PrefectBaseModel
22from prefect.server.utilities.text_search_parser import (
23 parse_text_search_query,
24)
25from prefect.types import DateTime
26from prefect.utilities.collections import AutoEnum
28from .schemas.events import Event, Resource, ResourceSpecification
30if TYPE_CHECKING: 30 ↛ 31line 30 didn't jump to line 31 because the condition on line 30 was never true
31 from sqlalchemy.sql.expression import ColumnElement, ColumnExpressionArgument
34class AutomationFilterCreated(PrefectFilterBaseModel):
35 """Filter by `Automation.created`."""
37 before_: Optional[DateTime] = Field(
38 default=None,
39 description="Only include automations created before this datetime",
40 )
42 def _get_filter_list(
43 self, db: PrefectDBInterface
44 ) -> Iterable[sa.ColumnElement[bool]]:
45 if self.before_ is not None: 45 ↛ 47line 45 didn't jump to line 47 because the condition on line 45 was always true
46 return [db.Automation.created <= self.before_]
47 return ()
50class AutomationFilterName(PrefectFilterBaseModel):
51 """Filter by `Automation.created`."""
53 any_: Optional[list[str]] = Field(
54 default=None,
55 description="Only include automations with names that match any of these strings",
56 )
58 def _get_filter_list(self, db: PrefectDBInterface) -> list[sa.ColumnElement[bool]]:
59 if self.any_ is not None: 59 ↛ 60line 59 didn't jump to line 60 because the condition on line 59 was never true
60 return [db.Automation.name.in_(self.any_)]
61 return []
64class AutomationFilterTags(PrefectOperatorFilterBaseModel):
65 """Filter by `Automation.tags`."""
67 all_: Optional[list[str]] = Field(
68 default=None,
69 examples=[["tag-1", "tag-2"]],
70 description="A list of tags. Automations will be returned only if their tags are a superset of the list",
71 )
72 any_: Optional[list[str]] = Field(
73 default=None,
74 examples=[["tag-1", "tag-2"]],
75 description="A list of tags. Automations will be returned if their tags contain any of the tags in the list",
76 )
77 is_null_: Optional[bool] = Field(
78 default=None, description="If true, only include automations without tags"
79 )
81 def _get_filter_list(
82 self, db: PrefectDBInterface
83 ) -> Iterable[sa.ColumnExpressionArgument[bool]]:
84 from prefect.server.schemas.filters import _as_array
86 filters: list[sa.ColumnElement[bool]] = []
87 if self.all_ is not None:
88 filters.append(db.Automation.tags.has_all(_as_array(self.all_)))
89 if self.any_ is not None:
90 filters.append(db.Automation.tags.has_any(_as_array(self.any_)))
91 if self.is_null_ is not None:
92 filters.append(
93 db.Automation.tags == [] if self.is_null_ else db.Automation.tags != []
94 )
95 return filters
98class AutomationFilter(PrefectOperatorFilterBaseModel):
99 name: Optional[AutomationFilterName] = Field(
100 default=None, description="Filter criteria for `Automation.name`"
101 )
102 created: Optional[AutomationFilterCreated] = Field(
103 default=None, description="Filter criteria for `Automation.created`"
104 )
105 tags: Optional[AutomationFilterTags] = Field(
106 default=None, description="Filter criteria for `Automation.tags`"
107 )
109 def _get_filter_list(
110 self, db: PrefectDBInterface
111 ) -> Iterable[sa.ColumnExpressionArgument[bool]]:
112 filters: list[sa.ColumnExpressionArgument[bool]] = []
114 if self.name is not None:
115 filters.append(self.name.as_sql_filter())
116 if self.created is not None:
117 filters.append(self.created.as_sql_filter())
118 if self.tags is not None:
119 filters.append(self.tags.as_sql_filter())
121 return filters
124class EventDataFilter(PrefectBaseModel, extra="forbid"):
125 """A base class for filtering event data."""
127 _top_level_filter: Optional[Select[tuple[UUID]]] = PrivateAttr(None)
129 def get_filters(self) -> list["EventDataFilter"]:
130 filters: list[EventDataFilter] = []
131 for filter in [
132 getattr(self, name) for name in self.__class__.model_fields.keys()
133 ]:
134 # Any embedded list of filters are flattened and thus ANDed together
135 subfilters: list[EventDataFilter] = (
136 filter if isinstance(filter, list) else [filter]
137 )
139 for subfilter in subfilters:
140 if not isinstance(subfilter, EventDataFilter):
141 continue
143 subfilter._top_level_filter = self._top_level_filter
144 filters.append(subfilter)
146 return filters
148 def includes(self, event: Event) -> bool:
149 """Does the given event match the criteria of this filter?"""
150 return all(filter.includes(event) for filter in self.get_filters())
152 def excludes(self, event: Event) -> bool:
153 """Would the given filter exclude this event?"""
154 return not self.includes(event)
156 def build_where_clauses(self) -> Sequence["ColumnExpressionArgument[bool]"]:
157 """Convert the criteria to a WHERE clause."""
158 clauses: list["ColumnExpressionArgument[bool]"] = []
159 for filter in self.get_filters():
160 clauses.extend(filter.build_where_clauses())
161 return clauses
164class EventOccurredFilter(EventDataFilter):
165 since: DateTime = Field(
166 default_factory=lambda: (
167 prefect.types._datetime.start_of_day(prefect.types._datetime.now("UTC"))
168 - timedelta(days=180)
169 ),
170 description="Only include events after this time (inclusive)",
171 )
172 until: DateTime = Field(
173 default_factory=lambda: prefect.types._datetime.now("UTC"),
174 description="Only include events prior to this time (inclusive)",
175 )
177 def clamp(self, max_duration: timedelta) -> None:
178 """Limit how far the query can look back based on the given duration"""
179 # Using datetime.now() instead of prefect.types._datetime.now() to avoid
180 # dropping timezone information which happens if pendulum is used
181 earliest = datetime.now(ZoneInfo("UTC")) - max_duration
182 self.since = max(earliest, self.since)
184 def includes(self, event: Event) -> bool:
185 return self.since <= event.occurred <= self.until
187 @db_injector
188 def build_where_clauses(
189 self, db: PrefectDBInterface
190 ) -> Sequence["ColumnExpressionArgument[bool]"]:
191 return [
192 db.Event.occurred >= self.since,
193 db.Event.occurred <= self.until,
194 ]
197class EventNameFilter(EventDataFilter):
198 prefix: Optional[list[str]] = Field(
199 default=None, description="Only include events matching one of these prefixes"
200 )
201 exclude_prefix: Optional[list[str]] = Field(
202 default=None, description="Exclude events matching one of these prefixes"
203 )
205 name: Optional[list[str]] = Field(
206 default=None,
207 description="Only include events matching one of these names exactly",
208 )
209 exclude_name: Optional[list[str]] = Field(
210 default=None, description="Exclude events matching one of these names exactly"
211 )
213 def includes(self, event: Event) -> bool:
214 if self.prefix:
215 if not any(event.event.startswith(prefix) for prefix in self.prefix):
216 return False
218 if self.exclude_prefix:
219 if any(event.event.startswith(prefix) for prefix in self.exclude_prefix):
220 return False
222 if self.name:
223 if not any(event.event == name for name in self.name):
224 return False
226 if self.exclude_name:
227 if any(event.event == name for name in self.exclude_name):
228 return False
230 return True
232 @db_injector
233 def build_where_clauses(
234 self, db: PrefectDBInterface
235 ) -> Sequence["ColumnExpressionArgument[bool]"]:
236 filters: list["ColumnExpressionArgument[bool]"] = []
238 if self.prefix: 238 ↛ 239line 238 didn't jump to line 239 because the condition on line 238 was never true
239 filters.append(
240 sa.or_(*(db.Event.event.startswith(prefix) for prefix in self.prefix))
241 )
243 if self.exclude_prefix: 243 ↛ 249line 243 didn't jump to line 249 because the condition on line 243 was always true
244 filters.extend(
245 sa.not_(db.Event.event.startswith(prefix))
246 for prefix in self.exclude_prefix
247 )
249 if self.name: 249 ↛ 252line 249 didn't jump to line 252 because the condition on line 249 was always true
250 filters.append(db.Event.event.in_(self.name))
252 if self.exclude_name: 252 ↛ 255line 252 didn't jump to line 255 because the condition on line 252 was always true
253 filters.append(db.Event.event.not_in(self.exclude_name))
255 return filters
258def _is_like_match_expression(
259 column: orm.InstrumentedAttribute,
260 match_expression: str,
261) -> sa.BinaryExpression:
262 """Translate a match expression to a SQL LIKE expression."""
263 is_negated = match_expression.startswith("!")
265 translation = str.maketrans(
266 {"*": "%", "?": "_", "\\": "\\\\", "%": "\\%", "_": "\\_"}
267 )
268 expression = column.like(
269 match_expression.removeprefix("!").translate(translation), escape="\\"
270 )
272 return ~expression if is_negated else expression
275class EventResourceFilter(EventDataFilter):
276 id: Optional[list[str]] = Field(
277 default=None, description="Only include events for resources with these IDs"
278 )
279 id_prefix: Optional[list[str]] = Field(
280 default=None,
281 description=(
282 "Only include events for resources with IDs starting with these prefixes."
283 ),
284 )
285 labels: Optional[ResourceSpecification] = Field(
286 default=None, description="Only include events for resources with these labels"
287 )
288 distinct: bool = Field(
289 default=False,
290 description="Only include events for distinct resources",
291 )
293 def includes(self, event: Event) -> bool:
294 if self.id:
295 if not any(event.resource.id == resource_id for resource_id in self.id):
296 return False
298 if self.id_prefix:
299 if not any(
300 event.resource.id.startswith(prefix) for prefix in self.id_prefix
301 ):
302 return False
304 if self.labels:
305 if not self.labels.matches(event.resource):
306 return False
308 return True
310 @db_injector
311 def build_where_clauses(
312 self, db: PrefectDBInterface
313 ) -> Sequence["ColumnExpressionArgument[bool]"]:
314 filters: list["ColumnExpressionArgument[bool]"] = []
316 # If we're doing an exact or prefix search on resource_id, this is efficient
317 # enough to do on the events table without going to the event_resources table
319 if self.id: 319 ↛ 320line 319 didn't jump to line 320 because the condition on line 319 was never true
320 filters.append(db.Event.resource_id.in_(self.id))
322 if self.id_prefix: 322 ↛ 323line 322 didn't jump to line 323 because the condition on line 322 was never true
323 filters.append(
324 sa.or_(
325 *(
326 db.Event.resource_id.startswith(prefix)
327 for prefix in self.id_prefix
328 )
329 )
330 )
332 if self.labels: 332 ↛ 333line 332 didn't jump to line 333 because the condition on line 332 was never true
333 labels = self.labels.deepcopy()
335 # We are explicitly searching for the primary resource here so the
336 # resource_role must be ''
337 label_filters = [db.EventResource.resource_role == ""]
339 # On the event_resources table, resource_id is unpacked
340 # into a column, so we should search for it there
341 if resource_ids := labels.pop("prefect.resource.id", None):
342 resource_id_column = db.EventResource.resource_id
343 label_filter = sa.or_(
344 *(
345 _is_like_match_expression(resource_id_column, match_expression)
346 for match_expression in resource_ids
347 )
348 )
349 label_filters.append(label_filter)
351 if labels:
352 for _, (label, values) in enumerate(labels.items()):
353 # Empty label value arrays should match nothing
354 if not values:
355 label_filters.append(sa.false())
356 continue
358 label_column = db.EventResource.resource[label].astext
359 label_filter = sa.or_(
360 *(
361 _is_like_match_expression(label_column, match_expression)
362 for match_expression in values
363 )
364 )
365 label_filters.append(label_filter)
367 assert self._top_level_filter is not None
368 filters.append(
369 db.Event.id.in_(self._top_level_filter.where(*label_filters))
370 )
372 return filters
375class EventRelatedFilter(EventDataFilter):
376 id: Optional[list[str]] = Field(
377 None, description="Only include events for related resources with these IDs"
378 )
379 role: Optional[list[str]] = Field(
380 None, description="Only include events for related resources in these roles"
381 )
382 resources_in_roles: Optional[list[tuple[str, str]]] = Field(
383 None,
384 description=(
385 "Only include events with specific related resources in specific roles"
386 ),
387 )
388 labels: Optional[ResourceSpecification] = Field(
389 None, description="Only include events for related resources with these labels"
390 )
392 @db_injector
393 def build_where_clauses(
394 self, db: PrefectDBInterface
395 ) -> Sequence["ColumnExpressionArgument[bool]"]:
396 filters: list["ColumnExpressionArgument[bool]"] = []
398 if self.id:
399 filters.append(db.EventResource.resource_id.in_(self.id))
401 if self.role:
402 filters.append(db.EventResource.resource_role.in_(self.role))
404 if self.resources_in_roles:
405 filters.append(
406 sa.or_(
407 *(
408 sa.and_(
409 db.EventResource.resource_id == resource_id,
410 db.EventResource.resource_role == role,
411 )
412 for resource_id, role in self.resources_in_roles
413 )
414 )
415 )
417 if self.labels:
418 label_filters: list[ColumnElement[bool]] = []
419 labels = self.labels.deepcopy()
421 # On the event_resources table, resource_id and resource_role are unpacked
422 # into columns, so we should search there for them
423 if resource_ids := labels.pop("prefect.resource.id", None):
424 resource_id_column = db.EventResource.resource_id
425 label_filter = sa.or_(
426 *(
427 _is_like_match_expression(resource_id_column, match_expression)
428 for match_expression in resource_ids
429 )
430 )
431 label_filters.append(label_filter)
433 if roles := labels.pop("prefect.resource.role", None):
434 label_filters.append(db.EventResource.resource_role.in_(roles))
436 if labels:
437 for _, (label, values) in enumerate(labels.items()):
438 # Empty label value arrays should match nothing
439 if not values:
440 label_filters.append(sa.false())
441 continue
443 label_column = db.EventResource.resource[label].astext
444 label_filter = sa.or_(
445 *(
446 _is_like_match_expression(label_column, match_expression)
447 for match_expression in values
448 )
449 )
450 label_filters.append(label_filter)
452 filters.append(sa.and_(*label_filters))
454 if filters:
455 # This filter is explicitly searching for related resources, so if no other
456 # role is specified, and we're doing any kind of filtering with this filter,
457 # also filter out primary resources (those with an empty role) for any of
458 # these queries
459 if not self.role:
460 filters.append(db.EventResource.resource_role != "")
462 assert self._top_level_filter is not None
463 filters = [db.Event.id.in_(self._top_level_filter.where(*filters))]
465 return filters
468class EventAnyResourceFilter(EventDataFilter):
469 id: Optional[list[str]] = Field(
470 default=None, description="Only include events for resources with these IDs"
471 )
472 id_prefix: Optional[list[str]] = Field(
473 default=None,
474 description=(
475 "Only include events for resources with IDs starting with these prefixes"
476 ),
477 )
478 labels: Optional[ResourceSpecification] = Field(
479 default=None,
480 description="Only include events for related resources with these labels",
481 )
483 def includes(self, event: Event) -> bool:
484 resources = [event.resource] + event.related
485 if not any(self._includes(resource) for resource in resources):
486 return False
487 return True
489 def _includes(self, resource: Resource) -> bool:
490 if self.id:
491 if not any(resource.id == resource_id for resource_id in self.id):
492 return False
494 if self.id_prefix:
495 if not any(resource.id.startswith(prefix) for prefix in self.id_prefix):
496 return False
498 if self.labels:
499 if not self.labels.matches(resource):
500 return False
502 return True
504 @db_injector
505 def build_where_clauses(
506 self, db: PrefectDBInterface
507 ) -> Sequence["ColumnExpressionArgument[bool]"]:
508 filters: list["ColumnExpressionArgument[bool]"] = []
510 if self.id:
511 filters.append(db.EventResource.resource_id.in_(self.id))
513 if self.id_prefix:
514 filters.append(
515 sa.or_(
516 *[
517 db.EventResource.resource_id.startswith(prefix)
518 for prefix in self.id_prefix
519 ]
520 )
521 )
523 if self.labels:
524 label_filters: list[ColumnElement[bool]] = []
525 labels = self.labels.deepcopy()
527 # On the event_resources table, resource_id and resource_role are unpacked
528 # into columns, so we should search there for them
529 if resource_ids := labels.pop("prefect.resource.id", None): 529 ↛ 530line 529 didn't jump to line 530 because the condition on line 529 was never true
530 resource_id_column = db.EventResource.resource_id
531 label_filter = sa.or_(
532 *(
533 _is_like_match_expression(resource_id_column, match_expression)
534 for match_expression in resource_ids
535 )
536 )
537 label_filters.append(label_filter)
539 if roles := labels.pop("prefect.resource.role", None): 539 ↛ 540line 539 didn't jump to line 540 because the condition on line 539 was never true
540 label_filters.append(db.EventResource.resource_role.in_(roles))
542 if labels: 542 ↛ 558line 542 didn't jump to line 558 because the condition on line 542 was always true
543 for _, (label, values) in enumerate(labels.items()):
544 # Empty label value arrays should match nothing
545 if not values: 545 ↛ 546line 545 didn't jump to line 546 because the condition on line 545 was never true
546 label_filters.append(sa.false())
547 continue
549 label_column = db.EventResource.resource[label].astext
550 label_filter = sa.or_(
551 *(
552 _is_like_match_expression(label_column, match_expression)
553 for match_expression in values
554 )
555 )
556 label_filters.append(label_filter)
558 filters.append(sa.and_(*label_filters))
560 if filters:
561 assert self._top_level_filter is not None
562 filters = [db.Event.id.in_(self._top_level_filter.where(*filters))]
564 return filters
567class EventIDFilter(EventDataFilter):
568 id: Optional[list[UUID]] = Field(
569 default=None, description="Only include events with one of these IDs"
570 )
572 def includes(self, event: Event) -> bool:
573 if self.id:
574 if not any(event.id == id for id in self.id):
575 return False
577 return True
579 @db_injector
580 def build_where_clauses(
581 self, db: PrefectDBInterface
582 ) -> Sequence["ColumnExpressionArgument[bool]"]:
583 filters: list["ColumnExpressionArgument[bool]"] = []
585 if self.id:
586 filters.append(db.Event.id.in_(self.id))
588 return filters
591class EventTextFilter(EventDataFilter):
592 """Filter by text search across event content."""
594 query: str = Field(
595 description="Text search query string",
596 examples=[
597 "error",
598 "error -debug",
599 '"connection timeout"',
600 "+required -excluded",
601 ],
602 max_length=200,
603 )
605 def includes(self, event: Event) -> bool:
606 """Check if this text filter includes the given event."""
607 # Parse query into components
608 parsed = parse_text_search_query(self.query)
610 # Build searchable text from all event string fields
611 searchable_text = self._build_searchable_text(event).lower()
613 # Check include terms (OR logic)
614 if parsed.include:
615 include_match = any(
616 term.lower() in searchable_text for term in parsed.include
617 )
618 if not include_match:
619 return False
621 # Check exclude terms (NOT logic)
622 if parsed.exclude:
623 exclude_match = any(
624 term.lower() in searchable_text for term in parsed.exclude
625 )
626 if exclude_match:
627 return False
629 # Check required terms (AND logic - future feature)
630 if parsed.required:
631 required_match = all(
632 term.lower() in searchable_text for term in parsed.required
633 )
634 if not required_match:
635 return False
637 return True
639 def _build_searchable_text(self, event: Event) -> str:
640 """Build searchable text from all relevant event fields"""
641 text_parts = [
642 event.event, # Event type/name
643 ]
645 # Add resource VALUES only (not keys)
646 text_parts.extend(event.resource.root.values())
648 # Add related resource VALUES only (not keys)
649 for related in event.related:
650 text_parts.extend(related.root.values())
652 # Add full payload as string (can include keys and values)
653 text_parts.append(str(event.payload))
655 return " ".join(text_parts)
657 @db_injector
658 def build_where_clauses(
659 self, db: PrefectDBInterface
660 ) -> Sequence["ColumnExpressionArgument[bool]"]:
661 """Build SQLAlchemy WHERE clauses for text search"""
662 filters: list["ColumnExpressionArgument[bool]"] = []
664 if not self.query.strip():
665 return filters
667 parsed = parse_text_search_query(self.query)
669 # Build combined searchable text field
670 # This concatenates event, resource VALUES only, related resource VALUES only, and full payload
671 event_field = db.Event.event
673 payload_field = sa.cast(db.Event.payload, sa.Text)
675 # For now, include resource and related as JSON strings - this will find the data
676 # even though it also includes keys. We can optimize later to extract only values.
677 resource_field = sa.cast(db.Event.resource, sa.Text)
678 related_field = sa.cast(db.Event.related, sa.Text)
680 # Combine all searchable fields
681 searchable_field = sa.func.concat(
682 event_field, " ", resource_field, " ", related_field, " ", payload_field
683 )
685 # Handle include terms (OR logic)
686 if parsed.include:
687 include_conditions = []
688 for term in parsed.include:
689 include_conditions.append(
690 sa.func.lower(searchable_field).contains(term.lower())
691 )
693 if include_conditions: 693 ↛ 697line 693 didn't jump to line 697 because the condition on line 693 was always true
694 filters.append(sa.or_(*include_conditions))
696 # Handle exclude terms (NOT logic)
697 if parsed.exclude:
698 exclude_conditions = []
699 for term in parsed.exclude:
700 exclude_conditions.append(
701 ~sa.func.lower(searchable_field).contains(term.lower())
702 )
704 if exclude_conditions: 704 ↛ 708line 704 didn't jump to line 708 because the condition on line 704 was always true
705 filters.append(sa.and_(*exclude_conditions))
707 # Handle required terms (AND logic - future feature)
708 if parsed.required:
709 required_conditions = []
710 for term in parsed.required:
711 required_conditions.append(
712 sa.func.lower(searchable_field).contains(term.lower())
713 )
715 if required_conditions: 715 ↛ 718line 715 didn't jump to line 718 because the condition on line 715 was always true
716 filters.append(sa.and_(*required_conditions))
718 return filters
721class EventOrder(AutoEnum):
722 ASC = "ASC"
723 DESC = "DESC"
726class EventFilter(EventDataFilter):
727 occurred: EventOccurredFilter = Field(
728 default_factory=lambda: EventOccurredFilter(),
729 description="Filter criteria for when the events occurred",
730 )
731 event: Optional[EventNameFilter] = Field(
732 default=None,
733 description="Filter criteria for the event name",
734 )
735 resource: Optional[EventResourceFilter] = Field(
736 default=None, description="Filter criteria for the resource of the event"
737 )
738 related: Optional[Union[EventRelatedFilter, list[EventRelatedFilter]]] = Field(
739 default=None,
740 description="Filter criteria for the related resources of the event",
741 )
742 any_resource: Optional[
743 Union[EventAnyResourceFilter, list[EventAnyResourceFilter]]
744 ] = Field(
745 default=None,
746 description="Filter criteria for any resource involved in the event",
747 )
748 id: EventIDFilter = Field(
749 default_factory=lambda: EventIDFilter(),
750 description="Filter criteria for the events' ID",
751 )
752 text: Optional[EventTextFilter] = Field(
753 default=None,
754 description="Filter criteria for text search across event content",
755 )
757 order: EventOrder = Field(
758 default=EventOrder.DESC,
759 description="The order to return filtered events",
760 )
762 @db_injector
763 def build_where_clauses(
764 self, db: PrefectDBInterface
765 ) -> Sequence["ColumnExpressionArgument[bool]"]:
766 self._top_level_filter = self._scoped_event_resources(db)
767 result = super().build_where_clauses()
768 self._top_level_filter = None
769 return result
771 def _scoped_event_resources(self, db: PrefectDBInterface) -> Select[tuple[UUID]]:
772 """Returns an event_resources query that is scoped to this filter's scope by occurred."""
773 query = sa.select(db.EventResource.event_id).where(
774 db.EventResource.occurred >= self.occurred.since,
775 db.EventResource.occurred <= self.occurred.until,
776 )
777 return query
779 @property
780 def logical_limit(self) -> int:
781 """The logical limit for this query, which is a maximum number of rows that it
782 _could_ return (regardless of what the caller has requested). May be used as
783 an optimization for DB queries"""
784 if self.id and self.id.id: 784 ↛ 787line 784 didn't jump to line 787 because the condition on line 784 was never true
785 # If we're asking for a specific set of IDs, the most we could get back is
786 # that number of rows
787 return len(self.id.id)
789 return sys.maxsize