import inspect import logging from copy import deepcopy from dataclasses import dataclass, field, replace from datetime import datetime, timezone from enum import Enum, auto from typing import TYPE_CHECKING, Any, Optional from icalendar import Timezone from icalendar.prop import TypesFactory from icalendar_searcher import Searcher from icalendar_searcher.collation import Collation from .calendarobjectresource import CalendarObjectResource, Event, Journal, Todo from .collection import Calendar from .elements import cdav, dav from .lib import error if TYPE_CHECKING: from .calendarobjectresource import ( CalendarObjectResource as AsyncCalendarObjectResource, ) from .collection import Calendar as AsyncCalendar _icalendar_types = TypesFactory() def _collation_to_caldav(collation: Collation, case_sensitive: bool = True) -> str: """Map a Collation enum value to a CalDAV collation identifier string.""" if collation == Collation.SIMPLE: return "i;octet" if case_sensitive else "i;ascii-casemap" elif collation == Collation.UNICODE: return "i;octet" if case_sensitive else "i;unicode-casemap" elif collation == Collation.LOCALE: return "i;ascii-casemap" return "i;octet" # backward-compat alias used inside this module collation_to_caldav = _collation_to_caldav def _filter_search_results( objects: list["CalendarObjectResource"], searcher: "Searcher", post_filter: bool | None = None, split_expanded: bool = True, server_expand: bool = False, ) -> list["CalendarObjectResource"]: """Apply client-side filtering and handle recurrence expansion/splitting.""" if not (post_filter or searcher.expand or (split_expanded and server_expand)): return objects result = [] for o in objects: if searcher.expand or post_filter: try: filtered = searcher.check_component(o, expand_only=not post_filter) except ValueError: filtered = [ x for x in o.icalendar_instance.subcomponents if not isinstance(x, Timezone) ] if not filtered: continue else: filtered = [ x for x in o.icalendar_instance.subcomponents if not isinstance(x, Timezone) ] i = o.icalendar_instance tz_ = [x for x in i.subcomponents if isinstance(x, Timezone)] i.subcomponents = tz_ for comp in filtered: if isinstance(comp, Timezone): continue if split_expanded: new_obj = o.copy(keep_uid=True) new_i = new_obj.icalendar_instance new_i.subcomponents = [] for tz in tz_: new_i.add_component(tz) result.append(new_obj) else: new_i = i new_i.add_component(comp) if not split_expanded: result.append(o) return result # backward-compat alias filter_search_results = _filter_search_results def _build_search_xml_query( searcher: "Searcher", server_expand: bool = False, props: list[Any] | None = None, filters: Any = None, _hacks: str | None = None, ) -> tuple[Any, type | None]: """Build a CalDAV calendar-query XML request body.""" data = cdav.CalendarData() if server_expand: if not searcher.start or not searcher.end: raise error.ReportError("can't expand without a date range") data += cdav.Expand(searcher.start, searcher.end) props_ = [data] if props is None else [data] + list(props) prop = dav.Prop() + props_ vcalendar = cdav.CompFilter("VCALENDAR") comp_filter = None comp_class = searcher.comp_class if filters: filters = deepcopy(filters) if hasattr(filters, "tag") and filters.tag == cdav.CompFilter.tag: comp_filter = filters filters = [] else: filters = [] vNotCompleted = cdav.TextMatch("COMPLETED", negate=True) vNotCancelled = cdav.TextMatch("CANCELLED", negate=True) vNeedsAction = cdav.TextMatch("NEEDS-ACTION") vStatusNotCompleted = cdav.PropFilter("STATUS") + vNotCompleted vStatusNotCancelled = cdav.PropFilter("STATUS") + vNotCancelled vStatusNeedsAction = cdav.PropFilter("STATUS") + vNeedsAction vStatusNotDefined = cdav.PropFilter("STATUS") + cdav.NotDefined() vNoCompleteDate = cdav.PropFilter("COMPLETED") + cdav.NotDefined() if _hacks == "ignore_completed1": filters.extend([vNoCompleteDate, vStatusNotCompleted, vStatusNotCancelled]) elif _hacks == "ignore_completed2": filters.extend([vNoCompleteDate, vStatusNotDefined]) elif _hacks == "ignore_completed3": filters.extend([vStatusNeedsAction]) if searcher.start or searcher.end: filters.append(cdav.TimeRange(searcher.start, searcher.end)) if searcher.alarm_start or searcher.alarm_end: filters.append( cdav.CompFilter("VALARM") + cdav.TimeRange(searcher.alarm_start, searcher.alarm_end) ) comp_mappings = [ ("event", "VEVENT", Event), ("todo", "VTODO", Todo), ("journal", "VJOURNAL", Journal), ] for flag, comp_name, sync_class in comp_mappings: flagged = getattr(searcher, flag, False) if flagged: if comp_class is not None and comp_class is not sync_class: raise error.ConsistencyError( f"inconsistent search parameters - comp_class = {comp_class}, want {sync_class}" ) comp_class = sync_class if comp_filter and comp_filter.attributes.get("name") == comp_name: comp_class = sync_class if ( flag == "todo" and not getattr(searcher, "todo", False) and searcher.include_completed is None ): searcher.include_completed = True setattr(searcher, flag, True) if comp_class is sync_class: if comp_filter: assert comp_filter.attributes.get("name") == comp_name else: comp_filter = cdav.CompFilter(comp_name) setattr(searcher, flag, True) if comp_class and not comp_filter: raise error.ConsistencyError(f"unsupported comp class {comp_class} for search") if _hacks == "no_comp_filter": comp_filter = None for property in searcher._property_operator: if searcher._property_operator[property] == "undef": match = cdav.NotDefined() filters.append(cdav.PropFilter(property.upper()) + match) else: value = searcher._property_filters[property] property_ = property.upper() if property.lower() == "category": property_ = "CATEGORIES" if property.lower() == "categories": values = value.cats else: values = [value] for value in values: if hasattr(value, "to_ical"): value = value.to_ical() collation_str = "i;octet" if ( hasattr(searcher, "_property_collation") and property in searcher._property_collation ): case_sensitive = searcher._property_case_sensitive.get(property, True) collation_str = _collation_to_caldav( searcher._property_collation[property], case_sensitive ) match = cdav.TextMatch(value, collation=collation_str) filters.append(cdav.PropFilter(property_) + match) if comp_filter and filters: comp_filter += filters vcalendar += comp_filter elif comp_filter: vcalendar += comp_filter elif filters: vcalendar += filters filter_elem = cdav.Filter() + vcalendar root = cdav.CalendarQuery() + [prop, filter_elem] return (root, comp_class) def _is_not_defined_supported(features: Any, prop: str) -> bool: """Check if is-not-defined search is supported for a specific property. Checks the property-specific sub-feature (e.g. search.is-not-defined.category or search.is-not-defined.dtend) if one is defined, otherwise falls back to the parent search.is-not-defined feature. The ``categories`` → ``category`` mapping exists because the feature is named after the singular form while the iCalendar property is plural. """ from .compatibility_hints import FeatureSet feature_prop = "category" if prop == "categories" else prop sub_feature = f"search.is-not-defined.{feature_prop}" if sub_feature in FeatureSet.FEATURES: return features.is_supported(sub_feature) return features.is_supported("search.is-not-defined") # Property filter attribute names used for cloning searchers with modified filters _PROPERTY_FILTER_ATTRS = ( "_property_filters", "_property_operator", "_property_locale", "_property_collation", ) class SearchAction(Enum): """Actions yielded by the search generator to be executed by sync/async wrappers.""" RECURSIVE_SEARCH = auto() # (clone, calendar, args) -> do recursive search SEARCH_WITH_COMPTYPES = auto() # (args) -> search with all comp types REQUEST_REPORT = auto() # (xml, comp_class, props) -> make CalDAV request LOAD_OBJECT = auto() # (obj) -> load object data RETURN = auto() # (result) -> return this value @dataclass class CalDAVSearcher(Searcher): """The baseclass (which is generic, and not CalDAV-specific) allows building up a search query search logic. The base class also allows for simple client-side filtering (and at some point in the future, more complex client-side filtering). The CalDAV protocol is difficult, ambigiuous and does not offer all kind of searches. Client-side filtering may be needed to smoothen over differences in how the different servers handle search queries, as well as allowing for more complex searches. A search may be performed by first setting up a CalDAVSearcher, populate it with filter options, and then initiate the search. The recommended approach (as of 3.0) is to create the searcher from a calendar: ``searcher = calendar.searcher(event=True, start=..., end=...)`` ``searcher.add_property_filter("SUMMARY", "meeting")`` ``results = searcher.search()`` For simple searches, the direct method call still works: ``calendar.search(event=True, start=..., end=..., ...)`` The ``todo``, ``event`` and ``journal`` parameters are booleans for filtering the component type. It's currently recommended to set one and only one of them to True, as of 2025-11 there is no guarantees for correct behaviour if setting two of them. Also, if none is given (the default), all objects should be returned - however, all examples in the CalDAV RFC filters things by component, and the different servers do different things when confronted with a search missing component type. With the correct ``compatibility_hints`` (``davclient.features``) configured for the caldav server, the algorithms will ensure correct behaviour. Both the iCalendar standard and the (Cal)DAV standard defines "properties". Make sure not to confuse those. iCalendar properties used for filtering can be passed using ``searcher.add_property_filter``. """ comp_class: Optional["CalendarObjectResource"] = None _explicit_operators: set = field(default_factory=set) _calendar: Optional["Calendar"] = field(default=None, repr=False) def add_property_filter( self, key: str, value: Any, operator: str = None, case_sensitive: bool = True, collation: Collation | None = None, locale: str | None = None, ) -> None: """Adds a filter for some specific iCalendar property. Examples of valid iCalendar properties: SUMMARY, LOCATION, DESCRIPTION, DTSTART, STATUS, CLASS, etc :param key: iCalendar property name (e.g., SUMMARY). Special virtual property "category" (singular) is also supported for substring matching within category names :param value: Filter value, should adhere to the type defined in the RFC :param operator: Comparison operator ("contains", "==", "undef"). If not specified, the server decides the matching behavior (usually substring search per RFC). If explicitly set to "contains" and the server doesn't support substring search, client-side filtering is used (may transfer more data from server). :param case_sensitive: If False, text comparisons are case-insensitive. Note: CalDAV standard case-insensitivity only applies to ASCII characters. :param collation: Advanced collation strategy for text comparison. May not work on all servers. :param locale: Locale string (e.g., "de_DE") for locale-aware collation. Only used with collation=Collation.LOCALE. May not work on all servers. **Supported operators:** * **contains** - substring match (e.g., "rain" matches "Training session" and "Singing in the rain") * **==** - exact match required, enforced client-side * **undef** - matches if property is not defined (value parameter ignored) **Special handling for categories:** - **"categories"** (plural): Exact category name matching - "contains": subset check (all filter categories must be in component) - "==": exact set equality (same categories, order doesn't matter) - Commas in filter values split into multiple categories - **"category"** (singular): Substring matching within category names - "contains": substring match (e.g., "out" matches "outdoor") - "==": exact match to at least one category name - Commas in filter values treated as literal characters Examples: # Case-insensitive search searcher.add_property_filter("SUMMARY", "meeting", case_sensitive=False) # Explicit substring search (guaranteed via client-side if needed) searcher.add_property_filter("LOCATION", "room", operator="contains") # Exact match searcher.add_property_filter("STATUS", "CONFIRMED", operator="==") """ if operator is not None: # Base class lowercases the key, so we need to as well self._explicit_operators.add(key.lower()) super().add_property_filter(key, value, operator, case_sensitive, collation, locale) else: # operator not specified - don't pass it, let base class use default # Don't track as explicit super().add_property_filter( key, value, case_sensitive=case_sensitive, collation=collation, locale=locale, ) def _clone_without_filters( self, filters_to_remove: list[str] | None = None, clear_all_filters: bool = False, ) -> "CalDAVSearcher": """Create a clone of this searcher with specified property filters removed. This is used for compatibility workarounds where we need to remove certain filters from the server query and apply them client-side instead. :param filters_to_remove: List of property filter keys to remove (e.g., ["categories", "category"]) :param clear_all_filters: If True, remove all property filters :return: Cloned searcher with filters removed """ replacements = {} for attr in _PROPERTY_FILTER_ATTRS: if clear_all_filters: replacements[attr] = {} elif filters_to_remove: replacements[attr] = getattr(self, attr).copy() for key in filters_to_remove: replacements[attr].pop(key, None) else: continue clone = replace(self, **replacements) # Update _explicit_operators if we removed specific filters if filters_to_remove and not clear_all_filters: clone._explicit_operators = self._explicit_operators - set(filters_to_remove) return clone def _search_impl( self, calendar: Calendar, server_expand: bool, split_expanded: bool, props: list[cdav.CalendarData] | None, xml: str | None, post_filter: bool | None, _hacks: str | None, ): """Core search implementation as a generator yielding actions. This generator contains all the search logic and yields (action, data) tuples that the caller (sync or async) executes. Results are sent back via .send(). TODO: refactoring beyond readability? Is this sane? If nothing else, the generator data flow better. Possibly this method is too long and should be split up for improved readability Yields: Tuples of (SearchAction, data) where data depends on action type """ if calendar is None: calendar = self._calendar if calendar is None: raise ValueError( "No calendar provided. Either pass a calendar to search() or " "create the searcher via calendar.searcher()" ) ## Workaround for servers where REPORT without a time range only returns ## objects within a sliding window (search.unlimited-time-range: broken). ## Inject a wide time range covering 1970–2126 so that year-2000 test ## objects and other old data are returned. if ( not self.start and not self.end and not (self.expand or server_expand) and not calendar.client.features.is_supported("search.unlimited-time-range") ): clone = replace( self, start=datetime(1970, 1, 1, tzinfo=timezone.utc), end=datetime(2126, 1, 1, tzinfo=timezone.utc), ) result = yield ( SearchAction.RECURSIVE_SEARCH, (clone, calendar, server_expand, split_expanded, props, xml, post_filter, _hacks), ) yield (SearchAction.RETURN, result) return ## Handle servers with broken component-type filtering (e.g., Bedework) comp_type_support = calendar.client.features.is_supported("search.comp-type", str) no_comp_filter = ( (self.comp_class or self.todo or self.event or self.journal) and comp_type_support == "broken" and post_filter is not False ) if no_comp_filter: if not _hacks: _hacks = "no_comp_filter" post_filter = True ## Setting default value for post_filter if post_filter is None and ( (self.todo and not self.include_completed) or self.expand or "categories" in self._property_filters or "category" in self._property_filters or not calendar.client.features.is_supported("search.text.case-sensitive") or not calendar.client.features.is_supported("search.time-range.accurate") ): post_filter = True ## split_expanded should only take effect on expanded data if not self.expand and not server_expand: split_expanded = False ## If the server stores exception VEVENTs as separate calendar objects, client-side ## expansion is unreliable (the master expands without knowing its exceptions, yielding ## duplicate occurrences). Fall back to server-side expansion when it handles exceptions. if ( self.expand and not server_expand and not calendar.client.features.is_supported("save-load.event.recurrences.exception") and calendar.client.features.is_supported("search.recurrences.expanded.exception") ): server_expand = True if self.expand or server_expand: if not self.start or not self.end: raise error.ReportError("can't expand without a date range") ## special compatibility-case for servers that do not support text search at all ## (e.g. purelymail where both i;octet and i;ascii-casemap collations are unsupported). ## Remove all text-value filters and rely on client-side post_filter instead. if ( not calendar.client.features.is_supported("search.text") and self._property_filters and post_filter is not False ): text_filter_props = [ prop for prop, op in self._property_operator.items() if op != "undef" ] if text_filter_props: clone = self._clone_without_filters(text_filter_props) objects = yield ( SearchAction.RECURSIVE_SEARCH, (clone, calendar, server_expand, split_expanded, props, xml, None, None), ) yield ( SearchAction.RETURN, self.filter(objects, post_filter, split_expanded, server_expand), ) return ## special compatbility-case for servers that does not ## support category search properly if ( not calendar.client.features.is_supported("search.text.category") and ("categories" in self._property_filters or "category" in self._property_filters) and post_filter is not False ): clone = self._clone_without_filters(["categories", "category"]) objects = yield ( SearchAction.RECURSIVE_SEARCH, (clone, calendar, server_expand, split_expanded, props, xml, None, None), ) yield ( SearchAction.RETURN, self.filter(objects, post_filter, split_expanded, server_expand), ) return ## special compatibility-case for servers that do not support is-not-defined ## for specific properties (e.g. search.is-not-defined.category or .dtend) if post_filter is not False: undef_props_without_support = [ prop for prop, op in self._property_operator.items() if op == "undef" and not _is_not_defined_supported(calendar.client.features, prop) ] if undef_props_without_support: clone = self._clone_without_filters(undef_props_without_support) objects = yield ( SearchAction.RECURSIVE_SEARCH, (clone, calendar, server_expand, split_expanded, props, xml, None, None), ) yield ( SearchAction.RETURN, self.filter( objects, post_filter=True, split_expanded=split_expanded, server_expand=server_expand, ), ) return ## special compatibility-case for servers that do not support substring search if ( not calendar.client.features.is_supported("search.text.substring") and post_filter is not False ): explicit_contains = [ prop for prop in self._property_operator if prop in self._explicit_operators and self._property_operator[prop] == "contains" ] if explicit_contains: clone = self._clone_without_filters(explicit_contains) objects = yield ( SearchAction.RECURSIVE_SEARCH, (clone, calendar, server_expand, split_expanded, props, xml, None, None), ) yield ( SearchAction.RETURN, self.filter( objects, post_filter=True, split_expanded=split_expanded, server_expand=server_expand, ), ) return ## special compatibility-case for servers that does not ## support combined searches very well if not calendar.client.features.is_supported("search.combined-is-logical-and"): if self.start or self.end: if self._property_filters: clone = self._clone_without_filters(clear_all_filters=True) objects = yield ( SearchAction.RECURSIVE_SEARCH, (clone, calendar, server_expand, split_expanded, props, xml, None, None), ) yield ( SearchAction.RETURN, self.filter(objects, post_filter, split_expanded, server_expand), ) return ## There are two ways to get the pending tasks - we can ## ask the server to filter them out, or we can do it ## client side. ## If the server does not support combined searches, then it's ## safest to do it client-side. ## There is a special case (observed with radicale as of ## 2025-11) where future recurrences of a task does not ## match when doing a server-side filtering, so for this ## case we also do client-side filtering (but the ## "feature" ## search.recurrences.includes-implicit.todo.pending will ## not be supported if the feature ## "search.recurrences.includes-implicit.todo" is not ## supported ... hence the weird or below) ## To be completely sure to get all pending tasks, for all ## server implementations and for all valid icalendar ## objects, we send three different searches to the ## server. This is probably bloated, and may in many ## cases be more expensive than to ask for all tasks. At ## the other hand, for a well-used and well-handled old ## todo-list, there may be a small set of pending tasks ## and heaps of done tasks. ## TODO: consider if not ignore_completed3 is sufficient, ## then the recursive part of the query here is moot, and ## we wouldn't waste so much time on repeated queries if self.todo and self.include_completed is False: clone = replace(self, include_completed=True) clone.include_completed = True ## Why? Isn't this redundant? clone.expand = False if ( not no_comp_filter and calendar.client.features.is_supported("search.text") and calendar.client.features.is_supported("search.combined-is-logical-and") and ( not calendar.client.features.is_supported( "search.recurrences.includes-implicit.todo" ) or calendar.client.features.is_supported( "search.recurrences.includes-implicit.todo.pending" ) ) ): matches = [] for hacks in ("ignore_completed1", "ignore_completed2", "ignore_completed3"): result = yield ( SearchAction.RECURSIVE_SEARCH, (clone, calendar, server_expand, False, props, xml, None, hacks), ) matches.extend(result) else: matches = yield ( SearchAction.RECURSIVE_SEARCH, (clone, calendar, server_expand, False, props, xml, None, _hacks), ) # Deduplicate by URL objects = [] match_set = set() for item in matches: if item.url not in match_set: match_set.add(item.url) objects.append(item) else: orig_xml = xml if not xml or (not isinstance(xml, str) and not xml.tag.endswith("calendar-query")): (xml, self.comp_class) = self.build_search_xml_query( server_expand, props=props, filters=xml, _hacks=_hacks ) if not self.comp_class and not calendar.client.features.is_supported( "search.comp-type.optional" ): if self.include_completed is None: self.include_completed = True result = yield ( SearchAction.SEARCH_WITH_COMPTYPES, (calendar, server_expand, split_expanded, props, orig_xml, _hacks, post_filter), ) yield (SearchAction.RETURN, result) return try: response, objects = yield ( SearchAction.REQUEST_REPORT, (calendar, xml, self.comp_class, props), ) except error.ReportError as err: if ( calendar.client.features.backward_compatibility_mode and not self.comp_class and "400" not in err.reason ): result = yield ( SearchAction.SEARCH_WITH_COMPTYPES, ( calendar, server_expand, split_expanded, props, orig_xml, _hacks, post_filter, ), ) yield (SearchAction.RETURN, result) return raise if not objects and not self.comp_class and _hacks == "insist": result = yield ( SearchAction.SEARCH_WITH_COMPTYPES, (calendar, server_expand, split_expanded, props, orig_xml, _hacks, post_filter), ) yield (SearchAction.RETURN, result) return ## If _hacks=="insist" and still no results despite having text property ## filters, the server may not support text search (e.g. purelymail, ## CCS with i;octet collation). Retry without the text filters and rely ## on client-side post_filter (which is guaranteed True in get_object_by_uid). if not objects and _hacks == "insist" and self._property_filters: non_undef_filters = [ prop for prop, op in self._property_operator.items() if op != "undef" ] if non_undef_filters: clone = self._clone_without_filters(non_undef_filters) result = yield ( SearchAction.RECURSIVE_SEARCH, ( clone, calendar, server_expand, split_expanded, props, orig_xml, None, None, ), ) yield ( SearchAction.RETURN, self.filter(result, post_filter, split_expanded, server_expand), ) return # Post-process: load objects obj2 = [] for o in objects: try: yield (SearchAction.LOAD_OBJECT, o) obj2.append(o) except Exception: logging.error( "Server does not want to reveal details about the calendar object", exc_info=True, ) objects = obj2 # Google sometimes returns empty objects objects = [o for o in objects if o.has_component()] objects = self.filter(objects, post_filter, split_expanded, server_expand) # Partial workaround for https://github.com/python-caldav/caldav/issues/201 for obj in objects: try: yield (SearchAction.LOAD_OBJECT, obj) except Exception: pass yield (SearchAction.RETURN, self.sort(objects)) def search( self, calendar: Calendar = None, server_expand: bool = False, split_expanded: bool = True, props: list[cdav.CalendarData] | None = None, xml: str = None, post_filter=None, _hacks: str = None, ) -> list[CalendarObjectResource]: """Do the search on a CalDAV calendar. Only CalDAV-specific parameters goes to this method. Those parameters are pretty obscure - mostly for power users and internal usage. Unless you have some very special needs, the recommendation is to not pass anything. :param calendar: Calendar to be searched (optional if searcher was created from a calendar via ``calendar.searcher()``) :param server_expand: Ask the CalDAV server to expand recurrences :param split_expanded: Send expanded recurrences as multiple objects :param props: CalDAV properties to send in the query :param xml: XML query to be sent to the server (string or elements) :param post_filter: Do client-side filtering after querying the server :param _hacks: Please don't ask! Make sure not to confuse he CalDAV properties with iCalendar properties. If ``xml`` is given, any other filtering will not be sent to the server. They may still be applied through client-side filtering. (TODO: work in progress) ``post_filter`` takes three values, ``True`` will always filter the results, ``False`` will never filter the results, and the default ``None`` will cause automagics to happen (not implemented yet). Or perhaps I'll just set it to True as default. TODO - make a decision here In the CalDAV protocol, a VCALENDAR object returned from the server may contain only one event/task/journal - but if the object is recurrent, it may contain several recurrences. ``split_expanded`` will split the recurrences into several objects. If you don't know what you're doing, then leave this flag on. """ gen = self._search_impl( calendar, server_expand, split_expanded, props, xml, post_filter, _hacks ) result = None try: action, data = gen.send(result) except StopIteration: return [] while True: try: if action == SearchAction.RECURSIVE_SEARCH: clone, cal, srv_exp, spl_exp, prp, xm, pf, hk = data result = clone.search(cal, srv_exp, spl_exp, prp, xm, pf, hk) elif action == SearchAction.SEARCH_WITH_COMPTYPES: cal, srv_exp, spl_exp, prp, xm, hk, pf = data result = self._search_with_comptypes(cal, srv_exp, spl_exp, prp, xm, hk, pf) elif action == SearchAction.REQUEST_REPORT: cal, xm, comp_cls, prp = data result = cal._request_report_build_resultlist(xm, comp_cls, props=prp) elif action == SearchAction.LOAD_OBJECT: data.load(only_if_unloaded=True) result = None elif action == SearchAction.RETURN: return data action, data = gen.send(result) except StopIteration: return [] def _search_with_comptypes( self, calendar: Calendar, server_expand: bool = False, split_expanded: bool = True, props: list[cdav.CalendarData] | None = None, xml: str = None, _hacks: str = None, post_filter: bool = None, ) -> list[CalendarObjectResource]: """ Internal method - does three searches, one for each comp class (event, journal, todo). """ if xml and (isinstance(xml, str) or "calendar-query" in xml.tag): # Full XML provided – cannot inject a comp-type filter into it. # Fall back to a single REPORT request with the XML as-is; the # server is expected to return whatever comp-types the query # matches, and comp_class detection falls back to auto-detect. _, objects = calendar._request_report_build_resultlist(xml, None, props) return self.sort(objects) objects = [] assert self.event is None and self.todo is None and self.journal is None for comp_class in (Event, Todo, Journal): if not calendar.client.features.is_supported( f"save-load.{comp_class.__name__.lower()}" ): continue clone = replace(self) clone.comp_class = comp_class objects += clone.search( calendar, server_expand, split_expanded, props, xml, post_filter, _hacks ) return self.sort(objects) async def async_search( self, calendar: "AsyncCalendar" = None, server_expand: bool = False, split_expanded: bool = True, props: list[cdav.CalendarData] | None = None, xml: str = None, post_filter=None, _hacks: str = None, ) -> list["AsyncCalendarObjectResource"]: """Async version of search() - does the search on an AsyncCalendar. This method mirrors the sync search() method but uses async HTTP operations. All the same compatibility logic is applied. See the sync search() method for full documentation. """ gen = self._search_impl( calendar, server_expand, split_expanded, props, xml, post_filter, _hacks ) result = None try: action, data = gen.send(result) except StopIteration: return [] while True: try: if action == SearchAction.RECURSIVE_SEARCH: clone, cal, srv_exp, spl_exp, prp, xm, pf, hk = data result = await clone.async_search(cal, srv_exp, spl_exp, prp, xm, pf, hk) elif action == SearchAction.SEARCH_WITH_COMPTYPES: cal, srv_exp, spl_exp, prp, xm, hk, pf = data result = await self._async_search_with_comptypes( cal, srv_exp, spl_exp, prp, xm, hk, pf ) elif action == SearchAction.REQUEST_REPORT: cal, xm, comp_cls, prp = data result = await cal._request_report_build_resultlist(xm, comp_cls, props=prp) elif action == SearchAction.LOAD_OBJECT: load_result = data.load(only_if_unloaded=True) if inspect.isawaitable(load_result): await load_result result = None elif action == SearchAction.RETURN: return data action, data = gen.send(result) except StopIteration: return [] async def _async_search_with_comptypes( self, calendar: "AsyncCalendar", server_expand: bool = False, split_expanded: bool = True, props: list[cdav.CalendarData] | None = None, xml: str = None, _hacks: str = None, post_filter: bool = None, ) -> list["AsyncCalendarObjectResource"]: """ Internal async method - does three searches, one for each comp class. """ if xml and (isinstance(xml, str) or "calendar-query" in xml.tag): # Full XML provided – cannot inject a comp-type filter into it. # Fall back to a single REPORT request with the XML as-is. _, objects = await calendar._request_report_build_resultlist(xml, None, props) return self.sort(objects) objects: list[AsyncCalendarObjectResource] = [] assert self.event is None and self.todo is None and self.journal is None for comp_class in (Event, Todo, Journal): if not calendar.client.features.is_supported( f"save-load.{comp_class.__name__.lower()}" ): continue clone = replace(self) clone.comp_class = comp_class results = await clone.async_search( calendar, server_expand, split_expanded, props, xml, post_filter, _hacks ) objects.extend(results) return self.sort(objects) def filter( self, objects: list[CalendarObjectResource], post_filter: bool | None = None, split_expanded: bool = True, server_expand: bool = False, ) -> list[CalendarObjectResource]: """Apply client-side filtering and handle recurrence expansion/splitting. This method delegates to the operations layer filter_search_results(). See that function for full documentation. :param objects: List of Event/Todo/Journal objects to filter :param post_filter: Whether to apply the searcher's filter logic :param split_expanded: Whether to split recurrence sets into separate objects :param server_expand: Whether server was asked to expand recurrences :return: Filtered and/or split list of CalendarObjectResource objects """ ## icalendar_searcher.Searcher.check_component() mutates include_completed from ## None to the effective default (not self.todo) on first use, and also mutates ## event/journal/todo flags from None to True/False. This breaks reuse of a ## CalDAVSearcher instance across multiple search() calls (issue #650). ## Use a copy with include_completed already resolved so the original is unchanged. searcher = self if self.include_completed is None: searcher = replace(self, include_completed=not self.todo if self.todo else True) return filter_search_results( objects=objects, searcher=searcher, post_filter=post_filter, split_expanded=split_expanded, server_expand=server_expand, ) def build_search_xml_query(self, server_expand=False, props=None, filters=None, _hacks=None): """Build a CalDAV calendar-query XML request. Delegates to the operations layer for the actual XML building. This method updates self.comp_class as a side effect based on the search parameters. :param server_expand: Ask server to expand recurrences :param props: Additional CalDAV properties to request :param filters: Pre-built filter elements (or None to build from self) :param _hacks: Compatibility hack mode :return: Tuple of (xml_element, comp_class) """ xml, comp_class = _build_search_xml_query( searcher=self, server_expand=server_expand, props=props, filters=filters, _hacks=_hacks, ) # Update self.comp_class from the result (side effect for compatibility) self.comp_class = comp_class return (xml, comp_class)