Source code for O365.utils.query

from __future__ import annotations

import datetime as dt
from abc import ABC, abstractmethod
from typing import Union, Optional, TYPE_CHECKING, Type, Iterator, Literal, TypeAlias

if TYPE_CHECKING:
    from O365.connection import Protocol

FilterWord: TypeAlias = Union[str, bool, None, dt.date, int, float]


[docs] class QueryBase(ABC): __slots__ = ()
[docs] @abstractmethod def as_params(self) -> dict: pass
[docs] @abstractmethod def render(self) -> str: pass
def __str__(self): return self.__repr__() def __repr__(self): return self.render() @abstractmethod def __and__(self, other): pass @abstractmethod def __or__(self, other): pass
[docs] def get_filter_by_attribute(self, attribute: str) -> Optional[str]: """ Returns a filter value by attribute name. It will match the attribute to the start of each filter attribute and return the first found. :param attribute: the attribute you want to search :return: The value applied to that attribute or None """ search_object: Optional[QueryFilter] = getattr(self, "_filter_instance", None) or getattr(self, "filters", None) if search_object is not None: # CompositeFilter, IterableFilter, ModifierQueryFilter (negate, group) return search_object.get_filter_by_attribute(attribute) search_object: Optional[list[QueryFilter]] = getattr(self, "_filter_instances", None) if search_object is not None: # ChainFilter for filter_obj in search_object: result = filter_obj.get_filter_by_attribute(attribute) if result is not None: return result return None search_object: Optional[str] = getattr(self, "_attribute", None) if search_object is not None: # LogicalFilter or FunctionFilter if search_object.lower().startswith(attribute.lower()): return getattr(self, "_word") return None
[docs] class QueryFilter(QueryBase, ABC): __slots__ = ()
[docs] @abstractmethod def render(self, item_name: Optional[str] = None) -> str: pass
[docs] def as_params(self) -> dict: return {"$filter": self.render()}
def __and__(self, other: Optional[QueryBase]) -> QueryBase: if other is None: return self if isinstance(other, QueryFilter): return ChainFilter("and", [self, other]) elif isinstance(other, OrderByFilter): return CompositeFilter(filters=self, order_by=other) elif isinstance(other, SearchFilter): raise ValueError("Can't mix search with filters or order by clauses.") elif isinstance(other, SelectFilter): return CompositeFilter(filters=self, select=other) elif isinstance(other, ExpandFilter): return CompositeFilter(filters=self, expand=other) else: raise ValueError(f"Can't mix {type(other)} with {type(self)}") def __or__(self, other: QueryFilter) -> ChainFilter: if not isinstance(other, QueryFilter): raise ValueError("Can't chain a non-query filter with and 'or' operator. Use 'and' instead.") return ChainFilter("or", [self, other])
[docs] class OperationQueryFilter(QueryFilter, ABC): __slots__ = ("_operation",)
[docs] def __init__(self, operation: str): self._operation: str = operation
[docs] class LogicalFilter(OperationQueryFilter): __slots__ = ("_operation", "_attribute", "_word")
[docs] def __init__(self, operation: str, attribute: str, word: str): super().__init__(operation) self._attribute: str = attribute self._word: str = word
def _prepare_attribute(self, item_name: str = None) -> str: if item_name: if self._attribute is None: # iteration will occur in the item itself return f"{item_name}" else: return f"{item_name}/{self._attribute}" else: return self._attribute
[docs] def render(self, item_name: Optional[str] = None) -> str: return f"{self._prepare_attribute(item_name)} {self._operation} {self._word}"
[docs] class FunctionFilter(LogicalFilter): __slots__ = ("_operation", "_attribute", "_word")
[docs] def render(self, item_name: Optional[str] = None) -> str: return f"{self._operation}({self._prepare_attribute(item_name)}, {self._word})"
[docs] class IterableFilter(OperationQueryFilter): __slots__ = ("_operation", "_collection", "_item_name", "_filter_instance")
[docs] def __init__(self, operation: str, collection: str, filter_instance: QueryFilter, *, item_name: str = "a"): super().__init__(operation) self._collection: str = collection self._item_name: str = item_name self._filter_instance: QueryFilter = filter_instance
[docs] def render(self, item_name: Optional[str] = None) -> str: # an iterable filter will always ignore external item names filter_instance_render = self._filter_instance.render(item_name=self._item_name) return f"{self._collection}/{self._operation}({self._item_name}: {filter_instance_render})"
[docs] class ChainFilter(OperationQueryFilter): __slots__ = ("_operation", "_filter_instances")
[docs] def __init__(self, operation: str, filter_instances: list[QueryFilter]): assert operation in ("and", "or") super().__init__(operation) self._filter_instances: list[QueryFilter] = filter_instances
[docs] def render(self, item_name: Optional[str] = None) -> str: return f" {self._operation} ".join([fi.render(item_name) for fi in self._filter_instances])
[docs] class ModifierQueryFilter(QueryFilter, ABC): __slots__ = ("_filter_instance",)
[docs] def __init__(self, filter_instance: QueryFilter): self._filter_instance: QueryFilter = filter_instance
[docs] class NegateFilter(ModifierQueryFilter): __slots__ = ("_filter_instance",)
[docs] def render(self, item_name: Optional[str] = None) -> str: return f"not {self._filter_instance.render(item_name=item_name)}"
[docs] class GroupFilter(ModifierQueryFilter): __slots__ = ("_filter_instance",)
[docs] def render(self, item_name: Optional[str] = None) -> str: return f"({self._filter_instance.render(item_name=item_name)})"
[docs] class SearchFilter(QueryBase): __slots__ = ("_search",)
[docs] def __init__(self, word: Optional[Union[str, int, bool]] = None, attribute: Optional[str] = None): if word: if attribute: self._search: str = f"{attribute}:{word}" else: self._search: str = word else: self._search: str = ""
def _combine(self, search_one: str, search_two: str, operator: str = "and"): self._search = f"{search_one} {operator} {search_two}"
[docs] def render(self) -> str: return f'"{self._search}"'
[docs] def as_params(self) -> dict: return {"$search": self.render()}
def __and__(self, other: Optional[QueryBase]) -> QueryBase: if other is None: return self if isinstance(other, SearchFilter): new_search = self.__class__() new_search._combine(self._search, other._search, operator="and") return new_search elif isinstance(other, QueryFilter): raise ValueError("Can't mix search with filters clauses.") elif isinstance(other, OrderByFilter): raise ValueError("Can't mix search with order by clauses.") elif isinstance(other, SelectFilter): return CompositeFilter(search=self, select=other) elif isinstance(other, ExpandFilter): return CompositeFilter(search=self, expand=other) else: raise ValueError(f"Can't mix {type(other)} with {type(self)}") def __or__(self, other: QueryBase) -> SearchFilter: if not isinstance(other, SearchFilter): raise ValueError("Can't chain a non-search filter with and 'or' operator. Use 'and' instead.") new_search = self.__class__() new_search._combine(self._search, other._search, operator="or") return new_search
[docs] class OrderByFilter(QueryBase): __slots__ = ("_orderby",)
[docs] def __init__(self): self._orderby: list[tuple[str, bool]] = []
def _sorted_attributes(self) -> list[str]: return [att for att, asc in self._orderby]
[docs] def add(self, attribute: str, ascending: bool = True) -> None: if not attribute: raise ValueError("Attribute can't be empty") if attribute not in self._sorted_attributes(): self._orderby.append((attribute, ascending))
[docs] def render(self) -> str: return ",".join(f"{att} {'' if asc else 'desc'}".strip() for att, asc in self._orderby)
[docs] def as_params(self) -> dict: return {"$orderby": self.render()}
def __and__(self, other: Optional[QueryBase]) -> QueryBase: if other is None: return self if isinstance(other, OrderByFilter): new_order_by = self.__class__() for att, asc in self._orderby: new_order_by.add(att, asc) for att, asc in other._orderby: new_order_by.add(att, asc) return new_order_by elif isinstance(other, SearchFilter): raise ValueError("Can't mix order by with search clauses.") elif isinstance(other, QueryFilter): return CompositeFilter(order_by=self, filters=other) elif isinstance(other, SelectFilter): return CompositeFilter(order_by=self, select=other) elif isinstance(other, ExpandFilter): return CompositeFilter(order_by=self, expand=other) else: raise ValueError(f"Can't mix {type(other)} with {type(self)}") def __or__(self, other: QueryBase): raise RuntimeError("Orderby clauses are mutually exclusive")
[docs] class ContainerQueryFilter(QueryBase): __slots__ = ("_container", "_keyword")
[docs] def __init__(self, *args: Union[str, tuple[str, SelectFilter]]): self._container: list[Union[str, tuple[str, SelectFilter]]] = list(args) self._keyword: str = ''
[docs] def append(self, item: Union[str, tuple[str, SelectFilter]]) -> None: self._container.append(item)
def __iter__(self) -> Iterator[Union[str, tuple[str, SelectFilter]]]: return iter(self._container) def __contains__(self, attribute: str) -> bool: return attribute in [item[0] if isinstance(item, tuple) else item for item in self._container] def __and__(self, other: Optional[QueryBase]) -> QueryBase: if other is None: return self if (isinstance(other, SelectFilter) and isinstance(self, SelectFilter) ) or (isinstance(other, ExpandFilter) and isinstance(self, ExpandFilter)): new_container = self.__class__(*self) for item in other: if isinstance(item, tuple): attribute = item[0] else: attribute = item if attribute not in new_container: new_container.append(item) return new_container elif isinstance(other, QueryFilter): return CompositeFilter(**{self._keyword: self, "filters": other}) elif isinstance(other, SearchFilter): return CompositeFilter(**{self._keyword: self, "search": other}) elif isinstance(other, OrderByFilter): return CompositeFilter(**{self._keyword: self, "order_by": other}) elif isinstance(other, SelectFilter): return CompositeFilter(**{self._keyword: self, "select": other}) elif isinstance(other, ExpandFilter): return CompositeFilter(**{self._keyword: self, "expand": other}) else: raise ValueError(f"Can't mix {type(other)} with {type(self)}") def __or__(self, other: Optional[QueryBase]): raise RuntimeError("Can't combine multiple composite filters with an 'or' statement. Use 'and' instead.")
[docs] def render(self) -> str: return ",".join(self._container)
[docs] def as_params(self) -> dict: return {f"${self._keyword}": self.render()}
[docs] class SelectFilter(ContainerQueryFilter): __slots__ = ("_container", "_keyword")
[docs] def __init__(self, *args: str): super().__init__(*args) self._keyword: str = "select"
[docs] class ExpandFilter(ContainerQueryFilter): __slots__ = ("_container", "_keyword")
[docs] def __init__(self, *args: Union[str, tuple[str, SelectFilter]]): super().__init__(*args) self._keyword: str = "expand"
[docs] def render(self) -> str: renders = [] for item in self._container: if isinstance(item, tuple): renders.append(f"{item[0]}($select={item[1].render()})") else: renders.append(item) return ",".join(renders)
[docs] class CompositeFilter(QueryBase): """ A Query object that holds all query parameters. """ __slots__ = ("filters", "search", "order_by", "select", "expand")
[docs] def __init__(self, *, filters: Optional[QueryFilter] = None, search: Optional[SearchFilter] = None, order_by: Optional[OrderByFilter] = None, select: Optional[SelectFilter] = None, expand: Optional[ExpandFilter] = None): self.filters: Optional[QueryFilter] = filters self.search: Optional[SearchFilter] = search self.order_by: Optional[OrderByFilter] = order_by self.select: Optional[SelectFilter] = select self.expand: Optional[ExpandFilter] = expand
[docs] def render(self) -> str: return ( f"Filters: {self.filters.render() if self.filters else ''}\n" f"Search: {self.search.render() if self.search else ''}\n" f"OrderBy: {self.order_by.render() if self.order_by else ''}\n" f"Select: {self.select.render() if self.select else ''}\n" f"Expand: {self.expand.render() if self.expand else ''}" )
@property def has_only_filters(self) -> bool: """ Returns true if it only has filters""" return (self.filters is not None and self.search is None and self.order_by is None and self.select is None and self.expand is None)
[docs] def as_params(self) -> dict: params = {} if self.filters: params.update(self.filters.as_params()) if self.search: params.update(self.search.as_params()) if self.order_by: params.update(self.order_by.as_params()) if self.expand: params.update(self.expand.as_params()) if self.select: params.update(self.select.as_params()) return params
def __and__(self, other: Optional[QueryBase]) -> CompositeFilter: """ Combine this CompositeFilter with another QueryBase object """ if other is None: return self nc = CompositeFilter(filters=self.filters, search=self.search, order_by=self.order_by, select=self.select, expand=self.expand) if isinstance(other, QueryFilter): if self.search is not None: raise ValueError("Can't mix search with filters or order by clauses.") nc.filters = nc.filters & other if nc.filters else other elif isinstance(other, OrderByFilter): if self.search is not None: raise ValueError("Can't mix search with filters or order by clauses.") nc.order_by = nc.order_by & other if nc.order_by else other elif isinstance(other, SearchFilter): if self.filters is not None or self.order_by is not None: raise ValueError("Can't mix search with filters or order by clauses.") nc.search = nc.search & other if nc.search else other elif isinstance(other, SelectFilter): nc.select = nc.select & other if nc.select else other elif isinstance(other, ExpandFilter): nc.expand = nc.expand & other if nc.expand else other elif isinstance(other, CompositeFilter): if (self.search and (other.filters or other.order_by) ) or (other.search and (self.filters or self.order_by)): raise ValueError("Can't mix search with filters or order by clauses.") nc.filters = nc.filters & other.filters if nc.filters else other.filters nc.search = nc.search & other.search if nc.search else other.search nc.order_by = nc.order_by & other.order_by if nc.order_by else other.order_by nc.select = nc.select & other.select if nc.select else other.select nc.expand = nc.expand & other.expand if nc.expand else other.expand return nc def __or__(self, other: Optional[QueryBase]) -> CompositeFilter: if isinstance(other, CompositeFilter): if self.has_only_filters and other.has_only_filters: return CompositeFilter(filters=self.filters | other.filters) raise RuntimeError("Can't combine multiple composite filters with an 'or' statement. Use 'and' instead.")
[docs] class QueryBuilder: _attribute_mapping = { "from": "from/emailAddress/address", "to": "toRecipients/emailAddress/address", "start": "start/DateTime", "end": "end/DateTime", "due": "duedatetime/DateTime", "reminder": "reminderdatetime/DateTime", "flag": "flag/flagStatus", "body": "body/content" }
[docs] def __init__(self, protocol: Union[Protocol, Type[Protocol]]): """ Build a query to apply OData filters https://docs.microsoft.com/en-us/graph/query-parameters :param Protocol protocol: protocol to retrieve the timezone from """ self.protocol = protocol() if isinstance(protocol, type) else protocol
def _parse_filter_word(self, word: FilterWord) -> str: """ Converts the word parameter into a string """ if isinstance(word, str): # string must be enclosed in quotes parsed_word = f"'{word}'" elif isinstance(word, bool): # bools are treated as lower case bools parsed_word = str(word).lower() elif word is None: parsed_word = "null" elif isinstance(word, dt.date): if isinstance(word, dt.datetime): if word.tzinfo is None: # if it's a naive datetime, localize the datetime. word = word.replace(tzinfo=self.protocol.timezone) # localize datetime into local tz # convert datetime to iso format parsed_word = f"{word.isoformat()}" else: # other cases like int or float, return as a string. parsed_word = str(word) return parsed_word def _get_attribute_from_mapping(self, attribute: str) -> str: """ Look up the provided attribute into the query builder mapping Applies a conversion to the appropriate casing defined by the protocol. :param attribute: attribute to look up :return: the attribute itself of if found the corresponding complete attribute in the mapping """ mapping = self._attribute_mapping.get(attribute) if mapping: attribute = "/".join( [self.protocol.convert_case(step) for step in mapping.split("/")]) else: attribute = self.protocol.convert_case(attribute) return attribute
[docs] def logical_operation(self, operation: str, attribute: str, word: FilterWord) -> CompositeFilter: """ Apply a logical operation like equals, less than, etc. :param operation: how to combine with a new one :param attribute: attribute to compare word with :param word: value to compare the attribute with :return: a CompositeFilter instance that can render the OData logical operation """ logical_filter = LogicalFilter(operation, self._get_attribute_from_mapping(attribute), self._parse_filter_word(word)) return CompositeFilter(filters=logical_filter)
[docs] def equals(self, attribute: str, word: FilterWord) -> CompositeFilter: """ Return an equals check :param attribute: attribute to compare word with :param word: word to compare with :return: a CompositeFilter instance that can render the OData this logical operation """ return self.logical_operation("eq", attribute, word)
[docs] def unequal(self, attribute: str, word: FilterWord) -> CompositeFilter: """ Return an unequal check :param attribute: attribute to compare word with :param word: word to compare with :return: a CompositeFilter instance that can render the OData this logical operation """ return self.logical_operation("ne", attribute, word)
[docs] def greater(self, attribute: str, word: FilterWord) -> CompositeFilter: """ Return a 'greater than' check :param attribute: attribute to compare word with :param word: word to compare with :return: a CompositeFilter instance that can render the OData this logical operation """ return self.logical_operation("gt", attribute, word)
[docs] def greater_equal(self, attribute: str, word: FilterWord) -> CompositeFilter: """ Return a 'greater than or equal to' check :param attribute: attribute to compare word with :param word: word to compare with :return: a CompositeFilter instance that can render the OData this logical operation """ return self.logical_operation("ge", attribute, word)
[docs] def less(self, attribute: str, word: FilterWord) -> CompositeFilter: """ Return a 'less than' check :param attribute: attribute to compare word with :param word: word to compare with :return: a CompositeFilter instance that can render the OData this logical operation """ return self.logical_operation("lt", attribute, word)
[docs] def less_equal(self, attribute: str, word: FilterWord) -> CompositeFilter: """ Return a 'less than or equal to' check :param attribute: attribute to compare word with :param word: word to compare with :return: a CompositeFilter instance that can render the OData this logical operation """ return self.logical_operation("le", attribute, word)
[docs] def function_operation(self, operation: str, attribute: str, word: FilterWord) -> CompositeFilter: """ Apply a function operation :param operation: function name to operate on attribute :param attribute: the name of the attribute on which to apply the function :param word: value to feed the function :return: a CompositeFilter instance that can render the OData function operation """ function_filter = FunctionFilter(operation, self._get_attribute_from_mapping(attribute), self._parse_filter_word(word)) return CompositeFilter(filters=function_filter)
[docs] def contains(self, attribute: str, word: FilterWord) -> CompositeFilter: """ Adds a contains word check :param attribute: the name of the attribute on which to apply the function :param word: value to feed the function :return: a CompositeFilter instance that can render the OData function operation """ return self.function_operation("contains", attribute, word)
[docs] def startswith(self, attribute: str, word: FilterWord) -> CompositeFilter: """ Adds a startswith word check :param attribute: the name of the attribute on which to apply the function :param word: value to feed the function :return: a CompositeFilter instance that can render the OData function operation """ return self.function_operation("startswith", attribute, word)
[docs] def endswith(self, attribute: str, word: FilterWord) -> CompositeFilter: """ Adds a endswith word check :param attribute: the name of the attribute on which to apply the function :param word: value to feed the function :return: a CompositeFilter instance that can render the OData function operation """ return self.function_operation("endswith", attribute, word)
[docs] def iterable_operation(self, operation: str, collection: str, filter_instance: CompositeFilter, *, item_name: str = "a") -> CompositeFilter: """ Performs the provided filter operation on a collection by iterating over it. For example: .. code-block:: python q.iterable( operation='any', collection='email_addresses', filter_instance=q.equals('address', 'george@best.com') ) will transform to a filter such as: emailAddresses/any(a:a/address eq 'george@best.com') :param operation: the iterable operation name :param collection: the collection to apply the iterable operation on :param filter_instance: a CompositeFilter instance on which you will apply the iterable operation :param item_name: the name of the collection item to be used on the filter_instance :return: a CompositeFilter instance that can render the OData iterable operation """ iterable_filter = IterableFilter(operation, self._get_attribute_from_mapping(collection), filter_instance.filters, item_name=item_name) return CompositeFilter(filters=iterable_filter)
[docs] def any(self, collection: str, filter_instance: CompositeFilter, *, item_name: str = "a") -> CompositeFilter: """ Performs a filter with the OData 'any' keyword on the collection For example: q.any(collection='email_addresses', filter_instance=q.equals('address', 'george@best.com')) will transform to a filter such as: emailAddresses/any(a:a/address eq 'george@best.com') :param collection: the collection to apply the iterable operation on :param filter_instance: a CompositeFilter Instance on which you will apply the iterable operation :param item_name: the name of the collection item to be used on the filter_instance :return: a CompositeFilter instance that can render the OData iterable operation """ return self.iterable_operation("any", collection=collection, filter_instance=filter_instance, item_name=item_name)
[docs] def all(self, collection: str, filter_instance: CompositeFilter, *, item_name: str = "a") -> CompositeFilter: """ Performs a filter with the OData 'all' keyword on the collection For example: q.all(collection='email_addresses', filter_instance=q.equals('address', 'george@best.com')) will transform to a filter such as: emailAddresses/all(a:a/address eq 'george@best.com') :param collection: the collection to apply the iterable operation on :param filter_instance: a CompositeFilter Instance on which you will apply the iterable operation :param item_name: the name of the collection item to be used on the filter_instance :return: a CompositeFilter instance that can render the OData iterable operation """ return self.iterable_operation("all", collection=collection, filter_instance=filter_instance, item_name=item_name)
[docs] @staticmethod def negate(filter_instance: CompositeFilter) -> CompositeFilter: """ Apply a not operator to the provided QueryFilter :param filter_instance: a CompositeFilter instance :return: a CompositeFilter with its filter negated """ negate_filter = NegateFilter(filter_instance=filter_instance.filters) return CompositeFilter(filters=negate_filter)
def _chain(self, operator: str, *filter_instances: CompositeFilter, group: bool = False) -> CompositeFilter: chain = ChainFilter(operation=operator, filter_instances=[fl.filters for fl in filter_instances]) chain = CompositeFilter(filters=chain) if group: return self.group(chain) else: return chain
[docs] def chain_and(self, *filter_instances: CompositeFilter, group: bool = False) -> CompositeFilter: """ Start a chain 'and' operation :param filter_instances: a list of other CompositeFilter you want to combine with the 'and' operation :param group: will group this chain operation if True :return: a CompositeFilter with the filter instances combined with an 'and' operation """ return self._chain("and", *filter_instances, group=group)
[docs] def chain_or(self, *filter_instances: CompositeFilter, group: bool = False) -> CompositeFilter: """ Start a chain 'or' operation. Will automatically apply a grouping. :param filter_instances: a list of other CompositeFilter you want to combine with the 'or' operation :param group: will group this chain operation if True :return: a CompositeFilter with the filter instances combined with an 'or' operation """ return self._chain("or", *filter_instances, group=group)
[docs] @staticmethod def group(filter_instance: CompositeFilter) -> CompositeFilter: """ Applies a grouping to the provided filter_instance """ group_filter = GroupFilter(filter_instance.filters) return CompositeFilter(filters=group_filter)
[docs] def search(self, word: Union[str, int, bool], attribute: Optional[str] = None) -> CompositeFilter: """ Perform a search. Note from graph docs: You can currently search only message and person collections. A $search request returns up to 250 results. You cannot use $filter or $orderby in a search request. :param word: the text to search :param attribute: the attribute to search the word on :return: a CompositeFilter instance that can render the OData search operation """ word = self._parse_filter_word(word) if attribute: attribute = self._get_attribute_from_mapping(attribute) search = SearchFilter(word=word, attribute=attribute) return CompositeFilter(search=search)
[docs] @staticmethod def orderby(*attributes: tuple[Union[str, tuple[str, bool]]]) -> CompositeFilter: """ Returns an 'order by' query param This is useful to order the result set of query from a resource. Note that not all attributes can be sorted and that all resources have different sort capabilities :param attributes: the attributes to orderby :return: a CompositeFilter instance that can render the OData order by operation """ new_order_by = OrderByFilter() for order_by_clause in attributes: if isinstance(order_by_clause, str): new_order_by.add(order_by_clause) elif isinstance(order_by_clause, tuple): new_order_by.add(order_by_clause[0], order_by_clause[1]) else: raise ValueError("Arguments must be attribute strings or tuples" " of attribute strings and ascending booleans") return CompositeFilter(order_by=new_order_by)
[docs] def select(self, *attributes: str) -> CompositeFilter: """ Returns a 'select' query param This is useful to return a limited set of attributes from a resource or return attributes that are not returned by default by the resource. :param attributes: a tuple of attribute names to select :return: a CompositeFilter instance that can render the OData select operation """ select = SelectFilter() for attribute in attributes: attribute = self.protocol.convert_case(attribute) if attribute.lower() in ["meetingmessagetype"]: attribute = f"{self.protocol.keyword_data_store['event_message_type']}/{attribute}" select.append(attribute) return CompositeFilter(select=select)
[docs] def expand(self, relationship: str, select: Optional[CompositeFilter] = None) -> CompositeFilter: """ Returns an 'expand' query param Important: If the 'expand' is a relationship (e.g. "event" or "attachments"), then the ApiComponent using this query should know how to handle the relationship (e.g. Message knows how to handle attachments, and event (if it's an EventMessage). Important: When using expand on multi-value relationships a max of 20 items will be returned. :param relationship: a relationship that will be expanded :param select: a CompositeFilter instance to select attributes on the expanded relationship :return: a CompositeFilter instance that can render the OData expand operation """ expand = ExpandFilter() # this will prepend the event message type tag based on the protocol if relationship == "event": relationship = f"{self.protocol.get_service_keyword('event_message_type')}/event" if select is not None: expand.append((relationship, select.select)) else: expand.append(relationship) return CompositeFilter(expand=expand)