Module exchangelib.services.sync_folder_hierarchy

Expand source code
import abc
import logging

from ..properties import FolderId
from ..util import MNS, TNS, create_element, xml_text_to_value
from .common import EWSAccountService, add_xml_child, folder_ids_element, parse_folder_elem, shape_element

log = logging.getLogger(__name__)


class SyncFolder(EWSAccountService, metaclass=abc.ABCMeta):
    """Base class for SyncFolderHierarchy and SyncFolderItems."""

    element_container_name = f"{{{MNS}}}Changes"
    # Change types
    CREATE = "create"
    UPDATE = "update"
    DELETE = "delete"
    CHANGE_TYPES = (CREATE, UPDATE, DELETE)
    shape_tag = None
    last_in_range_name = None
    change_types_map = {
        f"{{{TNS}}}Create": CREATE,
        f"{{{TNS}}}Update": UPDATE,
        f"{{{TNS}}}Delete": DELETE,
    }

    def __init__(self, *args, **kwargs):
        # These values are reset and set each time call() is consumed
        self.sync_state = None
        self.includes_last_item_in_range = None
        super().__init__(*args, **kwargs)

    def _get_element_container(self, message, name=None):
        self.sync_state = message.find(f"{{{MNS}}}SyncState").text
        log.debug("Sync state is: %s", self.sync_state)
        self.includes_last_item_in_range = xml_text_to_value(message.find(self.last_in_range_name).text, bool)
        log.debug("Includes last item in range: %s", self.includes_last_item_in_range)
        return super()._get_element_container(message=message, name=name)

    def _partial_get_payload(self, folder, shape, additional_fields, sync_state):
        payload = create_element(f"m:{self.SERVICE_NAME}")
        payload.append(
            shape_element(
                tag=self.shape_tag, shape=shape, additional_fields=additional_fields, version=self.account.version
            )
        )
        payload.append(folder_ids_element(folders=[folder], version=self.account.version, tag="m:SyncFolderId"))
        if sync_state:
            add_xml_child(payload, "m:SyncState", sync_state)
        return payload


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

    SERVICE_NAME = "SyncFolderHierarchy"
    shape_tag = "m:FolderShape"
    last_in_range_name = f"{{{MNS}}}IncludesLastFolderInRange"

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.folder = None  # A hack to communicate parsing args to _elems_to_objs()

    def call(self, folder, shape, additional_fields, sync_state):
        self.sync_state = sync_state
        self.folder = folder
        return self._elems_to_objs(
            self._get_elements(
                payload=self.get_payload(
                    folder=folder,
                    shape=shape,
                    additional_fields=additional_fields,
                    sync_state=sync_state,
                )
            )
        )

    def _elem_to_obj(self, elem):
        change_type = self.change_types_map[elem.tag]
        if change_type == self.DELETE:
            folder = FolderId.from_xml(elem=elem.find(FolderId.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 folder types, each with their own tag.
            folder_elem = elem[0]
            folder = parse_folder_elem(elem=folder_elem, folder=self.folder)
        return change_type, folder

    def get_payload(self, folder, shape, additional_fields, sync_state):
        return self._partial_get_payload(
            folder=folder, shape=shape, additional_fields=additional_fields, sync_state=sync_state
        )

Classes

class SyncFolder (*args, **kwargs)

Base class for SyncFolderHierarchy and SyncFolderItems.

Expand source code
class SyncFolder(EWSAccountService, metaclass=abc.ABCMeta):
    """Base class for SyncFolderHierarchy and SyncFolderItems."""

    element_container_name = f"{{{MNS}}}Changes"
    # Change types
    CREATE = "create"
    UPDATE = "update"
    DELETE = "delete"
    CHANGE_TYPES = (CREATE, UPDATE, DELETE)
    shape_tag = None
    last_in_range_name = None
    change_types_map = {
        f"{{{TNS}}}Create": CREATE,
        f"{{{TNS}}}Update": UPDATE,
        f"{{{TNS}}}Delete": DELETE,
    }

    def __init__(self, *args, **kwargs):
        # These values are reset and set each time call() is consumed
        self.sync_state = None
        self.includes_last_item_in_range = None
        super().__init__(*args, **kwargs)

    def _get_element_container(self, message, name=None):
        self.sync_state = message.find(f"{{{MNS}}}SyncState").text
        log.debug("Sync state is: %s", self.sync_state)
        self.includes_last_item_in_range = xml_text_to_value(message.find(self.last_in_range_name).text, bool)
        log.debug("Includes last item in range: %s", self.includes_last_item_in_range)
        return super()._get_element_container(message=message, name=name)

    def _partial_get_payload(self, folder, shape, additional_fields, sync_state):
        payload = create_element(f"m:{self.SERVICE_NAME}")
        payload.append(
            shape_element(
                tag=self.shape_tag, shape=shape, additional_fields=additional_fields, version=self.account.version
            )
        )
        payload.append(folder_ids_element(folders=[folder], version=self.account.version, tag="m:SyncFolderId"))
        if sync_state:
            add_xml_child(payload, "m:SyncState", sync_state)
        return payload

Ancestors

Subclasses

Class variables

var CHANGE_TYPES
var CREATE
var DELETE
var UPDATE
var change_types_map
var element_container_name
var last_in_range_name
var shape_tag

Inherited members

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

    SERVICE_NAME = "SyncFolderHierarchy"
    shape_tag = "m:FolderShape"
    last_in_range_name = f"{{{MNS}}}IncludesLastFolderInRange"

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.folder = None  # A hack to communicate parsing args to _elems_to_objs()

    def call(self, folder, shape, additional_fields, sync_state):
        self.sync_state = sync_state
        self.folder = folder
        return self._elems_to_objs(
            self._get_elements(
                payload=self.get_payload(
                    folder=folder,
                    shape=shape,
                    additional_fields=additional_fields,
                    sync_state=sync_state,
                )
            )
        )

    def _elem_to_obj(self, elem):
        change_type = self.change_types_map[elem.tag]
        if change_type == self.DELETE:
            folder = FolderId.from_xml(elem=elem.find(FolderId.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 folder types, each with their own tag.
            folder_elem = elem[0]
            folder = parse_folder_elem(elem=folder_elem, folder=self.folder)
        return change_type, folder

    def get_payload(self, folder, shape, additional_fields, sync_state):
        return self._partial_get_payload(
            folder=folder, shape=shape, additional_fields=additional_fields, sync_state=sync_state
        )

Ancestors

Class variables

var SERVICE_NAME
var last_in_range_name
var shape_tag

Methods

def call(self, folder, shape, additional_fields, sync_state)
Expand source code
def call(self, folder, shape, additional_fields, sync_state):
    self.sync_state = sync_state
    self.folder = folder
    return self._elems_to_objs(
        self._get_elements(
            payload=self.get_payload(
                folder=folder,
                shape=shape,
                additional_fields=additional_fields,
                sync_state=sync_state,
            )
        )
    )
def get_payload(self, folder, shape, additional_fields, sync_state)
Expand source code
def get_payload(self, folder, shape, additional_fields, sync_state):
    return self._partial_get_payload(
        folder=folder, shape=shape, additional_fields=additional_fields, sync_state=sync_state
    )

Inherited members