Module exchangelib.services.sync_folder_items

Expand source code
from ..errors import InvalidEnumValue, InvalidTypeError
from ..folders import BaseFolder
from ..properties import ItemId
from ..util import MNS, TNS, peek, xml_text_to_value
from .common import add_xml_child, item_ids_element
from .sync_folder_hierarchy import SyncFolder


class SyncFolderItems(SyncFolder):
    """MSDN:
    https://docs.microsoft.com/en-us/exchange/client-developer/web-service-reference/syncfolderitems-operation
    """

    SERVICE_NAME = "SyncFolderItems"
    SYNC_SCOPES = (
        "NormalItems",
        "NormalAndAssociatedItems",
    )
    # Extra change type
    READ_FLAG_CHANGE = "read_flag_change"
    CHANGE_TYPES = SyncFolder.CHANGE_TYPES + (READ_FLAG_CHANGE,)
    shape_tag = "m:ItemShape"
    last_in_range_name = f"{{{MNS}}}IncludesLastItemInRange"
    change_types_map = SyncFolder.change_types_map
    change_types_map[f"{{{TNS}}}ReadFlagChange"] = READ_FLAG_CHANGE

    def call(self, folder, shape, additional_fields, sync_state, ignore, max_changes_returned, sync_scope):
        self.sync_state = sync_state
        if max_changes_returned is None:
            max_changes_returned = 100
        if not isinstance(max_changes_returned, int):
            raise InvalidTypeError("max_changes_returned", max_changes_returned, int)
        if max_changes_returned <= 0:
            raise ValueError(f"'max_changes_returned' {max_changes_returned} must be a positive integer")
        if sync_scope is not None and sync_scope not in self.SYNC_SCOPES:
            raise InvalidEnumValue("sync_scope", sync_scope, self.SYNC_SCOPES)
        return self._elems_to_objs(
            self._get_elements(
                payload=self.get_payload(
                    folder=folder,
                    shape=shape,
                    additional_fields=additional_fields,
                    sync_state=sync_state,
                    ignore=ignore,
                    max_changes_returned=max_changes_returned,
                    sync_scope=sync_scope,
                )
            )
        )

    def _elem_to_obj(self, elem):
        change_type = self.change_types_map[elem.tag]
        if change_type == self.READ_FLAG_CHANGE:
            item = (
                ItemId.from_xml(elem=elem.find(ItemId.response_tag()), account=self.account),
                xml_text_to_value(elem.find(f"{{{TNS}}}IsRead").text, bool),
            )
        elif change_type == self.DELETE:
            item = ItemId.from_xml(elem=elem.find(ItemId.response_tag()), account=self.account)
        else:
            # We can't find() the element because we don't know which tag to look for. The change element can
            # contain multiple item types, each with their own tag.
            item_elem = elem[0]
            item = BaseFolder.item_model_from_tag(item_elem.tag).from_xml(elem=item_elem, account=self.account)
        return change_type, item

    def get_payload(self, folder, shape, additional_fields, sync_state, ignore, max_changes_returned, sync_scope):
        sync_folder_items = self._partial_get_payload(
            folder=folder, shape=shape, additional_fields=additional_fields, sync_state=sync_state
        )
        is_empty, ignore = (True, None) if ignore is None else peek(ignore)
        if not is_empty:
            sync_folder_items.append(item_ids_element(items=ignore, version=self.account.version, tag="m:Ignore"))
        add_xml_child(sync_folder_items, "m:MaxChangesReturned", max_changes_returned)
        if sync_scope:
            add_xml_child(sync_folder_items, "m:SyncScope", sync_scope)
        return sync_folder_items

Classes

class SyncFolderItems (*args, **kwargs)
Expand source code
class SyncFolderItems(SyncFolder):
    """MSDN:
    https://docs.microsoft.com/en-us/exchange/client-developer/web-service-reference/syncfolderitems-operation
    """

    SERVICE_NAME = "SyncFolderItems"
    SYNC_SCOPES = (
        "NormalItems",
        "NormalAndAssociatedItems",
    )
    # Extra change type
    READ_FLAG_CHANGE = "read_flag_change"
    CHANGE_TYPES = SyncFolder.CHANGE_TYPES + (READ_FLAG_CHANGE,)
    shape_tag = "m:ItemShape"
    last_in_range_name = f"{{{MNS}}}IncludesLastItemInRange"
    change_types_map = SyncFolder.change_types_map
    change_types_map[f"{{{TNS}}}ReadFlagChange"] = READ_FLAG_CHANGE

    def call(self, folder, shape, additional_fields, sync_state, ignore, max_changes_returned, sync_scope):
        self.sync_state = sync_state
        if max_changes_returned is None:
            max_changes_returned = 100
        if not isinstance(max_changes_returned, int):
            raise InvalidTypeError("max_changes_returned", max_changes_returned, int)
        if max_changes_returned <= 0:
            raise ValueError(f"'max_changes_returned' {max_changes_returned} must be a positive integer")
        if sync_scope is not None and sync_scope not in self.SYNC_SCOPES:
            raise InvalidEnumValue("sync_scope", sync_scope, self.SYNC_SCOPES)
        return self._elems_to_objs(
            self._get_elements(
                payload=self.get_payload(
                    folder=folder,
                    shape=shape,
                    additional_fields=additional_fields,
                    sync_state=sync_state,
                    ignore=ignore,
                    max_changes_returned=max_changes_returned,
                    sync_scope=sync_scope,
                )
            )
        )

    def _elem_to_obj(self, elem):
        change_type = self.change_types_map[elem.tag]
        if change_type == self.READ_FLAG_CHANGE:
            item = (
                ItemId.from_xml(elem=elem.find(ItemId.response_tag()), account=self.account),
                xml_text_to_value(elem.find(f"{{{TNS}}}IsRead").text, bool),
            )
        elif change_type == self.DELETE:
            item = ItemId.from_xml(elem=elem.find(ItemId.response_tag()), account=self.account)
        else:
            # We can't find() the element because we don't know which tag to look for. The change element can
            # contain multiple item types, each with their own tag.
            item_elem = elem[0]
            item = BaseFolder.item_model_from_tag(item_elem.tag).from_xml(elem=item_elem, account=self.account)
        return change_type, item

    def get_payload(self, folder, shape, additional_fields, sync_state, ignore, max_changes_returned, sync_scope):
        sync_folder_items = self._partial_get_payload(
            folder=folder, shape=shape, additional_fields=additional_fields, sync_state=sync_state
        )
        is_empty, ignore = (True, None) if ignore is None else peek(ignore)
        if not is_empty:
            sync_folder_items.append(item_ids_element(items=ignore, version=self.account.version, tag="m:Ignore"))
        add_xml_child(sync_folder_items, "m:MaxChangesReturned", max_changes_returned)
        if sync_scope:
            add_xml_child(sync_folder_items, "m:SyncScope", sync_scope)
        return sync_folder_items

Ancestors

Class variables

var CHANGE_TYPES
var READ_FLAG_CHANGE
var SERVICE_NAME
var SYNC_SCOPES
var change_types_map
var last_in_range_name
var shape_tag

Methods

def call(self, folder, shape, additional_fields, sync_state, ignore, max_changes_returned, sync_scope)
Expand source code
def call(self, folder, shape, additional_fields, sync_state, ignore, max_changes_returned, sync_scope):
    self.sync_state = sync_state
    if max_changes_returned is None:
        max_changes_returned = 100
    if not isinstance(max_changes_returned, int):
        raise InvalidTypeError("max_changes_returned", max_changes_returned, int)
    if max_changes_returned <= 0:
        raise ValueError(f"'max_changes_returned' {max_changes_returned} must be a positive integer")
    if sync_scope is not None and sync_scope not in self.SYNC_SCOPES:
        raise InvalidEnumValue("sync_scope", sync_scope, self.SYNC_SCOPES)
    return self._elems_to_objs(
        self._get_elements(
            payload=self.get_payload(
                folder=folder,
                shape=shape,
                additional_fields=additional_fields,
                sync_state=sync_state,
                ignore=ignore,
                max_changes_returned=max_changes_returned,
                sync_scope=sync_scope,
            )
        )
    )
def get_payload(self, folder, shape, additional_fields, sync_state, ignore, max_changes_returned, sync_scope)
Expand source code
def get_payload(self, folder, shape, additional_fields, sync_state, ignore, max_changes_returned, sync_scope):
    sync_folder_items = self._partial_get_payload(
        folder=folder, shape=shape, additional_fields=additional_fields, sync_state=sync_state
    )
    is_empty, ignore = (True, None) if ignore is None else peek(ignore)
    if not is_empty:
        sync_folder_items.append(item_ids_element(items=ignore, version=self.account.version, tag="m:Ignore"))
    add_xml_child(sync_folder_items, "m:MaxChangesReturned", max_changes_returned)
    if sync_scope:
        add_xml_child(sync_folder_items, "m:SyncScope", sync_scope)
    return sync_folder_items

Inherited members