import inspect import logging from dataclasses import dataclass, field, replace from enum import Enum, auto from typing import TYPE_CHECKING, Any, Optional 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 from .lib import error from .operations.search_ops import _build_search_xml_query from .operations.search_ops import _collation_to_caldav as collation_to_caldav from .operations.search_ops import _filter_search_results as filter_search_results if TYPE_CHECKING: from .calendarobjectresource import ( CalendarObjectResource as AsyncCalendarObjectResource, ) from .collection import Calendar as AsyncCalendar _icalendar_types = TypesFactory() # Re-export for backward compatibility _collation_to_caldav = collation_to_caldav 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(). 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()" ) ## 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 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: Don't collect a recurrence set in one ical calendar :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): raise NotImplementedError( "full xml given, and it has to be patched to include comp_type" ) objects = [] assert self.event is None and self.todo is None and self.journal is None for comp_class in (Event, Todo, Journal): 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): raise NotImplementedError( "full xml given, and it has to be patched to include comp_type" ) 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): 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 """ return filter_search_results( objects=objects, searcher=self, 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)