Module exchangelib.folders

Expand source code
from ..properties import DistinguishedFolderId, FolderId
from .base import BaseFolder, Folder
from .collections import FolderCollection
from .known_folders import (
    NON_DELETABLE_FOLDERS,
    AdminAuditLogs,
    AllContacts,
    AllItems,
    ApplicationData,
    ArchiveDeletedItems,
    ArchiveInbox,
    ArchiveMsgFolderRoot,
    ArchiveRecoverableItemsDeletions,
    ArchiveRecoverableItemsPurges,
    ArchiveRecoverableItemsRoot,
    ArchiveRecoverableItemsVersions,
    Audits,
    Birthdays,
    Calendar,
    CalendarLogging,
    CommonViews,
    Companies,
    Conflicts,
    Contacts,
    ConversationHistory,
    ConversationSettings,
    CrawlerData,
    DefaultFoldersChangeHistory,
    DeferredAction,
    DeletedItems,
    Directory,
    DlpPolicyEvaluation,
    Drafts,
    ExchangeSyncData,
    Favorites,
    Files,
    FreeBusyCache,
    FreebusyData,
    Friends,
    GALContacts,
    GraphAnalytics,
    IMContactList,
    Inbox,
    Journal,
    JunkEmail,
    LocalFailures,
    Location,
    MailboxAssociations,
    Messages,
    MsgFolderRoot,
    MyContacts,
    MyContactsExtended,
    NonDeletableFolderMixin,
    Notes,
    OrganizationalContacts,
    Outbox,
    ParkedMessages,
    PassThroughSearchResults,
    PdpProfileV2Secured,
    PeopleCentricConversationBuddies,
    PeopleConnect,
    QuickContacts,
    RecipientCache,
    RecoverableItemsDeletions,
    RecoverableItemsPurges,
    RecoverableItemsRoot,
    RecoverableItemsVersions,
    RecoveryPoints,
    Reminders,
    RSSFeeds,
    Schedule,
    SearchFolders,
    SentItems,
    ServerFailures,
    Sharing,
    Shortcuts,
    Signal,
    SkypeTeamsMessages,
    SmsAndChatsSync,
    SpoolerQueue,
    SwssItems,
    SyncIssues,
    System,
    Tasks,
    TemporarySaves,
    ToDoSearch,
    Views,
    VoiceMail,
    WellknownFolder,
    WorkingSet,
)
from .queryset import DEEP, FOLDER_TRAVERSAL_CHOICES, SHALLOW, SOFT_DELETED, FolderQuerySet, SingleFolderQuerySet
from .roots import ArchiveRoot, PublicFoldersRoot, Root, RootOfHierarchy

__all__ = [
    "AdminAuditLogs",
    "AllContacts",
    "AllItems",
    "ApplicationData",
    "ArchiveDeletedItems",
    "ArchiveInbox",
    "ArchiveMsgFolderRoot",
    "ArchiveRecoverableItemsDeletions",
    "ArchiveRecoverableItemsPurges",
    "ArchiveRecoverableItemsRoot",
    "ArchiveRecoverableItemsVersions",
    "ArchiveRoot",
    "Audits",
    "BaseFolder",
    "Birthdays",
    "Calendar",
    "CalendarLogging",
    "CommonViews",
    "Companies",
    "Conflicts",
    "Contacts",
    "ConversationHistory",
    "ConversationSettings",
    "CrawlerData",
    "DEEP",
    "DefaultFoldersChangeHistory",
    "DeferredAction",
    "DeletedItems",
    "Directory",
    "DistinguishedFolderId",
    "DlpPolicyEvaluation",
    "Drafts",
    "ExchangeSyncData",
    "FOLDER_TRAVERSAL_CHOICES",
    "Favorites",
    "Files",
    "Folder",
    "FolderCollection",
    "FolderId",
    "FolderQuerySet",
    "FreeBusyCache",
    "FreebusyData",
    "Friends",
    "GALContacts",
    "GraphAnalytics",
    "IMContactList",
    "Inbox",
    "Journal",
    "JunkEmail",
    "LocalFailures",
    "Location",
    "MailboxAssociations",
    "Messages",
    "MsgFolderRoot",
    "MyContacts",
    "MyContactsExtended",
    "NON_DELETABLE_FOLDERS",
    "NonDeletableFolderMixin",
    "Notes",
    "OrganizationalContacts",
    "Outbox",
    "ParkedMessages",
    "PassThroughSearchResults",
    "PdpProfileV2Secured",
    "PeopleCentricConversationBuddies",
    "PeopleConnect",
    "PublicFoldersRoot",
    "QuickContacts",
    "RSSFeeds",
    "RecipientCache",
    "RecoverableItemsDeletions",
    "RecoverableItemsPurges",
    "RecoverableItemsRoot",
    "RecoverableItemsVersions",
    "RecoveryPoints",
    "Reminders",
    "Root",
    "RootOfHierarchy",
    "SHALLOW",
    "SOFT_DELETED",
    "Schedule",
    "SearchFolders",
    "SentItems",
    "ServerFailures",
    "Sharing",
    "Shortcuts",
    "Signal",
    "SingleFolderQuerySet",
    "SkypeTeamsMessages",
    "SmsAndChatsSync",
    "SpoolerQueue",
    "SwssItems",
    "SyncIssues",
    "System",
    "Tasks",
    "TemporarySaves",
    "ToDoSearch",
    "Views",
    "VoiceMail",
    "WellknownFolder",
    "WorkingSet",
]

Sub-modules

exchangelib.folders.base
exchangelib.folders.collections
exchangelib.folders.known_folders
exchangelib.folders.queryset
exchangelib.folders.roots

Classes

class AdminAuditLogs (**kwargs)

Base class to use until we have a more specific folder implementation for this folder.

Expand source code
class AdminAuditLogs(WellknownFolder):
    DISTINGUISHED_FOLDER_ID = "adminauditlogs"
    supported_from = EXCHANGE_2013
    get_folder_allowed = False

Ancestors

Class variables

var DISTINGUISHED_FOLDER_ID
var get_folder_allowed
var supported_from

Inherited members

class AllContacts (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class AllContacts(NonDeletableFolderMixin, Contacts):
    CONTAINER_CLASS = "IPF.Note"

    LOCALIZED_NAMES = {
        None: ("AllContacts",),
    }

Ancestors

Class variables

var CONTAINER_CLASS
var LOCALIZED_NAMES

Inherited members

class AllItems (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class AllItems(NonDeletableFolderMixin, Folder):
    CONTAINER_CLASS = "IPF"

    LOCALIZED_NAMES = {
        None: ("AllItems",),
    }

Ancestors

Class variables

var CONTAINER_CLASS
var LOCALIZED_NAMES

Inherited members

class ApplicationData (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class ApplicationData(NonDeletableFolderMixin, Folder):
    CONTAINER_CLASS = "IPM.ApplicationData"

Ancestors

Class variables

var CONTAINER_CLASS

Inherited members

class ArchiveDeletedItems (**kwargs)

Base class to use until we have a more specific folder implementation for this folder.

Expand source code
class ArchiveDeletedItems(WellknownFolder):
    DISTINGUISHED_FOLDER_ID = "archivedeleteditems"
    supported_from = EXCHANGE_2010_SP1

Ancestors

Class variables

var DISTINGUISHED_FOLDER_ID
var supported_from

Inherited members

class ArchiveInbox (**kwargs)

Base class to use until we have a more specific folder implementation for this folder.

Expand source code
class ArchiveInbox(WellknownFolder):
    DISTINGUISHED_FOLDER_ID = "archiveinbox"
    supported_from = EXCHANGE_2013_SP1

Ancestors

Class variables

var DISTINGUISHED_FOLDER_ID
var supported_from

Inherited members

class ArchiveMsgFolderRoot (**kwargs)

Base class to use until we have a more specific folder implementation for this folder.

Expand source code
class ArchiveMsgFolderRoot(WellknownFolder):
    DISTINGUISHED_FOLDER_ID = "archivemsgfolderroot"
    supported_from = EXCHANGE_2010_SP1

Ancestors

Class variables

var DISTINGUISHED_FOLDER_ID
var supported_from

Inherited members

class ArchiveRecoverableItemsDeletions (**kwargs)

Base class to use until we have a more specific folder implementation for this folder.

Expand source code
class ArchiveRecoverableItemsDeletions(WellknownFolder):
    DISTINGUISHED_FOLDER_ID = "archiverecoverableitemsdeletions"
    supported_from = EXCHANGE_2010_SP1

Ancestors

Class variables

var DISTINGUISHED_FOLDER_ID
var supported_from

Inherited members

class ArchiveRecoverableItemsPurges (**kwargs)

Base class to use until we have a more specific folder implementation for this folder.

Expand source code
class ArchiveRecoverableItemsPurges(WellknownFolder):
    DISTINGUISHED_FOLDER_ID = "archiverecoverableitemspurges"
    supported_from = EXCHANGE_2010_SP1

Ancestors

Class variables

var DISTINGUISHED_FOLDER_ID
var supported_from

Inherited members

class ArchiveRecoverableItemsRoot (**kwargs)

Base class to use until we have a more specific folder implementation for this folder.

Expand source code
class ArchiveRecoverableItemsRoot(WellknownFolder):
    DISTINGUISHED_FOLDER_ID = "archiverecoverableitemsroot"
    supported_from = EXCHANGE_2010_SP1

Ancestors

Class variables

var DISTINGUISHED_FOLDER_ID
var supported_from

Inherited members

class ArchiveRecoverableItemsVersions (**kwargs)

Base class to use until we have a more specific folder implementation for this folder.

Expand source code
class ArchiveRecoverableItemsVersions(WellknownFolder):
    DISTINGUISHED_FOLDER_ID = "archiverecoverableitemsversions"
    supported_from = EXCHANGE_2010_SP1

Ancestors

Class variables

var DISTINGUISHED_FOLDER_ID
var supported_from

Inherited members

class ArchiveRoot (**kwargs)

The root of the archive folders hierarchy. Not available on all mailboxes.

Expand source code
class ArchiveRoot(RootOfHierarchy):
    """The root of the archive folders hierarchy. Not available on all mailboxes."""

    DISTINGUISHED_FOLDER_ID = "archiveroot"
    supported_from = EXCHANGE_2010_SP1
    WELLKNOWN_FOLDERS = WELLKNOWN_FOLDERS_IN_ARCHIVE_ROOT

Ancestors

Class variables

var DISTINGUISHED_FOLDER_ID
var WELLKNOWN_FOLDERS
var supported_from

Inherited members

class Audits (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class Audits(NonDeletableFolderMixin, Folder):
    LOCALIZED_NAMES = {
        None: ("Audits",),
    }
    get_folder_allowed = False

Ancestors

Class variables

var LOCALIZED_NAMES
var get_folder_allowed

Inherited members

class BaseFolder (**kwargs)

Base class for all classes that implement a folder.

Expand source code
class BaseFolder(RegisterMixIn, SearchableMixIn, metaclass=EWSMeta):
    """Base class for all classes that implement a folder."""

    ELEMENT_NAME = "Folder"
    NAMESPACE = TNS
    # See https://docs.microsoft.com/en-us/exchange/client-developer/web-service-reference/distinguishedfolderid
    DISTINGUISHED_FOLDER_ID = None
    # Default item type for this folder. See
    # https://docs.microsoft.com/en-us/openspecs/exchange_server_protocols/ms-oxosfld/68a85898-84fe-43c4-b166-4711c13cdd61
    CONTAINER_CLASS = None
    supported_item_models = ITEM_CLASSES  # The Item types that this folder can contain. Default is all
    # Marks the version from which a distinguished folder was introduced. A possibly authoritative source is:
    # https://github.com/OfficeDev/ews-managed-api/blob/master/Enumerations/WellKnownFolderName.cs
    supported_from = None
    # Whether this folder type is allowed with the GetFolder service
    get_folder_allowed = True
    DEFAULT_FOLDER_TRAVERSAL_DEPTH = DEEP_FOLDERS
    DEFAULT_ITEM_TRAVERSAL_DEPTH = SHALLOW_ITEMS
    LOCALIZED_NAMES = {}  # A map of (str)locale: (tuple)localized_folder_names
    ITEM_MODEL_MAP = {cls.response_tag(): cls for cls in ITEM_CLASSES}
    ID_ELEMENT_CLS = FolderId

    _id = IdElementField(field_uri="folder:FolderId", value_cls=ID_ELEMENT_CLS)
    parent_folder_id = EWSElementField(field_uri="folder:ParentFolderId", value_cls=ParentFolderId, is_read_only=True)
    folder_class = CharField(field_uri="folder:FolderClass", is_required_after_save=True)
    name = CharField(field_uri="folder:DisplayName")
    total_count = IntegerField(field_uri="folder:TotalCount", is_read_only=True)
    child_folder_count = IntegerField(field_uri="folder:ChildFolderCount", is_read_only=True)
    unread_count = IntegerField(field_uri="folder:UnreadCount", is_read_only=True)

    __slots__ = "is_distinguished", "item_sync_state", "folder_sync_state"

    # Used to register extended properties
    INSERT_AFTER_FIELD = "child_folder_count"

    def __init__(self, **kwargs):
        self.is_distinguished = kwargs.pop("is_distinguished", False)
        self.item_sync_state = kwargs.pop("item_sync_state", None)
        self.folder_sync_state = kwargs.pop("folder_sync_state", None)
        super().__init__(**kwargs)

    @property
    @abc.abstractmethod
    def account(self):
        """Return the account this folder belongs to"""

    @property
    @abc.abstractmethod
    def root(self):
        """Return the root folder this folder belongs to"""

    @property
    @abc.abstractmethod
    def parent(self):
        """Return the parent folder of this folder"""

    @property
    def is_deletable(self):
        return not self.is_distinguished

    def clean(self, version=None):
        super().clean(version=version)
        # Set a default folder class for new folders. A folder class cannot be changed after saving.
        if self.id is None and self.folder_class is None:
            self.folder_class = self.CONTAINER_CLASS

    @property
    def children(self):
        # It's dangerous to return a generator here because we may then call methods on a child that result in the
        # cache being updated while it's iterated.
        return FolderCollection(account=self.account, folders=self.root.get_children(self))

    @property
    def parts(self):
        parts = [self]
        f = self.parent
        while f:
            parts.insert(0, f)
            f = f.parent
        return parts

    @property
    def absolute(self):
        return "".join(f"/{p.name}" for p in self.parts)

    def _walk(self):
        for c in self.children:
            yield c
            yield from c.walk()

    def walk(self):
        return FolderCollection(account=self.account, folders=self._walk())

    def _glob(self, pattern):
        split_pattern = pattern.rsplit("/", 1)
        head, tail = (split_pattern[0], None) if len(split_pattern) == 1 else split_pattern
        if head == "":
            # We got an absolute path. Restart globbing at root
            yield from self.root.glob(tail or "*")
        elif head == "..":
            # Relative path with reference to parent. Restart globbing at parent
            if not self.parent:
                raise ValueError("Already at top")
            yield from self.parent.glob(tail or "*")
        elif head == "**":
            # Match anything here or in any subfolder at arbitrary depth
            for c in self.walk():
                # fnmatch() may be case-sensitive depending on operating system:
                # force a case-insensitive match since case appears not to
                # matter for folders in Exchange
                if fnmatch(c.name.lower(), (tail or "*").lower()):
                    yield c
        else:
            # Regular pattern
            for c in self.children:
                # See note above on fnmatch() case-sensitivity
                if not fnmatch(c.name.lower(), head.lower()):
                    continue
                if tail is None:
                    yield c
                    continue
                yield from c.glob(tail)

    def glob(self, pattern):
        return FolderCollection(account=self.account, folders=self._glob(pattern))

    def tree(self):
        """Return a string representation of the folder structure of this folder. Example:

        root
        ├── inbox
        │   └── todos
        └── archive
            ├── Last Job
            ├── exchangelib issues
            └── Mom
        """
        tree = f"{self.name}\n"
        children = list(self.children)
        for i, c in enumerate(sorted(children, key=attrgetter("name")), start=1):
            nodes = c.tree().split("\n")
            for j, node in enumerate(nodes, start=1):
                if i != len(children) and j == 1:
                    # Not the last child, but the first node, which is the name of the child
                    tree += f"├── {node}\n"
                elif i != len(children) and j > 1:
                    # Not the last child, and not name of child
                    tree += f"│   {node}\n"
                elif i == len(children) and j == 1:
                    # Not the last child, but the first node, which is the name of the child
                    tree += f"└── {node}\n"
                else:  # Last child, and not name of child
                    tree += f"    {node}\n"
        return tree.strip()

    @classmethod
    def supports_version(cls, version):
        # 'version' is a Version instance, for convenience by callers
        if not cls.supported_from:
            return True
        return version.build >= cls.supported_from

    @property
    def has_distinguished_name(self):
        return self.name and self.DISTINGUISHED_FOLDER_ID and self.name.lower() == self.DISTINGUISHED_FOLDER_ID.lower()

    @classmethod
    def localized_names(cls, locale):
        # Return localized names for a specific locale. If no locale-specific names exist, return the default names,
        # if any.
        return tuple(s.lower() for s in cls.LOCALIZED_NAMES.get(locale, cls.LOCALIZED_NAMES.get(None, [])))

    @staticmethod
    def folder_cls_from_container_class(container_class):
        """Return a reasonable folder class given a container class, e.g. 'IPF.Note'. Don't iterate WELLKNOWN_FOLDERS
        because many folder classes have the same CONTAINER_CLASS.

        :param container_class:
        :return:
        """
        from .known_folders import (
            ApplicationData,
            Calendar,
            Contacts,
            ConversationSettings,
            CrawlerData,
            DlpPolicyEvaluation,
            FreeBusyCache,
            GALContacts,
            Messages,
            RecipientCache,
            RecoveryPoints,
            Reminders,
            RSSFeeds,
            Signal,
            SwssItems,
            Tasks,
        )

        for folder_cls in (
            ApplicationData,
            Calendar,
            Contacts,
            ConversationSettings,
            CrawlerData,
            DlpPolicyEvaluation,
            FreeBusyCache,
            GALContacts,
            Messages,
            RSSFeeds,
            RecipientCache,
            RecoveryPoints,
            Reminders,
            Signal,
            SwssItems,
            Tasks,
        ):
            if folder_cls.CONTAINER_CLASS == container_class:
                return folder_cls
        raise KeyError()

    @classmethod
    def item_model_from_tag(cls, tag):
        try:
            return cls.ITEM_MODEL_MAP[tag]
        except KeyError:
            raise ValueError(f"Item type {tag} was unexpected in a {cls.__name__} folder")

    @classmethod
    def allowed_item_fields(cls, version):
        # Return non-ID fields of all item classes allowed in this folder type
        fields = set()
        for item_model in cls.supported_item_models:
            fields.update(set(item_model.supported_fields(version=version)))
        return fields

    def validate_item_field(self, field, version):
        FolderCollection(account=self.account, folders=[self]).validate_item_field(field=field, version=version)

    def normalize_fields(self, fields):
        # Takes a list of fieldnames, Field or FieldPath objects pointing to item fields. Turns them into FieldPath
        # objects and adds internal timezone fields if necessary. Assume fields are already validated.
        fields = list(fields)
        has_start, has_end = False, False
        for i, field_path in enumerate(fields):
            # Allow both Field and FieldPath instances and string field paths as input
            if isinstance(field_path, str):
                field_path = FieldPath.from_string(field_path=field_path, folder=self)
                fields[i] = field_path
            elif isinstance(field_path, Field):
                field_path = FieldPath(field=field_path)
                fields[i] = field_path
            if field_path.field.name == "start":
                has_start = True
            elif field_path.field.name == "end":
                has_end = True

        # For CalendarItem items, we want to inject internal timezone fields. See also CalendarItem.clean()
        if CalendarItem in self.supported_item_models:
            meeting_tz_field, start_tz_field, end_tz_field = CalendarItem.timezone_fields()
            if self.account.version.build < EXCHANGE_2010:
                if has_start or has_end:
                    fields.append(FieldPath(field=meeting_tz_field))
            else:
                if has_start:
                    fields.append(FieldPath(field=start_tz_field))
                if has_end:
                    fields.append(FieldPath(field=end_tz_field))
        return fields

    @classmethod
    def get_item_field_by_fieldname(cls, fieldname):
        for item_model in cls.supported_item_models:
            try:
                return item_model.get_field_by_fieldname(fieldname)
            except InvalidField:
                pass
        raise InvalidField(f"{fieldname!r} is not a valid field name on {cls.supported_item_models}")

    def get(self, *args, **kwargs):
        return FolderCollection(account=self.account, folders=[self]).get(*args, **kwargs)

    def all(self):
        return FolderCollection(account=self.account, folders=[self]).all()

    def none(self):
        return FolderCollection(account=self.account, folders=[self]).none()

    def filter(self, *args, **kwargs):
        return FolderCollection(account=self.account, folders=[self]).filter(*args, **kwargs)

    def exclude(self, *args, **kwargs):
        return FolderCollection(account=self.account, folders=[self]).exclude(*args, **kwargs)

    def people(self):
        # No point in using a FolderCollection because FindPeople only supports one folder
        return FolderCollection(account=self.account, folders=[self]).people()

    def bulk_create(self, items, *args, **kwargs):
        return self.account.bulk_create(folder=self, items=items, *args, **kwargs)

    def save(self, update_fields=None):
        from ..services import CreateFolder, UpdateFolder

        if self.id is None:
            # New folder
            if update_fields:
                raise ValueError("'update_fields' is only valid for updates")
            res = CreateFolder(account=self.account).get(parent_folder=self.parent, folders=[self])
            self._id = self.ID_ELEMENT_CLS(res.id, res.changekey)
            self.root.add_folder(self)  # Add this folder to the cache
            return self

        # Update folder
        if not update_fields:
            # The fields to update was not specified explicitly. Update all fields where update is possible
            update_fields = []
            for f in self.supported_fields(version=self.account.version):
                if f.is_read_only:
                    # These cannot be changed
                    continue
                if (f.is_required or f.is_required_after_save) and (
                    getattr(self, f.name) is None or (f.is_list and not getattr(self, f.name))
                ):
                    # These are required and cannot be deleted
                    continue
                update_fields.append(f.name)
        res = UpdateFolder(account=self.account).get(folders=[(self, update_fields)])
        folder_id, changekey = res.id, res.changekey
        if self.id != folder_id:
            raise ValueError("ID mismatch")
        # Don't check changekey value. It may not change on no-op updates
        self.changekey = changekey
        self.root.update_folder(self)  # Update the folder in the cache
        return self

    def move(self, to_folder):
        from ..services import MoveFolder

        res = MoveFolder(account=self.account).get(folders=[self], to_folder=to_folder)
        folder_id, changekey = res.id, res.changekey
        if self.id != folder_id:
            raise ValueError("ID mismatch")
        # Don't check changekey value. It may not change on no-op moves
        self.changekey = changekey
        self.parent_folder_id = ParentFolderId(id=to_folder.id, changekey=to_folder.changekey)
        self.root.update_folder(self)  # Update the folder in the cache

    def delete(self, delete_type=HARD_DELETE):
        from ..services import DeleteFolder

        DeleteFolder(account=self.account).get(folders=[self], delete_type=delete_type)
        self.root.remove_folder(self)  # Remove the updated folder from the cache
        self._id = None

    def empty(self, delete_type=HARD_DELETE, delete_sub_folders=False):
        from ..services import EmptyFolder

        EmptyFolder(account=self.account).get(
            folders=[self], delete_type=delete_type, delete_sub_folders=delete_sub_folders
        )
        if delete_sub_folders:
            # We don't know exactly what was deleted, so invalidate the entire folder cache to be safe
            self.root.clear_cache()

    def wipe(self, page_size=None, chunk_size=None, _seen=None, _level=0):
        # Recursively deletes all items in this folder, and all subfolders and their content. Attempts to protect
        # distinguished folders from being deleted. Use with caution!
        from .known_folders import Audits

        _seen = _seen or set()
        if self.id in _seen:
            raise RecursionError(f"We already tried to wipe {self}")
        if _level > 16:
            raise RecursionError(f"Max recursion level reached: {_level}")
        _seen.add(self.id)
        if isinstance(self, Audits):
            # Shortcircuit because this folder can have many items that are all non-deletable
            log.warning("Cannot wipe audits folder %s", self)
            return
        if self.is_distinguished and "recoverableitems" in self.DISTINGUISHED_FOLDER_ID:
            log.warning("Cannot wipe recoverable items folder %s", self)
            return
        log.warning("Wiping %s", self)
        has_distinguished_subfolders = any(f.is_distinguished for f in self.children)
        try:
            if has_distinguished_subfolders:
                self.empty(delete_sub_folders=False)
            else:
                self.empty(delete_sub_folders=True)
        except ErrorRecoverableItemsAccessDenied:
            log.warning("Access denied to %s. Skipping", self)
            return
        except DELETE_FOLDER_ERRORS:
            try:
                if has_distinguished_subfolders:
                    raise  # We already tried this
                self.empty(delete_sub_folders=False)
            except DELETE_FOLDER_ERRORS:
                log.warning("Not allowed to empty %s. Trying to delete items instead", self)
                kwargs = {}
                if page_size is not None:
                    kwargs["page_size"] = page_size
                if chunk_size is not None:
                    kwargs["chunk_size"] = chunk_size
                try:
                    self.all().delete(**kwargs)
                except DELETE_FOLDER_ERRORS:
                    log.warning("Not allowed to delete items in %s", self)
        _level += 1
        for f in self.children:
            f.wipe(page_size=page_size, chunk_size=chunk_size, _seen=_seen, _level=_level)
            # Remove non-distinguished children that are empty and have no subfolders
            if f.is_deletable and not f.children:
                log.warning("Deleting folder %s", f)
                try:
                    f.delete()
                except ErrorDeleteDistinguishedFolder:
                    log.warning("Tried to delete a distinguished folder (%s)", f)

    def test_access(self):
        """Does a simple FindItem to test (read) access to the folder. Maybe the account doesn't exist, maybe the
        service user doesn't have access to the calendar. This will throw the most common errors.
        """
        self.all().exists()
        return True

    @classmethod
    def _kwargs_from_elem(cls, elem, account):
        # Check for 'DisplayName' element before collecting kwargs because because that clears the elements
        has_name_elem = elem.find(cls.get_field_by_fieldname("name").response_tag()) is not None
        kwargs = {f.name: f.from_xml(elem=elem, account=account) for f in cls.FIELDS}
        if has_name_elem and not kwargs["name"]:
            # When we request the 'DisplayName' property, some folders may still be returned with an empty value.
            # Assign a default name to these folders.
            kwargs["name"] = cls.DISTINGUISHED_FOLDER_ID
        return kwargs

    def to_id(self):
        if self.is_distinguished:
            # Don't add the changekey here. When modifying folder content, we usually don't care if others have changed
            # the folder content since we fetched the changekey.
            if self.account:
                return DistinguishedFolderId(
                    id=self.DISTINGUISHED_FOLDER_ID, mailbox=Mailbox(email_address=self.account.primary_smtp_address)
                )
            return DistinguishedFolderId(id=self.DISTINGUISHED_FOLDER_ID)
        if self.id:
            return FolderId(id=self.id, changekey=self.changekey)
        raise ValueError("Must be a distinguished folder or have an ID")

    @classmethod
    def resolve(cls, account, folder):
        # Resolve a single folder
        folders = list(FolderCollection(account=account, folders=[folder]).resolve())
        if not folders:
            raise ErrorFolderNotFound(f"Could not find folder {folder!r}")
        if len(folders) != 1:
            raise ValueError(f"Expected result length 1, but got {folders}")
        f = folders[0]
        if isinstance(f, Exception):
            raise f
        if f.__class__ != cls:
            raise ValueError(f"Expected folder {f!r} to be a {cls} instance")
        return f

    @require_id
    def refresh(self):
        fresh_folder = self.resolve(account=self.account, folder=self)
        if self.id != fresh_folder.id:
            raise ValueError("ID mismatch")
        # Apparently, the changekey may get updated
        for f in self.FIELDS:
            setattr(self, f.name, getattr(fresh_folder, f.name))
        return self

    @require_id
    def get_user_configuration(self, name, properties=None):
        from ..services import GetUserConfiguration
        from ..services.get_user_configuration import ALL

        if properties is None:
            properties = ALL
        return GetUserConfiguration(account=self.account).get(
            user_configuration_name=UserConfigurationNameMNS(name=name, folder=self),
            properties=properties,
        )

    @require_id
    def create_user_configuration(self, name, dictionary=None, xml_data=None, binary_data=None):
        from ..services import CreateUserConfiguration

        user_configuration = UserConfiguration(
            user_configuration_name=UserConfigurationName(name=name, folder=self),
            dictionary=dictionary,
            xml_data=xml_data,
            binary_data=binary_data,
        )
        return CreateUserConfiguration(account=self.account).get(user_configuration=user_configuration)

    @require_id
    def update_user_configuration(self, name, dictionary=None, xml_data=None, binary_data=None):
        from ..services import UpdateUserConfiguration

        user_configuration = UserConfiguration(
            user_configuration_name=UserConfigurationName(name=name, folder=self),
            dictionary=dictionary,
            xml_data=xml_data,
            binary_data=binary_data,
        )
        return UpdateUserConfiguration(account=self.account).get(user_configuration=user_configuration)

    @require_id
    def delete_user_configuration(self, name):
        from ..services import DeleteUserConfiguration

        return DeleteUserConfiguration(account=self.account).get(
            user_configuration_name=UserConfigurationNameMNS(name=name, folder=self)
        )

    @require_id
    def subscribe_to_pull(self, event_types=None, watermark=None, timeout=60):
        """Create a pull subscription.

        :param event_types: List of event types to subscribe to. Possible values defined in SubscribeToPull.EVENT_TYPES
        :param watermark: An event bookmark as returned by some sync services
        :param timeout: Timeout of the subscription, in minutes. Timeout is reset when the server receives a
        GetEvents request for this subscription.
        :return: The subscription ID and a watermark
        """
        from ..services import SubscribeToPull

        if event_types is None:
            event_types = SubscribeToPull.EVENT_TYPES
        return FolderCollection(account=self.account, folders=[self]).subscribe_to_pull(
            event_types=event_types,
            watermark=watermark,
            timeout=timeout,
        )

    @require_id
    def subscribe_to_push(self, callback_url, event_types=None, watermark=None, status_frequency=1):
        """Create a push subscription.

        :param callback_url: A client-defined URL that the server will call
        :param event_types: List of event types to subscribe to. Possible values defined in SubscribeToPush.EVENT_TYPES
        :param watermark: An event bookmark as returned by some sync services
        :param status_frequency: The frequency, in minutes, that the callback URL will be called with.
        :return: The subscription ID and a watermark
        """
        from ..services import SubscribeToPush

        if event_types is None:
            event_types = SubscribeToPush.EVENT_TYPES
        return FolderCollection(account=self.account, folders=[self]).subscribe_to_push(
            event_types=event_types,
            watermark=watermark,
            status_frequency=status_frequency,
            callback_url=callback_url,
        )

    @require_id
    def subscribe_to_streaming(self, event_types=None):
        """Create a streaming subscription.

        :param event_types: List of event types to subscribe to. Possible values defined in SubscribeToPush.EVENT_TYPES
        :return: The subscription ID
        """
        from ..services import SubscribeToStreaming

        if event_types is None:
            event_types = SubscribeToStreaming.EVENT_TYPES
        return FolderCollection(account=self.account, folders=[self]).subscribe_to_streaming(event_types=event_types)

    @require_id
    def pull_subscription(self, **kwargs):
        return PullSubscription(folder=self, **kwargs)

    @require_id
    def push_subscription(self, **kwargs):
        return PushSubscription(folder=self, **kwargs)

    @require_id
    def streaming_subscription(self, **kwargs):
        return StreamingSubscription(folder=self, **kwargs)

    def unsubscribe(self, subscription_id):
        """Unsubscribe. Only applies to pull and streaming notifications.

        :param subscription_id: A subscription ID as acquired by .subscribe_to_[pull|streaming]()
        :return: True

        This method doesn't need the current folder instance, but it makes sense to keep the method along the other
        sync methods.
        """
        from ..services import Unsubscribe

        return Unsubscribe(account=self.account).get(subscription_id=subscription_id)

    def sync_items(self, sync_state=None, only_fields=None, ignore=None, max_changes_returned=None, sync_scope=None):
        """Return all item changes to a folder, as a generator. If sync_state is specified, get all item changes after
        this sync state. After fully consuming the generator, self.item_sync_state will hold the new sync state.

        :param sync_state: The state of the sync. Returned by a successful call to the SyncFolderItems service.
        :param only_fields: A list of string or FieldPath items specifying the fields to fetch. Default to all fields
        :param ignore: A list of Item IDs to ignore in the sync
        :param max_changes_returned: The max number of change
        :param sync_scope: Specify whether to return just items, or items and folder associated information. Possible
           values are specified in SyncFolderItems.SYNC_SCOPES
        :return: A generator of (change_type, item) tuples
        """
        if not sync_state:
            sync_state = self.item_sync_state
        try:
            yield from FolderCollection(account=self.account, folders=[self]).sync_items(
                sync_state=sync_state,
                only_fields=only_fields,
                ignore=ignore,
                max_changes_returned=max_changes_returned,
                sync_scope=sync_scope,
            )
        except SyncCompleted as e:
            # Set the new sync state on the folder instance
            self.item_sync_state = e.sync_state

    def sync_hierarchy(self, sync_state=None, only_fields=None):
        """Return all folder changes to a folder hierarchy, as a generator. If sync_state is specified, get all folder
        changes after this sync state. After fully consuming the generator, self.folder_sync_state will hold the new
        sync state.

        :param sync_state: The state of the sync. Returned by a successful call to the SyncFolderItems service.
        :param only_fields: A list of string or FieldPath items specifying the fields to fetch. Default to all fields
        :return:
        """
        if not sync_state:
            sync_state = self.folder_sync_state
        try:
            yield from FolderCollection(account=self.account, folders=[self]).sync_hierarchy(
                sync_state=sync_state,
                only_fields=only_fields,
            )
        except SyncCompleted as e:
            # Set the new sync state on the folder instance
            self.folder_sync_state = e.sync_state

    def get_events(self, subscription_id, watermark):
        """Get events since the given watermark. Non-blocking.

        :param subscription_id: A subscription ID as acquired by .subscribe_to_[pull|push]()
        :param watermark: Either the watermark from the subscription, or as returned in the last .get_events() call.
        :return: A Notification object containing a list of events

        This method doesn't need the current folder instance, but it makes sense to keep the method along the other
        sync methods.
        """
        from ..services import GetEvents

        svc = GetEvents(account=self.account)
        while True:
            notification = svc.get(subscription_id=subscription_id, watermark=watermark)
            yield notification
            if not notification.more_events:
                break

    def get_streaming_events(self, subscription_id_or_ids, connection_timeout=1, max_notifications_returned=None):
        """Get events since the subscription was created, in streaming mode. This method will block as many minutes
        as specified by 'connection_timeout'.

        :param subscription_id_or_ids: A subscription ID, or list of IDs, as acquired by .subscribe_to_streaming()
        :param connection_timeout: Timeout of the connection, in minutes. The connection is closed after this timeout
        is reached.
        :param max_notifications_returned: If specified, will exit after receiving this number of notifications
        :return: A generator of Notification objects, each containing a list of events

        This method doesn't need the current folder instance, but it makes sense to keep the method along the other
        sync methods.
        """
        from ..services import GetStreamingEvents

        svc = GetStreamingEvents(account=self.account)
        subscription_ids = (
            subscription_id_or_ids
            if is_iterable(subscription_id_or_ids, generators_allowed=True)
            else [subscription_id_or_ids]
        )
        for i, notification in enumerate(
            svc.call(subscription_ids=subscription_ids, connection_timeout=connection_timeout), start=1
        ):
            yield notification
            if max_notifications_returned and i >= max_notifications_returned:
                svc.stop_streaming()
                break

    def __floordiv__(self, other):
        """Support the some_folder // 'child_folder' // 'child_of_child_folder' navigation syntax.

        Works like as __truediv__ but does not touch the folder cache.

        This is useful if the folder hierarchy contains a huge number of folders and you don't want to fetch them all

        :param other:
        :return:
        """
        if other == "..":
            raise ValueError("Cannot get parent without a folder cache")

        if other == ".":
            return self

        # Assume an exact match on the folder name in a shallow search will only return at most one folder
        try:
            return SingleFolderQuerySet(account=self.account, folder=self).depth(SHALLOW_FOLDERS).get(name=other)
        except DoesNotExist:
            raise ErrorFolderNotFound(f"No subfolder with name {other!r}")

    def __truediv__(self, other):
        """Support the some_folder / 'child_folder' / 'child_of_child_folder' navigation syntax."""
        if other == "..":
            if not self.parent:
                raise ValueError("Already at top")
            return self.parent
        if other == ".":
            return self
        for c in self.children:
            if c.name == other:
                return c
        raise ErrorFolderNotFound(f"No subfolder with name {other!r}")

    def __repr__(self):
        return self.__class__.__name__ + repr(
            (
                self.root,
                self.name,
                self.total_count,
                self.unread_count,
                self.child_folder_count,
                self.folder_class,
                self.id,
                self.changekey,
            )
        )

    def __str__(self):
        return f"{self.__class__.__name__} ({self.name})"

Ancestors

Subclasses

Class variables

var CONTAINER_CLASS
var DEFAULT_FOLDER_TRAVERSAL_DEPTH
var DEFAULT_ITEM_TRAVERSAL_DEPTH
var DISTINGUISHED_FOLDER_ID
var ELEMENT_NAME
var FIELDS
var ID_ELEMENT_CLS
var INSERT_AFTER_FIELD
var ITEM_MODEL_MAP
var LOCALIZED_NAMES
var NAMESPACE
var get_folder_allowed
var supported_from
var supported_item_models

Static methods

def allowed_item_fields(version)
Expand source code
@classmethod
def allowed_item_fields(cls, version):
    # Return non-ID fields of all item classes allowed in this folder type
    fields = set()
    for item_model in cls.supported_item_models:
        fields.update(set(item_model.supported_fields(version=version)))
    return fields
def folder_cls_from_container_class(container_class)

Return a reasonable folder class given a container class, e.g. 'IPF.Note'. Don't iterate WELLKNOWN_FOLDERS because many folder classes have the same CONTAINER_CLASS.

:param container_class: :return:

Expand source code
@staticmethod
def folder_cls_from_container_class(container_class):
    """Return a reasonable folder class given a container class, e.g. 'IPF.Note'. Don't iterate WELLKNOWN_FOLDERS
    because many folder classes have the same CONTAINER_CLASS.

    :param container_class:
    :return:
    """
    from .known_folders import (
        ApplicationData,
        Calendar,
        Contacts,
        ConversationSettings,
        CrawlerData,
        DlpPolicyEvaluation,
        FreeBusyCache,
        GALContacts,
        Messages,
        RecipientCache,
        RecoveryPoints,
        Reminders,
        RSSFeeds,
        Signal,
        SwssItems,
        Tasks,
    )

    for folder_cls in (
        ApplicationData,
        Calendar,
        Contacts,
        ConversationSettings,
        CrawlerData,
        DlpPolicyEvaluation,
        FreeBusyCache,
        GALContacts,
        Messages,
        RSSFeeds,
        RecipientCache,
        RecoveryPoints,
        Reminders,
        Signal,
        SwssItems,
        Tasks,
    ):
        if folder_cls.CONTAINER_CLASS == container_class:
            return folder_cls
    raise KeyError()
def get_item_field_by_fieldname(fieldname)
Expand source code
@classmethod
def get_item_field_by_fieldname(cls, fieldname):
    for item_model in cls.supported_item_models:
        try:
            return item_model.get_field_by_fieldname(fieldname)
        except InvalidField:
            pass
    raise InvalidField(f"{fieldname!r} is not a valid field name on {cls.supported_item_models}")
def item_model_from_tag(tag)
Expand source code
@classmethod
def item_model_from_tag(cls, tag):
    try:
        return cls.ITEM_MODEL_MAP[tag]
    except KeyError:
        raise ValueError(f"Item type {tag} was unexpected in a {cls.__name__} folder")
def localized_names(locale)
Expand source code
@classmethod
def localized_names(cls, locale):
    # Return localized names for a specific locale. If no locale-specific names exist, return the default names,
    # if any.
    return tuple(s.lower() for s in cls.LOCALIZED_NAMES.get(locale, cls.LOCALIZED_NAMES.get(None, [])))
def resolve(account, folder)
Expand source code
@classmethod
def resolve(cls, account, folder):
    # Resolve a single folder
    folders = list(FolderCollection(account=account, folders=[folder]).resolve())
    if not folders:
        raise ErrorFolderNotFound(f"Could not find folder {folder!r}")
    if len(folders) != 1:
        raise ValueError(f"Expected result length 1, but got {folders}")
    f = folders[0]
    if isinstance(f, Exception):
        raise f
    if f.__class__ != cls:
        raise ValueError(f"Expected folder {f!r} to be a {cls} instance")
    return f
def supports_version(version)
Expand source code
@classmethod
def supports_version(cls, version):
    # 'version' is a Version instance, for convenience by callers
    if not cls.supported_from:
        return True
    return version.build >= cls.supported_from

Instance variables

var absolute
Expand source code
@property
def absolute(self):
    return "".join(f"/{p.name}" for p in self.parts)
var account

Return the account this folder belongs to

Expand source code
@property
@abc.abstractmethod
def account(self):
    """Return the account this folder belongs to"""
var child_folder_count
var children
Expand source code
@property
def children(self):
    # It's dangerous to return a generator here because we may then call methods on a child that result in the
    # cache being updated while it's iterated.
    return FolderCollection(account=self.account, folders=self.root.get_children(self))
var folder_class
var folder_sync_state

Return an attribute of instance, which is of type owner.

var has_distinguished_name
Expand source code
@property
def has_distinguished_name(self):
    return self.name and self.DISTINGUISHED_FOLDER_ID and self.name.lower() == self.DISTINGUISHED_FOLDER_ID.lower()
var is_deletable
Expand source code
@property
def is_deletable(self):
    return not self.is_distinguished
var is_distinguished

Return an attribute of instance, which is of type owner.

var item_sync_state

Return an attribute of instance, which is of type owner.

var name
var parent

Return the parent folder of this folder

Expand source code
@property
@abc.abstractmethod
def parent(self):
    """Return the parent folder of this folder"""
var parent_folder_id
var parts
Expand source code
@property
def parts(self):
    parts = [self]
    f = self.parent
    while f:
        parts.insert(0, f)
        f = f.parent
    return parts
var root

Return the root folder this folder belongs to

Expand source code
@property
@abc.abstractmethod
def root(self):
    """Return the root folder this folder belongs to"""
var total_count
var unread_count

Methods

def bulk_create(self, items, *args, **kwargs)
Expand source code
def bulk_create(self, items, *args, **kwargs):
    return self.account.bulk_create(folder=self, items=items, *args, **kwargs)
def clean(self, version=None)
Expand source code
def clean(self, version=None):
    super().clean(version=version)
    # Set a default folder class for new folders. A folder class cannot be changed after saving.
    if self.id is None and self.folder_class is None:
        self.folder_class = self.CONTAINER_CLASS
def create_user_configuration(self, name, dictionary=None, xml_data=None, binary_data=None)
Expand source code
@require_id
def create_user_configuration(self, name, dictionary=None, xml_data=None, binary_data=None):
    from ..services import CreateUserConfiguration

    user_configuration = UserConfiguration(
        user_configuration_name=UserConfigurationName(name=name, folder=self),
        dictionary=dictionary,
        xml_data=xml_data,
        binary_data=binary_data,
    )
    return CreateUserConfiguration(account=self.account).get(user_configuration=user_configuration)
def delete(self, delete_type='HardDelete')
Expand source code
def delete(self, delete_type=HARD_DELETE):
    from ..services import DeleteFolder

    DeleteFolder(account=self.account).get(folders=[self], delete_type=delete_type)
    self.root.remove_folder(self)  # Remove the updated folder from the cache
    self._id = None
def delete_user_configuration(self, name)
Expand source code
@require_id
def delete_user_configuration(self, name):
    from ..services import DeleteUserConfiguration

    return DeleteUserConfiguration(account=self.account).get(
        user_configuration_name=UserConfigurationNameMNS(name=name, folder=self)
    )
def empty(self, delete_type='HardDelete', delete_sub_folders=False)
Expand source code
def empty(self, delete_type=HARD_DELETE, delete_sub_folders=False):
    from ..services import EmptyFolder

    EmptyFolder(account=self.account).get(
        folders=[self], delete_type=delete_type, delete_sub_folders=delete_sub_folders
    )
    if delete_sub_folders:
        # We don't know exactly what was deleted, so invalidate the entire folder cache to be safe
        self.root.clear_cache()
def get_events(self, subscription_id, watermark)

Get events since the given watermark. Non-blocking.

:param subscription_id: A subscription ID as acquired by .subscribe_to_pull|push :param watermark: Either the watermark from the subscription, or as returned in the last .get_events() call. :return: A Notification object containing a list of events

This method doesn't need the current folder instance, but it makes sense to keep the method along the other sync methods.

Expand source code
def get_events(self, subscription_id, watermark):
    """Get events since the given watermark. Non-blocking.

    :param subscription_id: A subscription ID as acquired by .subscribe_to_[pull|push]()
    :param watermark: Either the watermark from the subscription, or as returned in the last .get_events() call.
    :return: A Notification object containing a list of events

    This method doesn't need the current folder instance, but it makes sense to keep the method along the other
    sync methods.
    """
    from ..services import GetEvents

    svc = GetEvents(account=self.account)
    while True:
        notification = svc.get(subscription_id=subscription_id, watermark=watermark)
        yield notification
        if not notification.more_events:
            break
def get_streaming_events(self, subscription_id_or_ids, connection_timeout=1, max_notifications_returned=None)

Get events since the subscription was created, in streaming mode. This method will block as many minutes as specified by 'connection_timeout'.

:param subscription_id_or_ids: A subscription ID, or list of IDs, as acquired by .subscribe_to_streaming() :param connection_timeout: Timeout of the connection, in minutes. The connection is closed after this timeout is reached. :param max_notifications_returned: If specified, will exit after receiving this number of notifications :return: A generator of Notification objects, each containing a list of events

This method doesn't need the current folder instance, but it makes sense to keep the method along the other sync methods.

Expand source code
def get_streaming_events(self, subscription_id_or_ids, connection_timeout=1, max_notifications_returned=None):
    """Get events since the subscription was created, in streaming mode. This method will block as many minutes
    as specified by 'connection_timeout'.

    :param subscription_id_or_ids: A subscription ID, or list of IDs, as acquired by .subscribe_to_streaming()
    :param connection_timeout: Timeout of the connection, in minutes. The connection is closed after this timeout
    is reached.
    :param max_notifications_returned: If specified, will exit after receiving this number of notifications
    :return: A generator of Notification objects, each containing a list of events

    This method doesn't need the current folder instance, but it makes sense to keep the method along the other
    sync methods.
    """
    from ..services import GetStreamingEvents

    svc = GetStreamingEvents(account=self.account)
    subscription_ids = (
        subscription_id_or_ids
        if is_iterable(subscription_id_or_ids, generators_allowed=True)
        else [subscription_id_or_ids]
    )
    for i, notification in enumerate(
        svc.call(subscription_ids=subscription_ids, connection_timeout=connection_timeout), start=1
    ):
        yield notification
        if max_notifications_returned and i >= max_notifications_returned:
            svc.stop_streaming()
            break
def get_user_configuration(self, name, properties=None)
Expand source code
@require_id
def get_user_configuration(self, name, properties=None):
    from ..services import GetUserConfiguration
    from ..services.get_user_configuration import ALL

    if properties is None:
        properties = ALL
    return GetUserConfiguration(account=self.account).get(
        user_configuration_name=UserConfigurationNameMNS(name=name, folder=self),
        properties=properties,
    )
def glob(self, pattern)
Expand source code
def glob(self, pattern):
    return FolderCollection(account=self.account, folders=self._glob(pattern))
def move(self, to_folder)
Expand source code
def move(self, to_folder):
    from ..services import MoveFolder

    res = MoveFolder(account=self.account).get(folders=[self], to_folder=to_folder)
    folder_id, changekey = res.id, res.changekey
    if self.id != folder_id:
        raise ValueError("ID mismatch")
    # Don't check changekey value. It may not change on no-op moves
    self.changekey = changekey
    self.parent_folder_id = ParentFolderId(id=to_folder.id, changekey=to_folder.changekey)
    self.root.update_folder(self)  # Update the folder in the cache
def normalize_fields(self, fields)
Expand source code
def normalize_fields(self, fields):
    # Takes a list of fieldnames, Field or FieldPath objects pointing to item fields. Turns them into FieldPath
    # objects and adds internal timezone fields if necessary. Assume fields are already validated.
    fields = list(fields)
    has_start, has_end = False, False
    for i, field_path in enumerate(fields):
        # Allow both Field and FieldPath instances and string field paths as input
        if isinstance(field_path, str):
            field_path = FieldPath.from_string(field_path=field_path, folder=self)
            fields[i] = field_path
        elif isinstance(field_path, Field):
            field_path = FieldPath(field=field_path)
            fields[i] = field_path
        if field_path.field.name == "start":
            has_start = True
        elif field_path.field.name == "end":
            has_end = True

    # For CalendarItem items, we want to inject internal timezone fields. See also CalendarItem.clean()
    if CalendarItem in self.supported_item_models:
        meeting_tz_field, start_tz_field, end_tz_field = CalendarItem.timezone_fields()
        if self.account.version.build < EXCHANGE_2010:
            if has_start or has_end:
                fields.append(FieldPath(field=meeting_tz_field))
        else:
            if has_start:
                fields.append(FieldPath(field=start_tz_field))
            if has_end:
                fields.append(FieldPath(field=end_tz_field))
    return fields
def pull_subscription(self, **kwargs)
Expand source code
@require_id
def pull_subscription(self, **kwargs):
    return PullSubscription(folder=self, **kwargs)
def push_subscription(self, **kwargs)
Expand source code
@require_id
def push_subscription(self, **kwargs):
    return PushSubscription(folder=self, **kwargs)
def refresh(self)
Expand source code
@require_id
def refresh(self):
    fresh_folder = self.resolve(account=self.account, folder=self)
    if self.id != fresh_folder.id:
        raise ValueError("ID mismatch")
    # Apparently, the changekey may get updated
    for f in self.FIELDS:
        setattr(self, f.name, getattr(fresh_folder, f.name))
    return self
def save(self, update_fields=None)
Expand source code
def save(self, update_fields=None):
    from ..services import CreateFolder, UpdateFolder

    if self.id is None:
        # New folder
        if update_fields:
            raise ValueError("'update_fields' is only valid for updates")
        res = CreateFolder(account=self.account).get(parent_folder=self.parent, folders=[self])
        self._id = self.ID_ELEMENT_CLS(res.id, res.changekey)
        self.root.add_folder(self)  # Add this folder to the cache
        return self

    # Update folder
    if not update_fields:
        # The fields to update was not specified explicitly. Update all fields where update is possible
        update_fields = []
        for f in self.supported_fields(version=self.account.version):
            if f.is_read_only:
                # These cannot be changed
                continue
            if (f.is_required or f.is_required_after_save) and (
                getattr(self, f.name) is None or (f.is_list and not getattr(self, f.name))
            ):
                # These are required and cannot be deleted
                continue
            update_fields.append(f.name)
    res = UpdateFolder(account=self.account).get(folders=[(self, update_fields)])
    folder_id, changekey = res.id, res.changekey
    if self.id != folder_id:
        raise ValueError("ID mismatch")
    # Don't check changekey value. It may not change on no-op updates
    self.changekey = changekey
    self.root.update_folder(self)  # Update the folder in the cache
    return self
def streaming_subscription(self, **kwargs)
Expand source code
@require_id
def streaming_subscription(self, **kwargs):
    return StreamingSubscription(folder=self, **kwargs)
def subscribe_to_pull(self, event_types=None, watermark=None, timeout=60)

Create a pull subscription.

:param event_types: List of event types to subscribe to. Possible values defined in SubscribeToPull.EVENT_TYPES :param watermark: An event bookmark as returned by some sync services :param timeout: Timeout of the subscription, in minutes. Timeout is reset when the server receives a GetEvents request for this subscription. :return: The subscription ID and a watermark

Expand source code
@require_id
def subscribe_to_pull(self, event_types=None, watermark=None, timeout=60):
    """Create a pull subscription.

    :param event_types: List of event types to subscribe to. Possible values defined in SubscribeToPull.EVENT_TYPES
    :param watermark: An event bookmark as returned by some sync services
    :param timeout: Timeout of the subscription, in minutes. Timeout is reset when the server receives a
    GetEvents request for this subscription.
    :return: The subscription ID and a watermark
    """
    from ..services import SubscribeToPull

    if event_types is None:
        event_types = SubscribeToPull.EVENT_TYPES
    return FolderCollection(account=self.account, folders=[self]).subscribe_to_pull(
        event_types=event_types,
        watermark=watermark,
        timeout=timeout,
    )
def subscribe_to_push(self, callback_url, event_types=None, watermark=None, status_frequency=1)

Create a push subscription.

:param callback_url: A client-defined URL that the server will call :param event_types: List of event types to subscribe to. Possible values defined in SubscribeToPush.EVENT_TYPES :param watermark: An event bookmark as returned by some sync services :param status_frequency: The frequency, in minutes, that the callback URL will be called with. :return: The subscription ID and a watermark

Expand source code
@require_id
def subscribe_to_push(self, callback_url, event_types=None, watermark=None, status_frequency=1):
    """Create a push subscription.

    :param callback_url: A client-defined URL that the server will call
    :param event_types: List of event types to subscribe to. Possible values defined in SubscribeToPush.EVENT_TYPES
    :param watermark: An event bookmark as returned by some sync services
    :param status_frequency: The frequency, in minutes, that the callback URL will be called with.
    :return: The subscription ID and a watermark
    """
    from ..services import SubscribeToPush

    if event_types is None:
        event_types = SubscribeToPush.EVENT_TYPES
    return FolderCollection(account=self.account, folders=[self]).subscribe_to_push(
        event_types=event_types,
        watermark=watermark,
        status_frequency=status_frequency,
        callback_url=callback_url,
    )
def subscribe_to_streaming(self, event_types=None)

Create a streaming subscription.

:param event_types: List of event types to subscribe to. Possible values defined in SubscribeToPush.EVENT_TYPES :return: The subscription ID

Expand source code
@require_id
def subscribe_to_streaming(self, event_types=None):
    """Create a streaming subscription.

    :param event_types: List of event types to subscribe to. Possible values defined in SubscribeToPush.EVENT_TYPES
    :return: The subscription ID
    """
    from ..services import SubscribeToStreaming

    if event_types is None:
        event_types = SubscribeToStreaming.EVENT_TYPES
    return FolderCollection(account=self.account, folders=[self]).subscribe_to_streaming(event_types=event_types)
def sync_hierarchy(self, sync_state=None, only_fields=None)

Return all folder changes to a folder hierarchy, as a generator. If sync_state is specified, get all folder changes after this sync state. After fully consuming the generator, self.folder_sync_state will hold the new sync state.

:param sync_state: The state of the sync. Returned by a successful call to the SyncFolderItems service. :param only_fields: A list of string or FieldPath items specifying the fields to fetch. Default to all fields :return:

Expand source code
def sync_hierarchy(self, sync_state=None, only_fields=None):
    """Return all folder changes to a folder hierarchy, as a generator. If sync_state is specified, get all folder
    changes after this sync state. After fully consuming the generator, self.folder_sync_state will hold the new
    sync state.

    :param sync_state: The state of the sync. Returned by a successful call to the SyncFolderItems service.
    :param only_fields: A list of string or FieldPath items specifying the fields to fetch. Default to all fields
    :return:
    """
    if not sync_state:
        sync_state = self.folder_sync_state
    try:
        yield from FolderCollection(account=self.account, folders=[self]).sync_hierarchy(
            sync_state=sync_state,
            only_fields=only_fields,
        )
    except SyncCompleted as e:
        # Set the new sync state on the folder instance
        self.folder_sync_state = e.sync_state
def sync_items(self, sync_state=None, only_fields=None, ignore=None, max_changes_returned=None, sync_scope=None)

Return all item changes to a folder, as a generator. If sync_state is specified, get all item changes after this sync state. After fully consuming the generator, self.item_sync_state will hold the new sync state.

:param sync_state: The state of the sync. Returned by a successful call to the SyncFolderItems service. :param only_fields: A list of string or FieldPath items specifying the fields to fetch. Default to all fields :param ignore: A list of Item IDs to ignore in the sync :param max_changes_returned: The max number of change :param sync_scope: Specify whether to return just items, or items and folder associated information. Possible values are specified in SyncFolderItems.SYNC_SCOPES :return: A generator of (change_type, item) tuples

Expand source code
def sync_items(self, sync_state=None, only_fields=None, ignore=None, max_changes_returned=None, sync_scope=None):
    """Return all item changes to a folder, as a generator. If sync_state is specified, get all item changes after
    this sync state. After fully consuming the generator, self.item_sync_state will hold the new sync state.

    :param sync_state: The state of the sync. Returned by a successful call to the SyncFolderItems service.
    :param only_fields: A list of string or FieldPath items specifying the fields to fetch. Default to all fields
    :param ignore: A list of Item IDs to ignore in the sync
    :param max_changes_returned: The max number of change
    :param sync_scope: Specify whether to return just items, or items and folder associated information. Possible
       values are specified in SyncFolderItems.SYNC_SCOPES
    :return: A generator of (change_type, item) tuples
    """
    if not sync_state:
        sync_state = self.item_sync_state
    try:
        yield from FolderCollection(account=self.account, folders=[self]).sync_items(
            sync_state=sync_state,
            only_fields=only_fields,
            ignore=ignore,
            max_changes_returned=max_changes_returned,
            sync_scope=sync_scope,
        )
    except SyncCompleted as e:
        # Set the new sync state on the folder instance
        self.item_sync_state = e.sync_state
def test_access(self)

Does a simple FindItem to test (read) access to the folder. Maybe the account doesn't exist, maybe the service user doesn't have access to the calendar. This will throw the most common errors.

Expand source code
def test_access(self):
    """Does a simple FindItem to test (read) access to the folder. Maybe the account doesn't exist, maybe the
    service user doesn't have access to the calendar. This will throw the most common errors.
    """
    self.all().exists()
    return True
def to_id(self)
Expand source code
def to_id(self):
    if self.is_distinguished:
        # Don't add the changekey here. When modifying folder content, we usually don't care if others have changed
        # the folder content since we fetched the changekey.
        if self.account:
            return DistinguishedFolderId(
                id=self.DISTINGUISHED_FOLDER_ID, mailbox=Mailbox(email_address=self.account.primary_smtp_address)
            )
        return DistinguishedFolderId(id=self.DISTINGUISHED_FOLDER_ID)
    if self.id:
        return FolderId(id=self.id, changekey=self.changekey)
    raise ValueError("Must be a distinguished folder or have an ID")
def tree(self)

Return a string representation of the folder structure of this folder. Example:

root ├── inbox │ └── todos └── archive ├── Last Job ├── exchangelib issues └── Mom

Expand source code
def tree(self):
    """Return a string representation of the folder structure of this folder. Example:

    root
    ├── inbox
    │   └── todos
    └── archive
        ├── Last Job
        ├── exchangelib issues
        └── Mom
    """
    tree = f"{self.name}\n"
    children = list(self.children)
    for i, c in enumerate(sorted(children, key=attrgetter("name")), start=1):
        nodes = c.tree().split("\n")
        for j, node in enumerate(nodes, start=1):
            if i != len(children) and j == 1:
                # Not the last child, but the first node, which is the name of the child
                tree += f"├── {node}\n"
            elif i != len(children) and j > 1:
                # Not the last child, and not name of child
                tree += f"│   {node}\n"
            elif i == len(children) and j == 1:
                # Not the last child, but the first node, which is the name of the child
                tree += f"└── {node}\n"
            else:  # Last child, and not name of child
                tree += f"    {node}\n"
    return tree.strip()
def unsubscribe(self, subscription_id)

Unsubscribe. Only applies to pull and streaming notifications.

:param subscription_id: A subscription ID as acquired by .subscribe_to_pull|streaming :return: True

This method doesn't need the current folder instance, but it makes sense to keep the method along the other sync methods.

Expand source code
def unsubscribe(self, subscription_id):
    """Unsubscribe. Only applies to pull and streaming notifications.

    :param subscription_id: A subscription ID as acquired by .subscribe_to_[pull|streaming]()
    :return: True

    This method doesn't need the current folder instance, but it makes sense to keep the method along the other
    sync methods.
    """
    from ..services import Unsubscribe

    return Unsubscribe(account=self.account).get(subscription_id=subscription_id)
def update_user_configuration(self, name, dictionary=None, xml_data=None, binary_data=None)
Expand source code
@require_id
def update_user_configuration(self, name, dictionary=None, xml_data=None, binary_data=None):
    from ..services import UpdateUserConfiguration

    user_configuration = UserConfiguration(
        user_configuration_name=UserConfigurationName(name=name, folder=self),
        dictionary=dictionary,
        xml_data=xml_data,
        binary_data=binary_data,
    )
    return UpdateUserConfiguration(account=self.account).get(user_configuration=user_configuration)
def validate_item_field(self, field, version)
Expand source code
def validate_item_field(self, field, version):
    FolderCollection(account=self.account, folders=[self]).validate_item_field(field=field, version=version)
def walk(self)
Expand source code
def walk(self):
    return FolderCollection(account=self.account, folders=self._walk())
def wipe(self, page_size=None, chunk_size=None)
Expand source code
def wipe(self, page_size=None, chunk_size=None, _seen=None, _level=0):
    # Recursively deletes all items in this folder, and all subfolders and their content. Attempts to protect
    # distinguished folders from being deleted. Use with caution!
    from .known_folders import Audits

    _seen = _seen or set()
    if self.id in _seen:
        raise RecursionError(f"We already tried to wipe {self}")
    if _level > 16:
        raise RecursionError(f"Max recursion level reached: {_level}")
    _seen.add(self.id)
    if isinstance(self, Audits):
        # Shortcircuit because this folder can have many items that are all non-deletable
        log.warning("Cannot wipe audits folder %s", self)
        return
    if self.is_distinguished and "recoverableitems" in self.DISTINGUISHED_FOLDER_ID:
        log.warning("Cannot wipe recoverable items folder %s", self)
        return
    log.warning("Wiping %s", self)
    has_distinguished_subfolders = any(f.is_distinguished for f in self.children)
    try:
        if has_distinguished_subfolders:
            self.empty(delete_sub_folders=False)
        else:
            self.empty(delete_sub_folders=True)
    except ErrorRecoverableItemsAccessDenied:
        log.warning("Access denied to %s. Skipping", self)
        return
    except DELETE_FOLDER_ERRORS:
        try:
            if has_distinguished_subfolders:
                raise  # We already tried this
            self.empty(delete_sub_folders=False)
        except DELETE_FOLDER_ERRORS:
            log.warning("Not allowed to empty %s. Trying to delete items instead", self)
            kwargs = {}
            if page_size is not None:
                kwargs["page_size"] = page_size
            if chunk_size is not None:
                kwargs["chunk_size"] = chunk_size
            try:
                self.all().delete(**kwargs)
            except DELETE_FOLDER_ERRORS:
                log.warning("Not allowed to delete items in %s", self)
    _level += 1
    for f in self.children:
        f.wipe(page_size=page_size, chunk_size=chunk_size, _seen=_seen, _level=_level)
        # Remove non-distinguished children that are empty and have no subfolders
        if f.is_deletable and not f.children:
            log.warning("Deleting folder %s", f)
            try:
                f.delete()
            except ErrorDeleteDistinguishedFolder:
                log.warning("Tried to delete a distinguished folder (%s)", f)

Inherited members

class Birthdays (**kwargs)
Expand source code
class Birthdays(Folder):
    CONTAINER_CLASS = "IPF.Appointment.Birthday"
    LOCALIZED_NAMES = {
        None: ("Birthdays",),
        "da_DK": ("Fødselsdage",),
    }

Ancestors

Class variables

var CONTAINER_CLASS
var LOCALIZED_NAMES

Inherited members

class Calendar (**kwargs)

An interface for the Exchange calendar.

Expand source code
class Calendar(Folder):
    """An interface for the Exchange calendar."""

    DISTINGUISHED_FOLDER_ID = "calendar"
    CONTAINER_CLASS = "IPF.Appointment"
    supported_item_models = (CalendarItem,)

    LOCALIZED_NAMES = {
        "da_DK": ("Kalender",),
        "de_DE": ("Kalender",),
        "en_US": ("Calendar",),
        "es_ES": ("Calendario",),
        "fr_CA": ("Calendrier",),
        "nl_NL": ("Agenda",),
        "ru_RU": ("Календарь",),
        "sv_SE": ("Kalender",),
        "zh_CN": ("日历",),
    }

    def view(self, *args, **kwargs):
        return FolderCollection(account=self.account, folders=[self]).view(*args, **kwargs)

Ancestors

Class variables

var CONTAINER_CLASS
var DISTINGUISHED_FOLDER_ID
var LOCALIZED_NAMES
var supported_item_models

Methods

def view(self, *args, **kwargs)
Expand source code
def view(self, *args, **kwargs):
    return FolderCollection(account=self.account, folders=[self]).view(*args, **kwargs)

Inherited members

class CalendarLogging (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class CalendarLogging(NonDeletableFolderMixin, Folder):
    LOCALIZED_NAMES = {
        None: ("Calendar Logging",),
    }

Ancestors

Class variables

var LOCALIZED_NAMES

Inherited members

class CommonViews (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class CommonViews(NonDeletableFolderMixin, Folder):
    DEFAULT_ITEM_TRAVERSAL_DEPTH = ASSOCIATED
    LOCALIZED_NAMES = {
        None: ("Common Views",),
    }

Ancestors

Class variables

var DEFAULT_ITEM_TRAVERSAL_DEPTH
var LOCALIZED_NAMES

Inherited members

class Companies (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class Companies(NonDeletableFolderMixin, Contacts):
    DISTINGUISHED_FOLDER_ID = None
    CONTAINTER_CLASS = "IPF.Contact.Company"
    LOCALIZED_NAMES = {
        None: ("Companies",),
        "da_DK": ("Firmaer",),
    }

Ancestors

Class variables

var CONTAINTER_CLASS
var DISTINGUISHED_FOLDER_ID
var LOCALIZED_NAMES

Inherited members

class Conflicts (**kwargs)

Base class to use until we have a more specific folder implementation for this folder.

Expand source code
class Conflicts(WellknownFolder):
    DISTINGUISHED_FOLDER_ID = "conflicts"
    supported_from = EXCHANGE_2013

Ancestors

Class variables

var DISTINGUISHED_FOLDER_ID
var supported_from

Inherited members

class Contacts (**kwargs)
Expand source code
class Contacts(Folder):
    DISTINGUISHED_FOLDER_ID = "contacts"
    CONTAINER_CLASS = "IPF.Contact"
    supported_item_models = (Contact, DistributionList)

    LOCALIZED_NAMES = {
        "da_DK": ("Kontaktpersoner",),
        "de_DE": ("Kontakte",),
        "en_US": ("Contacts",),
        "es_ES": ("Contactos",),
        "fr_CA": ("Contacts",),
        "nl_NL": ("Contactpersonen",),
        "ru_RU": ("Контакты",),
        "sv_SE": ("Kontakter",),
        "zh_CN": ("联系人",),
    }

Ancestors

Subclasses

Class variables

var CONTAINER_CLASS
var DISTINGUISHED_FOLDER_ID
var LOCALIZED_NAMES
var supported_item_models

Inherited members

class ConversationHistory (**kwargs)

Base class to use until we have a more specific folder implementation for this folder.

Expand source code
class ConversationHistory(WellknownFolder):
    DISTINGUISHED_FOLDER_ID = "conversationhistory"
    supported_from = EXCHANGE_2013

Ancestors

Class variables

var DISTINGUISHED_FOLDER_ID
var supported_from

Inherited members

class ConversationSettings (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class ConversationSettings(NonDeletableFolderMixin, Folder):
    CONTAINER_CLASS = "IPF.Configuration"
    LOCALIZED_NAMES = {
        "da_DK": ("Indstillinger for samtalehandlinger",),
    }

Ancestors

Class variables

var CONTAINER_CLASS
var LOCALIZED_NAMES

Inherited members

class CrawlerData (**kwargs)
Expand source code
class CrawlerData(Folder):
    CONTAINER_CLASS = "IPF.StoreItem.CrawlerData"

Ancestors

Class variables

var CONTAINER_CLASS

Inherited members

class DefaultFoldersChangeHistory (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class DefaultFoldersChangeHistory(NonDeletableFolderMixin, Folder):
    CONTAINER_CLASS = "IPM.DefaultFolderHistoryItem"
    LOCALIZED_NAMES = {
        None: ("DefaultFoldersChangeHistory",),
    }

Ancestors

Class variables

var CONTAINER_CLASS
var LOCALIZED_NAMES

Inherited members

class DeferredAction (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class DeferredAction(NonDeletableFolderMixin, Folder):
    LOCALIZED_NAMES = {
        None: ("Deferred Action",),
    }

Ancestors

Class variables

var LOCALIZED_NAMES

Inherited members

class DeletedItems (**kwargs)
Expand source code
class DeletedItems(Folder):
    DISTINGUISHED_FOLDER_ID = "deleteditems"
    CONTAINER_CLASS = "IPF.Note"
    supported_item_models = ITEM_CLASSES

    LOCALIZED_NAMES = {
        "da_DK": ("Slettet post",),
        "de_DE": ("Gelöschte Elemente",),
        "en_US": ("Deleted Items",),
        "es_ES": ("Elementos eliminados",),
        "fr_CA": ("Éléments supprimés",),
        "nl_NL": ("Verwijderde items",),
        "ru_RU": ("Удаленные",),
        "sv_SE": ("Borttaget",),
        "zh_CN": ("已删除邮件",),
    }

Ancestors

Class variables

var CONTAINER_CLASS
var DISTINGUISHED_FOLDER_ID
var LOCALIZED_NAMES
var supported_item_models

Inherited members

class Directory (**kwargs)

Base class to use until we have a more specific folder implementation for this folder.

Expand source code
class Directory(WellknownFolder):
    DISTINGUISHED_FOLDER_ID = "directory"
    supported_from = EXCHANGE_2013_SP1

Ancestors

Class variables

var DISTINGUISHED_FOLDER_ID
var supported_from

Inherited members

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

    ELEMENT_NAME = "DistinguishedFolderId"

    mailbox = MailboxField()

    def clean(self, version=None):
        from .folders import PublicFoldersRoot

        super().clean(version=version)
        if self.id == PublicFoldersRoot.DISTINGUISHED_FOLDER_ID:
            # Avoid "ErrorInvalidOperation: It is not valid to specify a mailbox with the public folder root" from EWS
            self.mailbox = None

Ancestors

Class variables

var ELEMENT_NAME
var FIELDS

Instance variables

var mailbox

Methods

def clean(self, version=None)
Expand source code
def clean(self, version=None):
    from .folders import PublicFoldersRoot

    super().clean(version=version)
    if self.id == PublicFoldersRoot.DISTINGUISHED_FOLDER_ID:
        # Avoid "ErrorInvalidOperation: It is not valid to specify a mailbox with the public folder root" from EWS
        self.mailbox = None

Inherited members

class DlpPolicyEvaluation (**kwargs)
Expand source code
class DlpPolicyEvaluation(Folder):
    CONTAINER_CLASS = "IPF.StoreItem.DlpPolicyEvaluation"

Ancestors

Class variables

var CONTAINER_CLASS

Inherited members

class Drafts (**kwargs)
Expand source code
class Drafts(Messages):
    DISTINGUISHED_FOLDER_ID = "drafts"

    LOCALIZED_NAMES = {
        "da_DK": ("Kladder",),
        "de_DE": ("Entwürfe",),
        "en_US": ("Drafts",),
        "es_ES": ("Borradores",),
        "fr_CA": ("Brouillons",),
        "nl_NL": ("Concepten",),
        "ru_RU": ("Черновики",),
        "sv_SE": ("Utkast",),
        "zh_CN": ("草稿",),
    }

Ancestors

Class variables

var DISTINGUISHED_FOLDER_ID
var LOCALIZED_NAMES

Inherited members

class ExchangeSyncData (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class ExchangeSyncData(NonDeletableFolderMixin, Folder):
    LOCALIZED_NAMES = {
        None: ("ExchangeSyncData",),
    }

Ancestors

Class variables

var LOCALIZED_NAMES

Inherited members

class Favorites (**kwargs)

Base class to use until we have a more specific folder implementation for this folder.

Expand source code
class Favorites(WellknownFolder):
    CONTAINER_CLASS = "IPF.Note"
    DISTINGUISHED_FOLDER_ID = "favorites"
    supported_from = EXCHANGE_2013

Ancestors

Class variables

var CONTAINER_CLASS
var DISTINGUISHED_FOLDER_ID
var supported_from

Inherited members

class Files (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class Files(NonDeletableFolderMixin, Folder):
    CONTAINER_CLASS = "IPF.Files"

    LOCALIZED_NAMES = {
        "da_DK": ("Filer",),
    }

Ancestors

Class variables

var CONTAINER_CLASS
var LOCALIZED_NAMES

Inherited members

class Folder (**kwargs)
Expand source code
class Folder(BaseFolder):
    """MSDN: https://docs.microsoft.com/en-us/exchange/client-developer/web-service-reference/folder"""

    permission_set = PermissionSetField(field_uri="folder:PermissionSet", supported_from=EXCHANGE_2007_SP1)
    effective_rights = EffectiveRightsField(
        field_uri="folder:EffectiveRights", is_read_only=True, supported_from=EXCHANGE_2007_SP1
    )

    __slots__ = ("_root",)

    def __init__(self, **kwargs):
        self._root = kwargs.pop("root", None)  # This is a pointer to the root of the folder hierarchy
        parent = kwargs.pop("parent", None)
        if parent:
            if self.root:
                if parent.root != self.root:
                    raise ValueError("'parent.root' must match 'root'")
            else:
                self.root = parent.root
            if "parent_folder_id" in kwargs and parent.id != kwargs["parent_folder_id"]:
                raise ValueError("'parent_folder_id' must match 'parent' ID")
            kwargs["parent_folder_id"] = ParentFolderId(id=parent.id, changekey=parent.changekey)
        super().__init__(**kwargs)

    @property
    def account(self):
        if self.root is None:
            return None
        return self.root.account

    @property
    def root(self):
        return self._root

    @root.setter
    def root(self, value):
        self._root = value

    @classmethod
    def register(cls, *args, **kwargs):
        if cls is not Folder:
            raise TypeError("For folders, custom fields must be registered on the Folder class")
        return super().register(*args, **kwargs)

    @classmethod
    def deregister(cls, *args, **kwargs):
        if cls is not Folder:
            raise TypeError("For folders, custom fields must be registered on the Folder class")
        return super().deregister(*args, **kwargs)

    @classmethod
    def get_distinguished(cls, root):
        """Get the distinguished folder for this folder class.

        :param root:
        :return:
        """
        try:
            return cls.resolve(
                account=root.account, folder=cls(root=root, name=cls.DISTINGUISHED_FOLDER_ID, is_distinguished=True)
            )
        except MISSING_FOLDER_ERRORS:
            raise ErrorFolderNotFound(f"Could not find distinguished folder {cls.DISTINGUISHED_FOLDER_ID!r}")

    @property
    def parent(self):
        if not self.parent_folder_id:
            return None
        if self.parent_folder_id.id == self.id:
            # Some folders have a parent that references itself. Avoid circular references here
            return None
        return self.root.get_folder(self.parent_folder_id)

    @parent.setter
    def parent(self, value):
        if value is None:
            self.parent_folder_id = None
        else:
            if not isinstance(value, BaseFolder):
                raise InvalidTypeError("value", value, BaseFolder)
            self.root = value.root
            self.parent_folder_id = ParentFolderId(id=value.id, changekey=value.changekey)

    def clean(self, version=None):
        from .roots import RootOfHierarchy

        super().clean(version=version)
        if self.root and not isinstance(self.root, RootOfHierarchy):
            raise InvalidTypeError("root", self.root, RootOfHierarchy)

    @classmethod
    def from_xml_with_root(cls, elem, root):
        folder = cls.from_xml(elem=elem, account=root.account)
        folder_cls = cls
        if cls == Folder:
            # We were called on the generic Folder class. Try to find a more specific class to return objects as.
            #
            # The "FolderClass" element value is the only indication we have in the FindFolder response of which
            # folder class we should create the folder with. And many folders share the same 'FolderClass' value, e.g.
            # Inbox and DeletedItems. We want to distinguish between these because otherwise we can't locate the right
            # folders types for e.g. Account.inbox and Account.trash.
            #
            # We should be able to just use the name, but apparently default folder names can be renamed to a set of
            # localized names using a PowerShell command:
            # https://docs.microsoft.com/en-us/powershell/module/exchange/client-access/Set-MailboxRegionalConfiguration
            #
            # Instead, search for a folder class using the localized name. If none are found, fall back to getting the
            # folder class by the "FolderClass" value.
            #
            # The returned XML may contain neither folder class nor name. In that case, we default to the generic
            # Folder class.
            if folder.name:
                try:
                    # TODO: fld_class.LOCALIZED_NAMES is most definitely neither complete nor authoritative
                    folder_cls = root.folder_cls_from_folder_name(folder_name=folder.name, locale=root.account.locale)
                    log.debug("Folder class %s matches localized folder name %s", folder_cls, folder.name)
                except KeyError:
                    pass
            if folder.folder_class and folder_cls == Folder:
                try:
                    folder_cls = cls.folder_cls_from_container_class(container_class=folder.folder_class)
                    log.debug(
                        "Folder class %s matches container class %s (%s)", folder_cls, folder.folder_class, folder.name
                    )
                except KeyError:
                    pass
            if folder_cls == Folder:
                log.debug("Fallback to class Folder (folder_class %s, name %s)", folder.folder_class, folder.name)
        return folder_cls(root=root, **{f.name: getattr(folder, f.name) for f in folder.FIELDS})

Ancestors

Subclasses

Class variables

var FIELDS

Static methods

def from_xml_with_root(elem, root)
Expand source code
@classmethod
def from_xml_with_root(cls, elem, root):
    folder = cls.from_xml(elem=elem, account=root.account)
    folder_cls = cls
    if cls == Folder:
        # We were called on the generic Folder class. Try to find a more specific class to return objects as.
        #
        # The "FolderClass" element value is the only indication we have in the FindFolder response of which
        # folder class we should create the folder with. And many folders share the same 'FolderClass' value, e.g.
        # Inbox and DeletedItems. We want to distinguish between these because otherwise we can't locate the right
        # folders types for e.g. Account.inbox and Account.trash.
        #
        # We should be able to just use the name, but apparently default folder names can be renamed to a set of
        # localized names using a PowerShell command:
        # https://docs.microsoft.com/en-us/powershell/module/exchange/client-access/Set-MailboxRegionalConfiguration
        #
        # Instead, search for a folder class using the localized name. If none are found, fall back to getting the
        # folder class by the "FolderClass" value.
        #
        # The returned XML may contain neither folder class nor name. In that case, we default to the generic
        # Folder class.
        if folder.name:
            try:
                # TODO: fld_class.LOCALIZED_NAMES is most definitely neither complete nor authoritative
                folder_cls = root.folder_cls_from_folder_name(folder_name=folder.name, locale=root.account.locale)
                log.debug("Folder class %s matches localized folder name %s", folder_cls, folder.name)
            except KeyError:
                pass
        if folder.folder_class and folder_cls == Folder:
            try:
                folder_cls = cls.folder_cls_from_container_class(container_class=folder.folder_class)
                log.debug(
                    "Folder class %s matches container class %s (%s)", folder_cls, folder.folder_class, folder.name
                )
            except KeyError:
                pass
        if folder_cls == Folder:
            log.debug("Fallback to class Folder (folder_class %s, name %s)", folder.folder_class, folder.name)
    return folder_cls(root=root, **{f.name: getattr(folder, f.name) for f in folder.FIELDS})
def get_distinguished(root)

Get the distinguished folder for this folder class.

:param root: :return:

Expand source code
@classmethod
def get_distinguished(cls, root):
    """Get the distinguished folder for this folder class.

    :param root:
    :return:
    """
    try:
        return cls.resolve(
            account=root.account, folder=cls(root=root, name=cls.DISTINGUISHED_FOLDER_ID, is_distinguished=True)
        )
    except MISSING_FOLDER_ERRORS:
        raise ErrorFolderNotFound(f"Could not find distinguished folder {cls.DISTINGUISHED_FOLDER_ID!r}")

Instance variables

var effective_rights
var permission_set

Methods

def clean(self, version=None)
Expand source code
def clean(self, version=None):
    from .roots import RootOfHierarchy

    super().clean(version=version)
    if self.root and not isinstance(self.root, RootOfHierarchy):
        raise InvalidTypeError("root", self.root, RootOfHierarchy)

Inherited members

class FolderCollection (account, folders)

A class that implements an API for searching folders.

Implement a search API on a collection of folders.

:param account: An Account object :param folders: An iterable of folders, e.g. Folder.walk(), Folder.glob(), or [a.calendar, a.inbox]

Expand source code
class FolderCollection(SearchableMixIn):
    """A class that implements an API for searching folders."""

    # These fields are required in a FindFolder or GetFolder call to properly identify folder types
    REQUIRED_FOLDER_FIELDS = ("name", "folder_class")

    def __init__(self, account, folders):
        """Implement a search API on a collection of folders.

        :param account: An Account object
        :param folders: An iterable of folders, e.g. Folder.walk(), Folder.glob(), or [a.calendar, a.inbox]
        """
        self.account = account
        self._folders = folders

    @threaded_cached_property
    def folders(self):
        # Resolve the list of folders, in case it's a generator
        return tuple(self._folders)

    def __len__(self):
        return len(self.folders)

    def __iter__(self):
        yield from self.folders

    def get(self, *args, **kwargs):
        return QuerySet(self).get(*args, **kwargs)

    def all(self):
        return QuerySet(self).all()

    def none(self):
        return QuerySet(self).none()

    def filter(self, *args, **kwargs):
        """Find items in the folder(s).

        Non-keyword args may be a list of Q instances.

        Optional extra keyword arguments follow a Django-like QuerySet filter syntax (see
           https://docs.djangoproject.com/en/1.10/ref/models/querysets/#field-lookups).

        We don't support '__year' and other date-related lookups. We also don't support '__endswith' or '__iendswith'.

        We support the additional '__not' lookup in place of Django's exclude() for simple cases. For more complicated
        cases you need to create a Q object and use ~Q().

        Examples:

            my_account.inbox.filter(datetime_received__gt=EWSDateTime(2016, 1, 1))
            my_account.calendar.filter(start__range=(EWSDateTime(2016, 1, 1), EWSDateTime(2017, 1, 1)))
            my_account.tasks.filter(subject='Hi mom')
            my_account.tasks.filter(subject__not='Hi mom')
            my_account.tasks.filter(subject__contains='Foo')
            my_account.tasks.filter(subject__icontains='foo')

        'endswith' and 'iendswith' could be emulated by searching with 'contains' or 'icontains' and then
        post-processing items. Fetch the field in question with additional_fields and remove items where the search
        string is not a postfix.
        """
        return QuerySet(self).filter(*args, **kwargs)

    def exclude(self, *args, **kwargs):
        return QuerySet(self).exclude(*args, **kwargs)

    def people(self):
        return QuerySet(self).people()

    def view(self, start, end, max_items=None, *args, **kwargs):
        """Implement the CalendarView option to FindItem. The difference between 'filter' and 'view' is that 'filter'
        only returns the master CalendarItem for recurring items, while 'view' unfolds recurring items and returns all
        CalendarItem occurrences as one would normally expect when presenting a calendar.

        Supports the same semantics as filter, except for 'start' and 'end' keyword attributes which are both required
        and behave differently than filter. Here, they denote the start and end of the timespan of the view. All items
        the overlap the timespan are returned (items that end exactly on 'start' are also returned, for some reason).

        EWS does not allow combining CalendarView with search restrictions (filter and exclude).

        'max_items' defines the maximum number of items returned in this view. Optional.

        :param start:
        :param end:
        :param max_items:  (Default value = None)
        :return:
        """
        qs = QuerySet(self).filter(*args, **kwargs)
        qs.calendar_view = CalendarView(start=start, end=end, max_items=max_items)
        return qs

    def allowed_item_fields(self):
        # Return non-ID fields of all item classes allowed in this folder type
        fields = set()
        for item_model in self.supported_item_models:
            fields.update(set(item_model.supported_fields(version=self.account.version)))
        return fields

    @property
    def supported_item_models(self):
        return tuple(item_model for folder in self.folders for item_model in folder.supported_item_models)

    def validate_item_field(self, field, version):
        # Takes a fieldname, Field or FieldPath object pointing to an item field, and checks that it is valid
        # for the item types supported by this folder collection.
        for item_model in self.supported_item_models:
            try:
                item_model.validate_field(field=field, version=version)
                break
            except InvalidField:
                continue
        else:
            raise InvalidField(f"{field!r} is not a valid field on {self.supported_item_models}")

    def _rinse_args(self, q, depth, additional_fields, field_validator):
        if depth is None:
            depth = self._get_default_item_traversal_depth()
        if additional_fields:
            for f in additional_fields:
                field_validator(field=f, version=self.account.version)
                if f.field.is_complex:
                    raise ValueError(f"Field {f.field.name!r} not supported for this service")

        # Build up any restrictions
        if q.is_empty():
            restriction = None
            query_string = None
        elif q.query_string:
            restriction = None
            query_string = Restriction(q, folders=self.folders, applies_to=Restriction.ITEMS)
        else:
            restriction = Restriction(q, folders=self.folders, applies_to=Restriction.ITEMS)
            query_string = None
        return depth, restriction, query_string

    def find_items(
        self,
        q,
        shape=ID_ONLY,
        depth=None,
        additional_fields=None,
        order_fields=None,
        calendar_view=None,
        page_size=None,
        max_items=None,
        offset=0,
    ):
        """Private method to call the FindItem service.

        :param q: a Q instance containing any restrictions
        :param shape: controls whether to return (id, changekey) tuples or Item objects. If additional_fields is
          non-null, we always return Item objects. (Default value = ID_ONLY)
        :param depth: controls the whether to return soft-deleted items or not. (Default value = None)
        :param additional_fields: the extra properties we want on the return objects. Default is no properties. Be aware
          that complex fields can only be fetched with fetch() (i.e. the GetItem service).
        :param order_fields: the SortOrder fields, if any (Default value = None)
        :param calendar_view: a CalendarView instance, if any (Default value = None)
        :param page_size: the requested number of items per page (Default value = None)
        :param max_items: the max number of items to return (Default value = None)
        :param offset: the offset relative to the first item in the item collection (Default value = 0)

        :return: a generator for the returned item IDs or items
        """
        from ..services import FindItem

        if not self.folders:
            log.debug("Folder list is empty")
            return
        if q.is_never():
            log.debug("Query will never return results")
            return
        depth, restriction, query_string = self._rinse_args(
            q=q, depth=depth, additional_fields=additional_fields, field_validator=self.validate_item_field
        )
        if calendar_view is not None and not isinstance(calendar_view, CalendarView):
            raise InvalidTypeError("calendar_view", calendar_view, CalendarView)

        log.debug(
            "Finding %s items in folders %s (shape: %s, depth: %s, additional_fields: %s, restriction: %s)",
            self.account,
            self.folders,
            shape,
            depth,
            additional_fields,
            restriction.q if restriction else None,
        )
        yield from FindItem(account=self.account, page_size=page_size).call(
            folders=self.folders,
            additional_fields=additional_fields,
            restriction=restriction,
            order_fields=order_fields,
            shape=shape,
            query_string=query_string,
            depth=depth,
            calendar_view=calendar_view,
            max_items=calendar_view.max_items if calendar_view else max_items,
            offset=offset,
        )

    def _get_single_folder(self):
        if len(self.folders) > 1:
            raise ValueError("Syncing folder hierarchy can only be done on a single folder")
        if not self.folders:
            log.debug("Folder list is empty")
            return None
        return self.folders[0]

    def find_people(
        self,
        q,
        shape=ID_ONLY,
        depth=None,
        additional_fields=None,
        order_fields=None,
        page_size=None,
        max_items=None,
        offset=0,
    ):
        """Private method to call the FindPeople service.

        :param q: a Q instance containing any restrictions
        :param shape: controls whether to return (id, changekey) tuples or Persona objects. If additional_fields is
          non-null, we always return Persona objects. (Default value = ID_ONLY)
        :param depth: controls the whether to return soft-deleted items or not. (Default value = None)
        :param additional_fields: the extra properties we want on the return objects. Default is no properties.
        :param order_fields: the SortOrder fields, if any (Default value = None)
        :param page_size: the requested number of items per page (Default value = None)
        :param max_items: the max number of items to return (Default value = None)
        :param offset: the offset relative to the first item in the item collection (Default value = 0)

        :return: a generator for the returned personas
        """
        from ..services import FindPeople

        folder = self._get_single_folder()
        if q.is_never():
            log.debug("Query will never return results")
            return
        depth, restriction, query_string = self._rinse_args(
            q=q, depth=depth, additional_fields=additional_fields, field_validator=Persona.validate_field
        )

        yield from FindPeople(account=self.account, page_size=page_size).call(
            folder=folder,
            additional_fields=additional_fields,
            restriction=restriction,
            order_fields=order_fields,
            shape=shape,
            query_string=query_string,
            depth=depth,
            max_items=max_items,
            offset=offset,
        )

    def get_folder_fields(self, target_cls, is_complex=None):
        return {
            FieldPath(field=f)
            for f in target_cls.supported_fields(version=self.account.version)
            if is_complex is None or f.is_complex is is_complex
        }

    def _get_target_cls(self):
        # We may have root folders that don't support the same set of fields as normal folders. If there is a mix of
        # both folder types in self.folders, raise an error so we don't risk losing some fields in the query.
        from .base import Folder
        from .roots import RootOfHierarchy

        has_roots = False
        has_non_roots = False
        for f in self.folders:
            if isinstance(f, RootOfHierarchy):
                if has_non_roots:
                    raise ValueError(f"Cannot call GetFolder on a mix of folder types: {self.folders}")
                has_roots = True
            else:
                if has_roots:
                    raise ValueError(f"Cannot call GetFolder on a mix of folder types: {self.folders}")
                has_non_roots = True
        return RootOfHierarchy if has_roots else Folder

    def _get_default_traversal_depth(self, traversal_attr):
        unique_depths = {getattr(f, traversal_attr) for f in self.folders}
        if len(unique_depths) == 1:
            return unique_depths.pop()
        raise ValueError(
            f"Folders in this collection do not have a common {traversal_attr} value. You need to define an explicit "
            f"traversal depth with QuerySet.depth() (values: {unique_depths})"
        )

    def _get_default_item_traversal_depth(self):
        # When searching folders, some folders require 'Shallow' and others 'Associated' traversal depth.
        return self._get_default_traversal_depth("DEFAULT_ITEM_TRAVERSAL_DEPTH")

    def _get_default_folder_traversal_depth(self):
        # When searching folders, some folders require 'Shallow' and others 'Deep' traversal depth.
        return self._get_default_traversal_depth("DEFAULT_FOLDER_TRAVERSAL_DEPTH")

    def resolve(self):
        # Looks up the folders or folder IDs in the collection and returns full Folder instances with all fields set.
        from .base import BaseFolder

        resolveable_folders = []
        for f in self.folders:
            if isinstance(f, BaseFolder) and not f.get_folder_allowed:
                log.debug("GetFolder not allowed on folder %s. Non-complex fields must be fetched with FindFolder", f)
                yield f
            else:
                resolveable_folders.append(f)
        # Fetch all properties for the remaining folders of folder IDs
        additional_fields = self.get_folder_fields(target_cls=self._get_target_cls(), is_complex=None)
        yield from self.__class__(account=self.account, folders=resolveable_folders).get_folders(
            additional_fields=additional_fields
        )

    @require_account
    def find_folders(
        self, q=None, shape=ID_ONLY, depth=None, additional_fields=None, page_size=None, max_items=None, offset=0
    ):
        from ..services import FindFolder

        # 'depth' controls whether to return direct children or recurse into sub-folders
        from .base import BaseFolder, Folder

        if q is None:
            q = Q()
        if not self.folders:
            log.debug("Folder list is empty")
            return
        if q.is_never():
            log.debug("Query will never return results")
            return
        if q.is_empty():
            restriction = None
        else:
            restriction = Restriction(q, folders=self.folders, applies_to=Restriction.FOLDERS)
        if depth is None:
            depth = self._get_default_folder_traversal_depth()
        if additional_fields is None:
            # Default to all non-complex properties. Subfolders will always be of class Folder
            additional_fields = self.get_folder_fields(target_cls=Folder, is_complex=False)
        else:
            for f in additional_fields:
                if f.field.is_complex:
                    raise ValueError(f"find_folders() does not support field {f.field.name!r}. Use get_folders().")

        # Add required fields
        additional_fields.update(
            (FieldPath(field=BaseFolder.get_field_by_fieldname(f)) for f in self.REQUIRED_FOLDER_FIELDS)
        )

        yield from FindFolder(account=self.account, page_size=page_size).call(
            folders=self.folders,
            additional_fields=additional_fields,
            restriction=restriction,
            shape=shape,
            depth=depth,
            max_items=max_items,
            offset=offset,
        )

    def get_folders(self, additional_fields=None):
        from ..services import GetFolder

        # Expand folders with their full set of properties
        from .base import BaseFolder

        if not self.folders:
            log.debug("Folder list is empty")
            return
        if additional_fields is None:
            # Default to all complex properties
            additional_fields = self.get_folder_fields(target_cls=self._get_target_cls(), is_complex=True)

        # Add required fields
        additional_fields.update(
            (FieldPath(field=BaseFolder.get_field_by_fieldname(f)) for f in self.REQUIRED_FOLDER_FIELDS)
        )

        yield from GetFolder(account=self.account).call(
            folders=self.folders,
            additional_fields=additional_fields,
            shape=ID_ONLY,
        )

    def subscribe_to_pull(self, event_types=None, watermark=None, timeout=60):
        from ..services import SubscribeToPull

        if not self.folders:
            log.debug("Folder list is empty")
            return None
        if event_types is None:
            event_types = SubscribeToPull.EVENT_TYPES
        return SubscribeToPull(account=self.account).get(
            folders=self.folders,
            event_types=event_types,
            watermark=watermark,
            timeout=timeout,
        )

    def subscribe_to_push(self, callback_url, event_types=None, watermark=None, status_frequency=1):
        from ..services import SubscribeToPush

        if not self.folders:
            log.debug("Folder list is empty")
            return None
        if event_types is None:
            event_types = SubscribeToPush.EVENT_TYPES
        return SubscribeToPush(account=self.account).get(
            folders=self.folders,
            event_types=event_types,
            watermark=watermark,
            status_frequency=status_frequency,
            url=callback_url,
        )

    def subscribe_to_streaming(self, event_types=None):
        from ..services import SubscribeToStreaming

        if not self.folders:
            log.debug("Folder list is empty")
            return None
        if event_types is None:
            event_types = SubscribeToStreaming.EVENT_TYPES
        return SubscribeToStreaming(account=self.account).get(folders=self.folders, event_types=event_types)

    def pull_subscription(self, **kwargs):
        return PullSubscription(folder=self, **kwargs)

    def push_subscription(self, **kwargs):
        return PushSubscription(folder=self, **kwargs)

    def streaming_subscription(self, **kwargs):
        return StreamingSubscription(folder=self, **kwargs)

    def unsubscribe(self, subscription_id):
        """Unsubscribe. Only applies to pull and streaming notifications.

        :param subscription_id: A subscription ID as acquired by .subscribe_to_[pull|streaming]()
        :return: True

        This method doesn't need the current collection instance, but it makes sense to keep the method along the other
        sync methods.
        """
        from ..services import Unsubscribe

        return Unsubscribe(account=self.account).get(subscription_id=subscription_id)

    def sync_items(self, sync_state=None, only_fields=None, ignore=None, max_changes_returned=None, sync_scope=None):
        from ..services import SyncFolderItems

        folder = self._get_single_folder()
        if only_fields is None:
            # We didn't restrict list of field paths. Get all fields from the server, including extended properties.
            additional_fields = {FieldPath(field=f) for f in folder.allowed_item_fields(version=self.account.version)}
        else:
            for field in only_fields:
                folder.validate_item_field(field=field, version=self.account.version)
            # Remove ItemId and ChangeKey. We get them unconditionally
            additional_fields = {f for f in folder.normalize_fields(fields=only_fields) if not f.field.is_attribute}

        svc = SyncFolderItems(account=self.account)
        while True:
            yield from svc.call(
                folder=folder,
                shape=ID_ONLY,
                additional_fields=additional_fields,
                sync_state=sync_state,
                ignore=ignore,
                max_changes_returned=max_changes_returned,
                sync_scope=sync_scope,
            )
            if svc.sync_state == sync_state:
                # We sometimes get the same sync_state back, even though includes_last_item_in_range is False. Stop here
                break
            sync_state = svc.sync_state  # Set the new sync state in the next call
            if svc.includes_last_item_in_range:  # Try again if there are more items
                break
        raise SyncCompleted(sync_state=svc.sync_state)

    def sync_hierarchy(self, sync_state=None, only_fields=None):
        from ..services import SyncFolderHierarchy

        folder = self._get_single_folder()
        if only_fields is None:
            # We didn't restrict list of field paths. Get all fields from the server, including extended properties.
            additional_fields = {FieldPath(field=f) for f in folder.supported_fields(version=self.account.version)}
        else:
            additional_fields = set()
            for field_name in only_fields:
                folder.validate_field(field=field_name, version=self.account.version)
                f = folder.get_field_by_fieldname(fieldname=field_name)
                if not f.is_attribute:
                    # Remove ItemId and ChangeKey. We get them unconditionally
                    additional_fields.add(FieldPath(field=f))

        # Add required fields
        additional_fields.update(
            (FieldPath(field=folder.get_field_by_fieldname(f)) for f in self.REQUIRED_FOLDER_FIELDS)
        )

        svc = SyncFolderHierarchy(account=self.account)
        while True:
            yield from svc.call(
                folder=folder,
                shape=ID_ONLY,
                additional_fields=additional_fields,
                sync_state=sync_state,
            )
            if svc.sync_state == sync_state:
                # We sometimes get the same sync_state back, even though includes_last_item_in_range is False. Stop here
                break
            sync_state = svc.sync_state  # Set the new sync state in the next call
            if svc.includes_last_item_in_range:  # Try again if there are more items
                break
        raise SyncCompleted(sync_state=svc.sync_state)

Ancestors

Class variables

var REQUIRED_FOLDER_FIELDS

Instance variables

var folders
Expand source code
def __get__(self, obj, cls):
    if obj is None:
        return self

    obj_dict = obj.__dict__
    name = self.func.__name__
    with self.lock:
        try:
            # check if the value was computed before the lock was acquired
            return obj_dict[name]

        except KeyError:
            # if not, do the calculation and release the lock
            return obj_dict.setdefault(name, self.func(obj))
var supported_item_models
Expand source code
@property
def supported_item_models(self):
    return tuple(item_model for folder in self.folders for item_model in folder.supported_item_models)

Methods

def allowed_item_fields(self)
Expand source code
def allowed_item_fields(self):
    # Return non-ID fields of all item classes allowed in this folder type
    fields = set()
    for item_model in self.supported_item_models:
        fields.update(set(item_model.supported_fields(version=self.account.version)))
    return fields
def filter(self, *args, **kwargs)

Find items in the folder(s).

Non-keyword args may be a list of Q instances.

Optional extra keyword arguments follow a Django-like QuerySet filter syntax (see https://docs.djangoproject.com/en/1.10/ref/models/querysets/#field-lookups).

We don't support '__year' and other date-related lookups. We also don't support '__endswith' or '__iendswith'.

We support the additional '__not' lookup in place of Django's exclude() for simple cases. For more complicated cases you need to create a Q object and use ~Q().

Examples

my_account.inbox.filter(datetime_received__gt=EWSDateTime(2016, 1, 1)) my_account.calendar.filter(start__range=(EWSDateTime(2016, 1, 1), EWSDateTime(2017, 1, 1))) my_account.tasks.filter(subject='Hi mom') my_account.tasks.filter(subject__not='Hi mom') my_account.tasks.filter(subject__contains='Foo') my_account.tasks.filter(subject__icontains='foo')

'endswith' and 'iendswith' could be emulated by searching with 'contains' or 'icontains' and then post-processing items. Fetch the field in question with additional_fields and remove items where the search string is not a postfix.

Expand source code
def filter(self, *args, **kwargs):
    """Find items in the folder(s).

    Non-keyword args may be a list of Q instances.

    Optional extra keyword arguments follow a Django-like QuerySet filter syntax (see
       https://docs.djangoproject.com/en/1.10/ref/models/querysets/#field-lookups).

    We don't support '__year' and other date-related lookups. We also don't support '__endswith' or '__iendswith'.

    We support the additional '__not' lookup in place of Django's exclude() for simple cases. For more complicated
    cases you need to create a Q object and use ~Q().

    Examples:

        my_account.inbox.filter(datetime_received__gt=EWSDateTime(2016, 1, 1))
        my_account.calendar.filter(start__range=(EWSDateTime(2016, 1, 1), EWSDateTime(2017, 1, 1)))
        my_account.tasks.filter(subject='Hi mom')
        my_account.tasks.filter(subject__not='Hi mom')
        my_account.tasks.filter(subject__contains='Foo')
        my_account.tasks.filter(subject__icontains='foo')

    'endswith' and 'iendswith' could be emulated by searching with 'contains' or 'icontains' and then
    post-processing items. Fetch the field in question with additional_fields and remove items where the search
    string is not a postfix.
    """
    return QuerySet(self).filter(*args, **kwargs)
def find_folders(self, q=None, shape='IdOnly', depth=None, additional_fields=None, page_size=None, max_items=None, offset=0)
Expand source code
@require_account
def find_folders(
    self, q=None, shape=ID_ONLY, depth=None, additional_fields=None, page_size=None, max_items=None, offset=0
):
    from ..services import FindFolder

    # 'depth' controls whether to return direct children or recurse into sub-folders
    from .base import BaseFolder, Folder

    if q is None:
        q = Q()
    if not self.folders:
        log.debug("Folder list is empty")
        return
    if q.is_never():
        log.debug("Query will never return results")
        return
    if q.is_empty():
        restriction = None
    else:
        restriction = Restriction(q, folders=self.folders, applies_to=Restriction.FOLDERS)
    if depth is None:
        depth = self._get_default_folder_traversal_depth()
    if additional_fields is None:
        # Default to all non-complex properties. Subfolders will always be of class Folder
        additional_fields = self.get_folder_fields(target_cls=Folder, is_complex=False)
    else:
        for f in additional_fields:
            if f.field.is_complex:
                raise ValueError(f"find_folders() does not support field {f.field.name!r}. Use get_folders().")

    # Add required fields
    additional_fields.update(
        (FieldPath(field=BaseFolder.get_field_by_fieldname(f)) for f in self.REQUIRED_FOLDER_FIELDS)
    )

    yield from FindFolder(account=self.account, page_size=page_size).call(
        folders=self.folders,
        additional_fields=additional_fields,
        restriction=restriction,
        shape=shape,
        depth=depth,
        max_items=max_items,
        offset=offset,
    )
def find_items(self, q, shape='IdOnly', depth=None, additional_fields=None, order_fields=None, calendar_view=None, page_size=None, max_items=None, offset=0)

Private method to call the FindItem service.

:param q: a Q instance containing any restrictions :param shape: controls whether to return (id, changekey) tuples or Item objects. If additional_fields is non-null, we always return Item objects. (Default value = ID_ONLY) :param depth: controls the whether to return soft-deleted items or not. (Default value = None) :param additional_fields: the extra properties we want on the return objects. Default is no properties. Be aware that complex fields can only be fetched with fetch() (i.e. the GetItem service). :param order_fields: the SortOrder fields, if any (Default value = None) :param calendar_view: a CalendarView instance, if any (Default value = None) :param page_size: the requested number of items per page (Default value = None) :param max_items: the max number of items to return (Default value = None) :param offset: the offset relative to the first item in the item collection (Default value = 0)

:return: a generator for the returned item IDs or items

Expand source code
def find_items(
    self,
    q,
    shape=ID_ONLY,
    depth=None,
    additional_fields=None,
    order_fields=None,
    calendar_view=None,
    page_size=None,
    max_items=None,
    offset=0,
):
    """Private method to call the FindItem service.

    :param q: a Q instance containing any restrictions
    :param shape: controls whether to return (id, changekey) tuples or Item objects. If additional_fields is
      non-null, we always return Item objects. (Default value = ID_ONLY)
    :param depth: controls the whether to return soft-deleted items or not. (Default value = None)
    :param additional_fields: the extra properties we want on the return objects. Default is no properties. Be aware
      that complex fields can only be fetched with fetch() (i.e. the GetItem service).
    :param order_fields: the SortOrder fields, if any (Default value = None)
    :param calendar_view: a CalendarView instance, if any (Default value = None)
    :param page_size: the requested number of items per page (Default value = None)
    :param max_items: the max number of items to return (Default value = None)
    :param offset: the offset relative to the first item in the item collection (Default value = 0)

    :return: a generator for the returned item IDs or items
    """
    from ..services import FindItem

    if not self.folders:
        log.debug("Folder list is empty")
        return
    if q.is_never():
        log.debug("Query will never return results")
        return
    depth, restriction, query_string = self._rinse_args(
        q=q, depth=depth, additional_fields=additional_fields, field_validator=self.validate_item_field
    )
    if calendar_view is not None and not isinstance(calendar_view, CalendarView):
        raise InvalidTypeError("calendar_view", calendar_view, CalendarView)

    log.debug(
        "Finding %s items in folders %s (shape: %s, depth: %s, additional_fields: %s, restriction: %s)",
        self.account,
        self.folders,
        shape,
        depth,
        additional_fields,
        restriction.q if restriction else None,
    )
    yield from FindItem(account=self.account, page_size=page_size).call(
        folders=self.folders,
        additional_fields=additional_fields,
        restriction=restriction,
        order_fields=order_fields,
        shape=shape,
        query_string=query_string,
        depth=depth,
        calendar_view=calendar_view,
        max_items=calendar_view.max_items if calendar_view else max_items,
        offset=offset,
    )
def find_people(self, q, shape='IdOnly', depth=None, additional_fields=None, order_fields=None, page_size=None, max_items=None, offset=0)

Private method to call the FindPeople service.

:param q: a Q instance containing any restrictions :param shape: controls whether to return (id, changekey) tuples or Persona objects. If additional_fields is non-null, we always return Persona objects. (Default value = ID_ONLY) :param depth: controls the whether to return soft-deleted items or not. (Default value = None) :param additional_fields: the extra properties we want on the return objects. Default is no properties. :param order_fields: the SortOrder fields, if any (Default value = None) :param page_size: the requested number of items per page (Default value = None) :param max_items: the max number of items to return (Default value = None) :param offset: the offset relative to the first item in the item collection (Default value = 0)

:return: a generator for the returned personas

Expand source code
def find_people(
    self,
    q,
    shape=ID_ONLY,
    depth=None,
    additional_fields=None,
    order_fields=None,
    page_size=None,
    max_items=None,
    offset=0,
):
    """Private method to call the FindPeople service.

    :param q: a Q instance containing any restrictions
    :param shape: controls whether to return (id, changekey) tuples or Persona objects. If additional_fields is
      non-null, we always return Persona objects. (Default value = ID_ONLY)
    :param depth: controls the whether to return soft-deleted items or not. (Default value = None)
    :param additional_fields: the extra properties we want on the return objects. Default is no properties.
    :param order_fields: the SortOrder fields, if any (Default value = None)
    :param page_size: the requested number of items per page (Default value = None)
    :param max_items: the max number of items to return (Default value = None)
    :param offset: the offset relative to the first item in the item collection (Default value = 0)

    :return: a generator for the returned personas
    """
    from ..services import FindPeople

    folder = self._get_single_folder()
    if q.is_never():
        log.debug("Query will never return results")
        return
    depth, restriction, query_string = self._rinse_args(
        q=q, depth=depth, additional_fields=additional_fields, field_validator=Persona.validate_field
    )

    yield from FindPeople(account=self.account, page_size=page_size).call(
        folder=folder,
        additional_fields=additional_fields,
        restriction=restriction,
        order_fields=order_fields,
        shape=shape,
        query_string=query_string,
        depth=depth,
        max_items=max_items,
        offset=offset,
    )
def get_folder_fields(self, target_cls, is_complex=None)
Expand source code
def get_folder_fields(self, target_cls, is_complex=None):
    return {
        FieldPath(field=f)
        for f in target_cls.supported_fields(version=self.account.version)
        if is_complex is None or f.is_complex is is_complex
    }
def get_folders(self, additional_fields=None)
Expand source code
def get_folders(self, additional_fields=None):
    from ..services import GetFolder

    # Expand folders with their full set of properties
    from .base import BaseFolder

    if not self.folders:
        log.debug("Folder list is empty")
        return
    if additional_fields is None:
        # Default to all complex properties
        additional_fields = self.get_folder_fields(target_cls=self._get_target_cls(), is_complex=True)

    # Add required fields
    additional_fields.update(
        (FieldPath(field=BaseFolder.get_field_by_fieldname(f)) for f in self.REQUIRED_FOLDER_FIELDS)
    )

    yield from GetFolder(account=self.account).call(
        folders=self.folders,
        additional_fields=additional_fields,
        shape=ID_ONLY,
    )
def pull_subscription(self, **kwargs)
Expand source code
def pull_subscription(self, **kwargs):
    return PullSubscription(folder=self, **kwargs)
def push_subscription(self, **kwargs)
Expand source code
def push_subscription(self, **kwargs):
    return PushSubscription(folder=self, **kwargs)
def resolve(self)
Expand source code
def resolve(self):
    # Looks up the folders or folder IDs in the collection and returns full Folder instances with all fields set.
    from .base import BaseFolder

    resolveable_folders = []
    for f in self.folders:
        if isinstance(f, BaseFolder) and not f.get_folder_allowed:
            log.debug("GetFolder not allowed on folder %s. Non-complex fields must be fetched with FindFolder", f)
            yield f
        else:
            resolveable_folders.append(f)
    # Fetch all properties for the remaining folders of folder IDs
    additional_fields = self.get_folder_fields(target_cls=self._get_target_cls(), is_complex=None)
    yield from self.__class__(account=self.account, folders=resolveable_folders).get_folders(
        additional_fields=additional_fields
    )
def streaming_subscription(self, **kwargs)
Expand source code
def streaming_subscription(self, **kwargs):
    return StreamingSubscription(folder=self, **kwargs)
def subscribe_to_pull(self, event_types=None, watermark=None, timeout=60)
Expand source code
def subscribe_to_pull(self, event_types=None, watermark=None, timeout=60):
    from ..services import SubscribeToPull

    if not self.folders:
        log.debug("Folder list is empty")
        return None
    if event_types is None:
        event_types = SubscribeToPull.EVENT_TYPES
    return SubscribeToPull(account=self.account).get(
        folders=self.folders,
        event_types=event_types,
        watermark=watermark,
        timeout=timeout,
    )
def subscribe_to_push(self, callback_url, event_types=None, watermark=None, status_frequency=1)
Expand source code
def subscribe_to_push(self, callback_url, event_types=None, watermark=None, status_frequency=1):
    from ..services import SubscribeToPush

    if not self.folders:
        log.debug("Folder list is empty")
        return None
    if event_types is None:
        event_types = SubscribeToPush.EVENT_TYPES
    return SubscribeToPush(account=self.account).get(
        folders=self.folders,
        event_types=event_types,
        watermark=watermark,
        status_frequency=status_frequency,
        url=callback_url,
    )
def subscribe_to_streaming(self, event_types=None)
Expand source code
def subscribe_to_streaming(self, event_types=None):
    from ..services import SubscribeToStreaming

    if not self.folders:
        log.debug("Folder list is empty")
        return None
    if event_types is None:
        event_types = SubscribeToStreaming.EVENT_TYPES
    return SubscribeToStreaming(account=self.account).get(folders=self.folders, event_types=event_types)
def sync_hierarchy(self, sync_state=None, only_fields=None)
Expand source code
def sync_hierarchy(self, sync_state=None, only_fields=None):
    from ..services import SyncFolderHierarchy

    folder = self._get_single_folder()
    if only_fields is None:
        # We didn't restrict list of field paths. Get all fields from the server, including extended properties.
        additional_fields = {FieldPath(field=f) for f in folder.supported_fields(version=self.account.version)}
    else:
        additional_fields = set()
        for field_name in only_fields:
            folder.validate_field(field=field_name, version=self.account.version)
            f = folder.get_field_by_fieldname(fieldname=field_name)
            if not f.is_attribute:
                # Remove ItemId and ChangeKey. We get them unconditionally
                additional_fields.add(FieldPath(field=f))

    # Add required fields
    additional_fields.update(
        (FieldPath(field=folder.get_field_by_fieldname(f)) for f in self.REQUIRED_FOLDER_FIELDS)
    )

    svc = SyncFolderHierarchy(account=self.account)
    while True:
        yield from svc.call(
            folder=folder,
            shape=ID_ONLY,
            additional_fields=additional_fields,
            sync_state=sync_state,
        )
        if svc.sync_state == sync_state:
            # We sometimes get the same sync_state back, even though includes_last_item_in_range is False. Stop here
            break
        sync_state = svc.sync_state  # Set the new sync state in the next call
        if svc.includes_last_item_in_range:  # Try again if there are more items
            break
    raise SyncCompleted(sync_state=svc.sync_state)
def sync_items(self, sync_state=None, only_fields=None, ignore=None, max_changes_returned=None, sync_scope=None)
Expand source code
def sync_items(self, sync_state=None, only_fields=None, ignore=None, max_changes_returned=None, sync_scope=None):
    from ..services import SyncFolderItems

    folder = self._get_single_folder()
    if only_fields is None:
        # We didn't restrict list of field paths. Get all fields from the server, including extended properties.
        additional_fields = {FieldPath(field=f) for f in folder.allowed_item_fields(version=self.account.version)}
    else:
        for field in only_fields:
            folder.validate_item_field(field=field, version=self.account.version)
        # Remove ItemId and ChangeKey. We get them unconditionally
        additional_fields = {f for f in folder.normalize_fields(fields=only_fields) if not f.field.is_attribute}

    svc = SyncFolderItems(account=self.account)
    while True:
        yield from svc.call(
            folder=folder,
            shape=ID_ONLY,
            additional_fields=additional_fields,
            sync_state=sync_state,
            ignore=ignore,
            max_changes_returned=max_changes_returned,
            sync_scope=sync_scope,
        )
        if svc.sync_state == sync_state:
            # We sometimes get the same sync_state back, even though includes_last_item_in_range is False. Stop here
            break
        sync_state = svc.sync_state  # Set the new sync state in the next call
        if svc.includes_last_item_in_range:  # Try again if there are more items
            break
    raise SyncCompleted(sync_state=svc.sync_state)
def unsubscribe(self, subscription_id)

Unsubscribe. Only applies to pull and streaming notifications.

:param subscription_id: A subscription ID as acquired by .subscribe_to_pull|streaming :return: True

This method doesn't need the current collection instance, but it makes sense to keep the method along the other sync methods.

Expand source code
def unsubscribe(self, subscription_id):
    """Unsubscribe. Only applies to pull and streaming notifications.

    :param subscription_id: A subscription ID as acquired by .subscribe_to_[pull|streaming]()
    :return: True

    This method doesn't need the current collection instance, but it makes sense to keep the method along the other
    sync methods.
    """
    from ..services import Unsubscribe

    return Unsubscribe(account=self.account).get(subscription_id=subscription_id)
def validate_item_field(self, field, version)
Expand source code
def validate_item_field(self, field, version):
    # Takes a fieldname, Field or FieldPath object pointing to an item field, and checks that it is valid
    # for the item types supported by this folder collection.
    for item_model in self.supported_item_models:
        try:
            item_model.validate_field(field=field, version=version)
            break
        except InvalidField:
            continue
    else:
        raise InvalidField(f"{field!r} is not a valid field on {self.supported_item_models}")
def view(self, start, end, max_items=None, *args, **kwargs)

Implement the CalendarView option to FindItem. The difference between 'filter' and 'view' is that 'filter' only returns the master CalendarItem for recurring items, while 'view' unfolds recurring items and returns all CalendarItem occurrences as one would normally expect when presenting a calendar.

Supports the same semantics as filter, except for 'start' and 'end' keyword attributes which are both required and behave differently than filter. Here, they denote the start and end of the timespan of the view. All items the overlap the timespan are returned (items that end exactly on 'start' are also returned, for some reason).

EWS does not allow combining CalendarView with search restrictions (filter and exclude).

'max_items' defines the maximum number of items returned in this view. Optional.

:param start: :param end: :param max_items: (Default value = None) :return:

Expand source code
def view(self, start, end, max_items=None, *args, **kwargs):
    """Implement the CalendarView option to FindItem. The difference between 'filter' and 'view' is that 'filter'
    only returns the master CalendarItem for recurring items, while 'view' unfolds recurring items and returns all
    CalendarItem occurrences as one would normally expect when presenting a calendar.

    Supports the same semantics as filter, except for 'start' and 'end' keyword attributes which are both required
    and behave differently than filter. Here, they denote the start and end of the timespan of the view. All items
    the overlap the timespan are returned (items that end exactly on 'start' are also returned, for some reason).

    EWS does not allow combining CalendarView with search restrictions (filter and exclude).

    'max_items' defines the maximum number of items returned in this view. Optional.

    :param start:
    :param end:
    :param max_items:  (Default value = None)
    :return:
    """
    qs = QuerySet(self).filter(*args, **kwargs)
    qs.calendar_view = CalendarView(start=start, end=end, max_items=max_items)
    return qs

Inherited members

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

    ELEMENT_NAME = "FolderId"

Ancestors

Subclasses

Class variables

var ELEMENT_NAME

Inherited members

class FolderQuerySet (folder_collection)

A QuerySet-like class for finding subfolders of a folder collection.

Expand source code
class FolderQuerySet:
    """A QuerySet-like class for finding subfolders of a folder collection."""

    def __init__(self, folder_collection):
        from .collections import FolderCollection

        if not isinstance(folder_collection, FolderCollection):
            raise InvalidTypeError("folder_collection", folder_collection, FolderCollection)
        self.folder_collection = folder_collection
        self.q = Q()  # Default to no restrictions
        self.only_fields = None
        self._depth = None

    def _copy_cls(self):
        return self.__class__(folder_collection=self.folder_collection)

    def _copy_self(self):
        """Chaining operations must make a copy of self before making any modifications."""
        new_qs = self._copy_cls()
        new_qs.q = deepcopy(self.q)
        new_qs.only_fields = self.only_fields
        new_qs._depth = self._depth
        return new_qs

    def only(self, *args):
        """Restrict the fields returned. 'name' and 'folder_class' are always returned."""
        from .base import Folder

        # Subfolders will always be of class Folder
        all_fields = self.folder_collection.get_folder_fields(target_cls=Folder, is_complex=None)
        all_fields.update(Folder.attribute_fields())
        only_fields = []
        for arg in args:
            for field_path in all_fields:
                if field_path.field.name == arg:
                    only_fields.append(field_path)
                    break
            else:
                raise InvalidField(f"Unknown field {arg!r} on folders {self.folder_collection.folders}")
        new_qs = self._copy_self()
        new_qs.only_fields = only_fields
        return new_qs

    def depth(self, depth):
        """Specify the search depth. Possible values are: SHALLOW or DEEP.

        :param depth:
        """
        new_qs = self._copy_self()
        new_qs._depth = depth
        return new_qs

    def get(self, *args, **kwargs):
        """Return the query result as exactly one item. Raises DoesNotExist if there are no results, and
        MultipleObjectsReturned if there are multiple results.
        """
        from .collections import FolderCollection

        if not args and set(kwargs) in ({"id"}, {"id", "changekey"}):
            folders = list(
                FolderCollection(account=self.folder_collection.account, folders=[FolderId(**kwargs)]).resolve()
            )
        elif args or kwargs:
            folders = list(self.filter(*args, **kwargs))
        else:
            folders = list(self.all())
        if not folders:
            raise DoesNotExist("Could not find a child folder matching the query")
        if len(folders) != 1:
            raise MultipleObjectsReturned(f"Expected result length 1, but got {folders}")
        f = folders[0]
        if isinstance(f, Exception):
            raise f
        return f

    def all(self):
        """ """
        new_qs = self._copy_self()
        return new_qs

    def filter(self, *args, **kwargs):
        """Add restrictions to the folder search."""
        new_qs = self._copy_self()
        q = Q(*args, **kwargs)
        new_qs.q = new_qs.q & q
        return new_qs

    def __iter__(self):
        return self._query()

    def _query(self):
        from .base import Folder
        from .collections import FolderCollection

        if self.only_fields is None:
            # Subfolders will always be of class Folder
            non_complex_fields = self.folder_collection.get_folder_fields(target_cls=Folder, is_complex=False)
            complex_fields = self.folder_collection.get_folder_fields(target_cls=Folder, is_complex=True)
        else:
            non_complex_fields = {f for f in self.only_fields if not f.field.is_complex}
            complex_fields = {f for f in self.only_fields if f.field.is_complex}

        # First, fetch all non-complex fields using FindFolder. We do this because some folders do not support
        # GetFolder but we still want to get as much information as possible.
        folders = self.folder_collection.find_folders(q=self.q, depth=self._depth, additional_fields=non_complex_fields)
        if not complex_fields:
            yield from folders
            return

        # Fetch all properties for the found folders
        resolveable_folders = []
        for f in folders:
            if isinstance(f, Exception):
                yield f
                continue
            if not f.get_folder_allowed:
                log.debug("GetFolder not allowed on folder %s. Non-complex fields must be fetched with FindFolder", f)
                yield f
            else:
                resolveable_folders.append(f)

        # Get the complex fields using GetFolder, for the folders that support it, and add the extra field values
        complex_folders = FolderCollection(
            account=self.folder_collection.account, folders=resolveable_folders
        ).get_folders(additional_fields=complex_fields)
        for f, complex_f in zip(resolveable_folders, complex_folders):
            if isinstance(f, MISSING_FOLDER_ERRORS):
                # We were unlucky. The folder disappeared between the FindFolder and the GetFolder calls
                continue
            if isinstance(complex_f, Exception):
                yield complex_f
                continue
            # Add the extra field values to the folders we fetched with find_folders()
            if f.__class__ != complex_f.__class__:
                raise ValueError(f"Type mismatch: {f} vs {complex_f}")
            for complex_field in complex_fields:
                field_name = complex_field.field.name
                setattr(f, field_name, getattr(complex_f, field_name))
            yield f

Subclasses

Methods

def all(self)
Expand source code
def all(self):
    """ """
    new_qs = self._copy_self()
    return new_qs
def depth(self, depth)

Specify the search depth. Possible values are: SHALLOW or DEEP.

:param depth:

Expand source code
def depth(self, depth):
    """Specify the search depth. Possible values are: SHALLOW or DEEP.

    :param depth:
    """
    new_qs = self._copy_self()
    new_qs._depth = depth
    return new_qs
def filter(self, *args, **kwargs)

Add restrictions to the folder search.

Expand source code
def filter(self, *args, **kwargs):
    """Add restrictions to the folder search."""
    new_qs = self._copy_self()
    q = Q(*args, **kwargs)
    new_qs.q = new_qs.q & q
    return new_qs
def get(self, *args, **kwargs)

Return the query result as exactly one item. Raises DoesNotExist if there are no results, and MultipleObjectsReturned if there are multiple results.

Expand source code
def get(self, *args, **kwargs):
    """Return the query result as exactly one item. Raises DoesNotExist if there are no results, and
    MultipleObjectsReturned if there are multiple results.
    """
    from .collections import FolderCollection

    if not args and set(kwargs) in ({"id"}, {"id", "changekey"}):
        folders = list(
            FolderCollection(account=self.folder_collection.account, folders=[FolderId(**kwargs)]).resolve()
        )
    elif args or kwargs:
        folders = list(self.filter(*args, **kwargs))
    else:
        folders = list(self.all())
    if not folders:
        raise DoesNotExist("Could not find a child folder matching the query")
    if len(folders) != 1:
        raise MultipleObjectsReturned(f"Expected result length 1, but got {folders}")
    f = folders[0]
    if isinstance(f, Exception):
        raise f
    return f
def only(self, *args)

Restrict the fields returned. 'name' and 'folder_class' are always returned.

Expand source code
def only(self, *args):
    """Restrict the fields returned. 'name' and 'folder_class' are always returned."""
    from .base import Folder

    # Subfolders will always be of class Folder
    all_fields = self.folder_collection.get_folder_fields(target_cls=Folder, is_complex=None)
    all_fields.update(Folder.attribute_fields())
    only_fields = []
    for arg in args:
        for field_path in all_fields:
            if field_path.field.name == arg:
                only_fields.append(field_path)
                break
        else:
            raise InvalidField(f"Unknown field {arg!r} on folders {self.folder_collection.folders}")
    new_qs = self._copy_self()
    new_qs.only_fields = only_fields
    return new_qs
class FreeBusyCache (**kwargs)
Expand source code
class FreeBusyCache(Folder):
    CONTAINER_CLASS = "IPF.StoreItem.FreeBusyCache"

Ancestors

Class variables

var CONTAINER_CLASS

Inherited members

class FreebusyData (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class FreebusyData(NonDeletableFolderMixin, Folder):
    LOCALIZED_NAMES = {
        None: ("Freebusy Data",),
    }

Ancestors

Class variables

var LOCALIZED_NAMES

Inherited members

class Friends (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class Friends(NonDeletableFolderMixin, Contacts):
    CONTAINER_CLASS = "IPF.Note"

    LOCALIZED_NAMES = {
        "de_DE": ("Bekannte",),
    }

Ancestors

Class variables

var CONTAINER_CLASS
var LOCALIZED_NAMES

Inherited members

class GALContacts (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class GALContacts(NonDeletableFolderMixin, Contacts):
    DISTINGUISHED_FOLDER_ID = None
    CONTAINER_CLASS = "IPF.Contact.GalContacts"

    LOCALIZED_NAMES = {
        None: ("GAL Contacts",),
    }

Ancestors

Class variables

var CONTAINER_CLASS
var DISTINGUISHED_FOLDER_ID
var LOCALIZED_NAMES

Inherited members

class GraphAnalytics (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class GraphAnalytics(NonDeletableFolderMixin, Folder):
    CONTAINER_CLASS = "IPF.StoreItem.GraphAnalytics"
    LOCALIZED_NAMES = {
        None: ("GraphAnalytics",),
    }

Ancestors

Class variables

var CONTAINER_CLASS
var LOCALIZED_NAMES

Inherited members

class IMContactList (**kwargs)

Base class to use until we have a more specific folder implementation for this folder.

Expand source code
class IMContactList(WellknownFolder):
    CONTAINER_CLASS = "IPF.Contact.MOC.ImContactList"
    DISTINGUISHED_FOLDER_ID = "imcontactlist"
    supported_from = EXCHANGE_2013

Ancestors

Class variables

var CONTAINER_CLASS
var DISTINGUISHED_FOLDER_ID
var supported_from

Inherited members

class Inbox (**kwargs)
Expand source code
class Inbox(Messages):
    DISTINGUISHED_FOLDER_ID = "inbox"

    LOCALIZED_NAMES = {
        "da_DK": ("Indbakke",),
        "de_DE": ("Posteingang",),
        "en_US": ("Inbox",),
        "es_ES": ("Bandeja de entrada",),
        "fr_CA": ("Boîte de réception",),
        "nl_NL": ("Postvak IN",),
        "ru_RU": ("Входящие",),
        "sv_SE": ("Inkorgen",),
        "zh_CN": ("收件箱",),
    }

Ancestors

Class variables

var DISTINGUISHED_FOLDER_ID
var LOCALIZED_NAMES

Inherited members

class Journal (**kwargs)

Base class to use until we have a more specific folder implementation for this folder.

Expand source code
class Journal(WellknownFolder):
    CONTAINER_CLASS = "IPF.Journal"
    DISTINGUISHED_FOLDER_ID = "journal"

Ancestors

Class variables

var CONTAINER_CLASS
var DISTINGUISHED_FOLDER_ID

Inherited members

class JunkEmail (**kwargs)
Expand source code
class JunkEmail(Messages):
    DISTINGUISHED_FOLDER_ID = "junkemail"

    LOCALIZED_NAMES = {
        "da_DK": ("Uønsket e-mail",),
        "de_DE": ("Junk-E-Mail",),
        "en_US": ("Junk E-mail",),
        "es_ES": ("Correo no deseado",),
        "fr_CA": ("Courrier indésirables",),
        "nl_NL": ("Ongewenste e-mail",),
        "ru_RU": ("Нежелательная почта",),
        "sv_SE": ("Skräppost",),
        "zh_CN": ("垃圾邮件",),
    }

Ancestors

Class variables

var DISTINGUISHED_FOLDER_ID
var LOCALIZED_NAMES

Inherited members

class LocalFailures (**kwargs)

Base class to use until we have a more specific folder implementation for this folder.

Expand source code
class LocalFailures(WellknownFolder):
    DISTINGUISHED_FOLDER_ID = "localfailures"
    supported_from = EXCHANGE_2013

Ancestors

Class variables

var DISTINGUISHED_FOLDER_ID
var supported_from

Inherited members

class Location (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class Location(NonDeletableFolderMixin, Folder):
    LOCALIZED_NAMES = {
        None: ("Location",),
    }

Ancestors

Class variables

var LOCALIZED_NAMES

Inherited members

class MailboxAssociations (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class MailboxAssociations(NonDeletableFolderMixin, Folder):
    LOCALIZED_NAMES = {
        None: ("MailboxAssociations",),
    }

Ancestors

Class variables

var LOCALIZED_NAMES

Inherited members

class Messages (**kwargs)
Expand source code
class Messages(Folder):
    CONTAINER_CLASS = "IPF.Note"
    supported_item_models = (Message, MeetingRequest, MeetingResponse, MeetingCancellation)

Ancestors

Subclasses

Class variables

var CONTAINER_CLASS
var supported_item_models

Inherited members

class MsgFolderRoot (**kwargs)

Also known as the 'Top of Information Store' folder.

Expand source code
class MsgFolderRoot(WellknownFolder):
    """Also known as the 'Top of Information Store' folder."""

    DISTINGUISHED_FOLDER_ID = "msgfolderroot"
    LOCALIZED_NAMES = {
        None: ("Top of Information Store",),
        "da_DK": ("Informationslagerets øverste niveau",),
        "zh_CN": ("信息存储顶部",),
    }

Ancestors

Class variables

var DISTINGUISHED_FOLDER_ID
var LOCALIZED_NAMES

Inherited members

class MyContacts (**kwargs)

Base class to use until we have a more specific folder implementation for this folder.

Expand source code
class MyContacts(WellknownFolder):
    CONTAINER_CLASS = "IPF.Note"
    DISTINGUISHED_FOLDER_ID = "mycontacts"
    supported_from = EXCHANGE_2013

Ancestors

Class variables

var CONTAINER_CLASS
var DISTINGUISHED_FOLDER_ID
var supported_from

Inherited members

class MyContactsExtended (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class MyContactsExtended(NonDeletableFolderMixin, Contacts):
    CONTAINER_CLASS = "IPF.Note"
    LOCALIZED_NAMES = {
        None: ("MyContactsExtended",),
    }

Ancestors

Class variables

var CONTAINER_CLASS
var LOCALIZED_NAMES

Inherited members

class NonDeletableFolderMixin

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class NonDeletableFolderMixin:
    """A mixin for non-wellknown folders than that are not deletable."""

    @property
    def is_deletable(self):
        return False

Subclasses

Instance variables

var is_deletable
Expand source code
@property
def is_deletable(self):
    return False
class Notes (**kwargs)

Base class to use until we have a more specific folder implementation for this folder.

Expand source code
class Notes(WellknownFolder):
    CONTAINER_CLASS = "IPF.StickyNote"
    DISTINGUISHED_FOLDER_ID = "notes"
    LOCALIZED_NAMES = {
        "da_DK": ("Noter",),
    }

Ancestors

Class variables

var CONTAINER_CLASS
var DISTINGUISHED_FOLDER_ID
var LOCALIZED_NAMES

Inherited members

class OrganizationalContacts (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class OrganizationalContacts(NonDeletableFolderMixin, Contacts):
    DISTINGUISHED_FOLDER_ID = None
    CONTAINTER_CLASS = "IPF.Contact.OrganizationalContacts"
    LOCALIZED_NAMES = {
        None: ("Organizational Contacts",),
    }

Ancestors

Class variables

var CONTAINTER_CLASS
var DISTINGUISHED_FOLDER_ID
var LOCALIZED_NAMES

Inherited members

class Outbox (**kwargs)
Expand source code
class Outbox(Messages):
    DISTINGUISHED_FOLDER_ID = "outbox"

    LOCALIZED_NAMES = {
        "da_DK": ("Udbakke",),
        "de_DE": ("Postausgang",),
        "en_US": ("Outbox",),
        "es_ES": ("Bandeja de salida",),
        "fr_CA": ("Boîte d'envoi",),
        "nl_NL": ("Postvak UIT",),
        "ru_RU": ("Исходящие",),
        "sv_SE": ("Utkorgen",),
        "zh_CN": ("发件箱",),
    }

Ancestors

Class variables

var DISTINGUISHED_FOLDER_ID
var LOCALIZED_NAMES

Inherited members

class ParkedMessages (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class ParkedMessages(NonDeletableFolderMixin, Folder):
    CONTAINER_CLASS = None
    LOCALIZED_NAMES = {
        None: ("ParkedMessages",),
    }

Ancestors

Class variables

var CONTAINER_CLASS
var LOCALIZED_NAMES

Inherited members

class PassThroughSearchResults (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class PassThroughSearchResults(NonDeletableFolderMixin, Folder):
    CONTAINER_CLASS = "IPF.StoreItem.PassThroughSearchResults"
    LOCALIZED_NAMES = {
        None: ("Pass-Through Search Results",),
    }

Ancestors

Class variables

var CONTAINER_CLASS
var LOCALIZED_NAMES

Inherited members

class PdpProfileV2Secured (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class PdpProfileV2Secured(NonDeletableFolderMixin, Folder):
    CONTAINER_CLASS = "IPF.StoreItem.PdpProfileSecured"
    LOCALIZED_NAMES = {
        None: ("PdpProfileV2Secured",),
    }

Ancestors

Class variables

var CONTAINER_CLASS
var LOCALIZED_NAMES

Inherited members

class PeopleCentricConversationBuddies (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class PeopleCentricConversationBuddies(NonDeletableFolderMixin, Contacts):
    DISTINGUISHED_FOLDER_ID = None
    CONTAINTER_CLASS = "IPF.Contact.PeopleCentricConversationBuddies"
    LOCALIZED_NAMES = {
        None: ("PeopleCentricConversation Buddies",),
    }

Ancestors

Class variables

var CONTAINTER_CLASS
var DISTINGUISHED_FOLDER_ID
var LOCALIZED_NAMES

Inherited members

class PeopleConnect (**kwargs)

Base class to use until we have a more specific folder implementation for this folder.

Expand source code
class PeopleConnect(WellknownFolder):
    DISTINGUISHED_FOLDER_ID = "peopleconnect"
    supported_from = EXCHANGE_2013

Ancestors

Class variables

var DISTINGUISHED_FOLDER_ID
var supported_from

Inherited members

class PublicFoldersRoot (**kwargs)

The root of the public folders hierarchy. Not available on all mailboxes.

Expand source code
class PublicFoldersRoot(RootOfHierarchy):
    """The root of the public folders hierarchy. Not available on all mailboxes."""

    DISTINGUISHED_FOLDER_ID = "publicfoldersroot"
    DEFAULT_FOLDER_TRAVERSAL_DEPTH = SHALLOW
    supported_from = EXCHANGE_2007_SP1

    def get_children(self, folder):
        # EWS does not allow deep traversal of public folders, so self._folders_map will only populate the top-level
        # subfolders. To traverse public folders at arbitrary depth, we need to get child folders on demand.

        # Let's check if this folder already has any cached children. If so, assume we can just return those.
        children = list(super().get_children(folder=folder))
        if children:
            # Return a generator like our parent does
            yield from children
            return

        # Also return early if the server told us that there are no child folders.
        if folder.child_folder_count == 0:
            return

        children_map = {}
        try:
            for f in (
                SingleFolderQuerySet(account=self.account, folder=folder)
                .depth(self.DEFAULT_FOLDER_TRAVERSAL_DEPTH)
                .all()
            ):
                if isinstance(f, MISSING_FOLDER_ERRORS):
                    # We were unlucky. The folder disappeared between the FindFolder and the GetFolder calls
                    continue
                if isinstance(f, Exception):
                    raise f
                children_map[f.id] = f
        except ErrorAccessDenied:
            # No access to this folder
            pass

        # Let's update the cache atomically, to avoid partial reads of the cache.
        with self._subfolders_lock:
            self._subfolders.update(children_map)

        # Child folders have been cached now. Try super().get_children() again.
        yield from super().get_children(folder=folder)

Ancestors

Class variables

var DEFAULT_FOLDER_TRAVERSAL_DEPTH
var DISTINGUISHED_FOLDER_ID
var supported_from

Methods

def get_children(self, folder)
Expand source code
def get_children(self, folder):
    # EWS does not allow deep traversal of public folders, so self._folders_map will only populate the top-level
    # subfolders. To traverse public folders at arbitrary depth, we need to get child folders on demand.

    # Let's check if this folder already has any cached children. If so, assume we can just return those.
    children = list(super().get_children(folder=folder))
    if children:
        # Return a generator like our parent does
        yield from children
        return

    # Also return early if the server told us that there are no child folders.
    if folder.child_folder_count == 0:
        return

    children_map = {}
    try:
        for f in (
            SingleFolderQuerySet(account=self.account, folder=folder)
            .depth(self.DEFAULT_FOLDER_TRAVERSAL_DEPTH)
            .all()
        ):
            if isinstance(f, MISSING_FOLDER_ERRORS):
                # We were unlucky. The folder disappeared between the FindFolder and the GetFolder calls
                continue
            if isinstance(f, Exception):
                raise f
            children_map[f.id] = f
    except ErrorAccessDenied:
        # No access to this folder
        pass

    # Let's update the cache atomically, to avoid partial reads of the cache.
    with self._subfolders_lock:
        self._subfolders.update(children_map)

    # Child folders have been cached now. Try super().get_children() again.
    yield from super().get_children(folder=folder)

Inherited members

class QuickContacts (**kwargs)

Base class to use until we have a more specific folder implementation for this folder.

Expand source code
class QuickContacts(WellknownFolder):
    CONTAINER_CLASS = "IPF.Contact.MOC.QuickContacts"
    DISTINGUISHED_FOLDER_ID = "quickcontacts"
    supported_from = EXCHANGE_2013

Ancestors

Class variables

var CONTAINER_CLASS
var DISTINGUISHED_FOLDER_ID
var supported_from

Inherited members

class RSSFeeds (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class RSSFeeds(NonDeletableFolderMixin, Folder):
    CONTAINER_CLASS = "IPF.Note.OutlookHomepage"
    LOCALIZED_NAMES = {
        None: ("RSS Feeds",),
    }

Ancestors

Class variables

var CONTAINER_CLASS
var LOCALIZED_NAMES

Inherited members

class RecipientCache (**kwargs)
Expand source code
class RecipientCache(Contacts):
    DISTINGUISHED_FOLDER_ID = "recipientcache"
    CONTAINER_CLASS = "IPF.Contact.RecipientCache"
    supported_from = EXCHANGE_2013

    LOCALIZED_NAMES = {}

Ancestors

Class variables

var CONTAINER_CLASS
var DISTINGUISHED_FOLDER_ID
var LOCALIZED_NAMES
var supported_from

Inherited members

class RecoverableItemsDeletions (**kwargs)

Base class to use until we have a more specific folder implementation for this folder.

Expand source code
class RecoverableItemsDeletions(WellknownFolder):
    DISTINGUISHED_FOLDER_ID = "recoverableitemsdeletions"
    supported_from = EXCHANGE_2010_SP1

Ancestors

Class variables

var DISTINGUISHED_FOLDER_ID
var supported_from

Inherited members

class RecoverableItemsPurges (**kwargs)

Base class to use until we have a more specific folder implementation for this folder.

Expand source code
class RecoverableItemsPurges(WellknownFolder):
    DISTINGUISHED_FOLDER_ID = "recoverableitemspurges"
    supported_from = EXCHANGE_2010_SP1

Ancestors

Class variables

var DISTINGUISHED_FOLDER_ID
var supported_from

Inherited members

class RecoverableItemsRoot (**kwargs)

Base class to use until we have a more specific folder implementation for this folder.

Expand source code
class RecoverableItemsRoot(WellknownFolder):
    DISTINGUISHED_FOLDER_ID = "recoverableitemsroot"
    supported_from = EXCHANGE_2010_SP1

Ancestors

Class variables

var DISTINGUISHED_FOLDER_ID
var supported_from

Inherited members

class RecoverableItemsVersions (**kwargs)

Base class to use until we have a more specific folder implementation for this folder.

Expand source code
class RecoverableItemsVersions(WellknownFolder):
    DISTINGUISHED_FOLDER_ID = "recoverableitemsversions"
    supported_from = EXCHANGE_2010_SP1

Ancestors

Class variables

var DISTINGUISHED_FOLDER_ID
var supported_from

Inherited members

class RecoveryPoints (**kwargs)
Expand source code
class RecoveryPoints(Folder):
    CONTAINER_CLASS = "IPF.StoreItem.RecoveryPoints"

Ancestors

Class variables

var CONTAINER_CLASS

Inherited members

class Reminders (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class Reminders(NonDeletableFolderMixin, Folder):
    CONTAINER_CLASS = "Outlook.Reminder"
    LOCALIZED_NAMES = {
        "da_DK": ("Påmindelser",),
    }

Ancestors

Class variables

var CONTAINER_CLASS
var LOCALIZED_NAMES

Inherited members

class Root (**kwargs)

The root of the standard folder hierarchy.

Expand source code
class Root(RootOfHierarchy):
    """The root of the standard folder hierarchy."""

    DISTINGUISHED_FOLDER_ID = "root"
    WELLKNOWN_FOLDERS = WELLKNOWN_FOLDERS_IN_ROOT

    @property
    def tois(self):
        # 'Top of Information Store' is a folder available in some Exchange accounts. It usually contains the
        # distinguished folders belonging to the account (inbox, calendar, trash etc.).
        return self.get_default_folder(MsgFolderRoot)

    def get_default_folder(self, folder_cls):
        try:
            return super().get_default_folder(folder_cls)
        except MISSING_FOLDER_ERRORS:
            pass

        # Try to pick a suitable default folder. we do this by:
        #  1. Searching the full folder list for a folder with the distinguished folder name
        #  2. Searching TOIS for a direct child folder of the same type that is marked as distinguished
        #  3. Searching TOIS for a direct child folder of the same type that has a localized name
        #  4. Searching root for a direct child folder of the same type that is marked as distinguished
        #  5. Searching root for a direct child folder of the same type that has a localized name
        log.debug("Searching default %s folder in full folder list", folder_cls)

        for f in self._folders_map.values():
            # Require exact type, to avoid matching with subclasses (e.g. RecipientCache and Contacts)
            if f.__class__ == folder_cls and f.has_distinguished_name:
                log.debug("Found cached %s folder with default distinguished name", folder_cls)
                return f

        # Try direct children of TOIS first, unless we're trying to get the TOIS folder
        if folder_cls != MsgFolderRoot:
            try:
                return self._get_candidate(folder_cls=folder_cls, folder_coll=self.tois.children)
            except MISSING_FOLDER_ERRORS:
                # No candidates, or TOIS does not exist, or we don't have access to TOIS
                pass

        # Finally, try direct children of root
        return self._get_candidate(folder_cls=folder_cls, folder_coll=self.children)

    def _get_candidate(self, folder_cls, folder_coll):
        # Look for a single useful folder of type folder_cls in folder_coll
        same_type = [f for f in folder_coll if f.__class__ == folder_cls]
        are_distinguished = [f for f in same_type if f.is_distinguished]
        if are_distinguished:
            candidates = are_distinguished
        else:
            candidates = [f for f in same_type if f.name.lower() in folder_cls.localized_names(self.account.locale)]
        if not candidates:
            raise ErrorFolderNotFound(f"No usable default {folder_cls} folders")
        if len(candidates) > 1:
            raise ValueError(f"Multiple possible default {folder_cls} folders: {[f.name for f in candidates]}")
        candidate = candidates[0]
        if candidate.is_distinguished:
            log.debug("Found distinguished %s folder", folder_cls)
        else:
            log.debug("Found %s folder with localized name %s", folder_cls, candidate.name)
        return candidate

Ancestors

Class variables

var DISTINGUISHED_FOLDER_ID
var WELLKNOWN_FOLDERS

Instance variables

var tois
Expand source code
@property
def tois(self):
    # 'Top of Information Store' is a folder available in some Exchange accounts. It usually contains the
    # distinguished folders belonging to the account (inbox, calendar, trash etc.).
    return self.get_default_folder(MsgFolderRoot)

Inherited members

class RootOfHierarchy (**kwargs)

Base class for folders that implement the root of a folder hierarchy.

Expand source code
class RootOfHierarchy(BaseFolder, metaclass=EWSMeta):
    """Base class for folders that implement the root of a folder hierarchy."""

    # A list of wellknown, or "distinguished", folders that are belong in this folder hierarchy. See
    # https://docs.microsoft.com/en-us/dotnet/api/microsoft.exchange.webservices.data.wellknownfoldername
    # and https://docs.microsoft.com/en-us/exchange/client-developer/web-service-reference/distinguishedfolderid
    # 'RootOfHierarchy' subclasses must not be in this list.
    WELLKNOWN_FOLDERS = []

    _subfolders_lock = Lock()

    # This folder type also has 'folder:PermissionSet' on some server versions, but requesting it sometimes causes
    # 'ErrorAccessDenied', as reported by some users. Ignore it entirely for root folders - it's usefulness is
    # deemed minimal at best.
    effective_rights = EffectiveRightsField(
        field_uri="folder:EffectiveRights", is_read_only=True, supported_from=EXCHANGE_2007_SP1
    )

    __slots__ = "_account", "_subfolders"

    # A special folder that acts as the top of a folder hierarchy. Finds and caches subfolders at arbitrary depth.
    def __init__(self, **kwargs):
        self._account = kwargs.pop("account", None)  # A pointer back to the account holding the folder hierarchy
        super().__init__(**kwargs)
        self._subfolders = None  # See self._folders_map()

    @property
    def account(self):
        return self._account

    @property
    def root(self):
        return self

    @property
    def parent(self):
        return None

    @classmethod
    def register(cls, *args, **kwargs):
        if cls is not RootOfHierarchy:
            raise TypeError("For folder roots, custom fields must be registered on the RootOfHierarchy class")
        return super().register(*args, **kwargs)

    @classmethod
    def deregister(cls, *args, **kwargs):
        if cls is not RootOfHierarchy:
            raise TypeError("For folder roots, custom fields must be registered on the RootOfHierarchy class")
        return super().deregister(*args, **kwargs)

    def get_folder(self, folder):
        if not folder.id:
            raise ValueError("'folder' must have an ID")
        return self._folders_map.get(folder.id)

    def add_folder(self, folder):
        if not folder.id:
            raise ValueError("'folder' must have an ID")
        self._folders_map[folder.id] = folder

    def update_folder(self, folder):
        if not folder.id:
            raise ValueError("'folder' must have an ID")
        self._folders_map[folder.id] = folder

    def remove_folder(self, folder):
        if not folder.id:
            raise ValueError("'folder' must have an ID")
        try:
            del self._folders_map[folder.id]
        except KeyError:
            pass

    def clear_cache(self):
        with self._subfolders_lock:
            self._subfolders = None

    def get_children(self, folder):
        for f in self._folders_map.values():
            if not f.parent:
                continue
            if f.parent.id == folder.id:
                yield f

    @classmethod
    def get_distinguished(cls, account):
        """Get the distinguished folder for this folder class.

        :param account:
        """
        if not cls.DISTINGUISHED_FOLDER_ID:
            raise ValueError(f"Class {cls} must have a DISTINGUISHED_FOLDER_ID value")
        try:
            return cls.resolve(
                account=account, folder=cls(account=account, name=cls.DISTINGUISHED_FOLDER_ID, is_distinguished=True)
            )
        except MISSING_FOLDER_ERRORS:
            raise ErrorFolderNotFound(f"Could not find distinguished folder {cls.DISTINGUISHED_FOLDER_ID}")

    def get_default_folder(self, folder_cls):
        """Return the distinguished folder instance of type folder_cls belonging to this account. If no distinguished
        folder was found, try as best we can to return the default folder of type 'folder_cls'
        """
        if not folder_cls.DISTINGUISHED_FOLDER_ID:
            raise ValueError(f"'folder_cls' {folder_cls} must have a DISTINGUISHED_FOLDER_ID value")
        # Use cached distinguished folder instance, but only if cache has already been prepped. This is an optimization
        # for accessing e.g. 'account.contacts' without fetching all folders of the account.
        if self._subfolders is not None:
            for f in self._folders_map.values():
                # Require exact class, to not match subclasses, e.g. RecipientCache instead of Contacts
                if f.__class__ == folder_cls and f.is_distinguished:
                    log.debug("Found cached distinguished %s folder", folder_cls)
                    return f
        try:
            log.debug("Requesting distinguished %s folder explicitly", folder_cls)
            return folder_cls.get_distinguished(root=self)
        except ErrorAccessDenied:
            # Maybe we just don't have GetFolder access? Try FindItems instead
            log.debug("Testing default %s folder with FindItem", folder_cls)
            fld = folder_cls(root=self, name=folder_cls.DISTINGUISHED_FOLDER_ID, is_distinguished=True)
            fld.test_access()
            return self._folders_map.get(fld.id, fld)  # Use cached instance if available
        except MISSING_FOLDER_ERRORS:
            # The Exchange server does not return a distinguished folder of this type
            pass
        raise ErrorFolderNotFound(f"No usable default {folder_cls} folders")

    @property
    def _folders_map(self):
        if self._subfolders is not None:
            return self._subfolders

        with self._subfolders_lock:
            # Map root, and all subfolders of root, at arbitrary depth by folder ID. First get distinguished folders,
            # so we are sure to apply the correct Folder class, then fetch all subfolders of this root.
            folders_map = {self.id: self}
            distinguished_folders = [
                cls(root=self, name=cls.DISTINGUISHED_FOLDER_ID, is_distinguished=True)
                for cls in self.WELLKNOWN_FOLDERS
                if cls.get_folder_allowed and cls.supports_version(self.account.version)
            ]
            for f in FolderCollection(account=self.account, folders=distinguished_folders).resolve():
                if isinstance(f, MISSING_FOLDER_ERRORS):
                    # This is just a distinguished folder the server does not have
                    continue
                if isinstance(f, ErrorInvalidOperation):
                    # This is probably a distinguished folder the server does not have. We previously tested the exact
                    # error message (f.value), but some Exchange servers return localized error messages, so that's not
                    # possible to do reliably.
                    continue
                if isinstance(f, ErrorAccessDenied):
                    # We may not have GetFolder access, either to this folder or at all
                    continue
                if isinstance(f, Exception):
                    raise f
                folders_map[f.id] = f
            for f in (
                SingleFolderQuerySet(account=self.account, folder=self).depth(self.DEFAULT_FOLDER_TRAVERSAL_DEPTH).all()
            ):
                if isinstance(f, ErrorAccessDenied):
                    # We may not have FindFolder access, or GetFolder access, either to this folder or at all
                    continue
                if isinstance(f, MISSING_FOLDER_ERRORS):
                    # We were unlucky. The folder disappeared between the FindFolder and the GetFolder calls
                    continue
                if isinstance(f, Exception):
                    raise f
                if f.id in folders_map:
                    # Already exists. Probably a distinguished folder
                    continue
                folders_map[f.id] = f
            self._subfolders = folders_map
            return folders_map

    @classmethod
    def from_xml(cls, elem, account):
        kwargs = cls._kwargs_from_elem(elem=elem, account=account)
        cls._clear(elem)
        return cls(account=account, **kwargs)

    @classmethod
    def folder_cls_from_folder_name(cls, folder_name, locale):
        """Return the folder class that matches a localized folder name.

        :param folder_name:
        :param locale: a string, e.g. 'da_DK'
        """
        for folder_cls in cls.WELLKNOWN_FOLDERS + NON_DELETABLE_FOLDERS + MISC_FOLDERS:
            if folder_name.lower() in folder_cls.localized_names(locale):
                return folder_cls
        raise KeyError()

    def __repr__(self):
        # Let's not create an infinite loop when printing self.root
        return self.__class__.__name__ + repr(
            (
                self.account,
                "[self]",
                self.name,
                self.total_count,
                self.unread_count,
                self.child_folder_count,
                self.folder_class,
                self.id,
                self.changekey,
            )
        )

Ancestors

Subclasses

Class variables

var FIELDS
var WELLKNOWN_FOLDERS

Static methods

def folder_cls_from_folder_name(folder_name, locale)

Return the folder class that matches a localized folder name.

:param folder_name: :param locale: a string, e.g. 'da_DK'

Expand source code
@classmethod
def folder_cls_from_folder_name(cls, folder_name, locale):
    """Return the folder class that matches a localized folder name.

    :param folder_name:
    :param locale: a string, e.g. 'da_DK'
    """
    for folder_cls in cls.WELLKNOWN_FOLDERS + NON_DELETABLE_FOLDERS + MISC_FOLDERS:
        if folder_name.lower() in folder_cls.localized_names(locale):
            return folder_cls
    raise KeyError()
def from_xml(elem, account)
Expand source code
@classmethod
def from_xml(cls, elem, account):
    kwargs = cls._kwargs_from_elem(elem=elem, account=account)
    cls._clear(elem)
    return cls(account=account, **kwargs)
def get_distinguished(account)

Get the distinguished folder for this folder class.

:param account:

Expand source code
@classmethod
def get_distinguished(cls, account):
    """Get the distinguished folder for this folder class.

    :param account:
    """
    if not cls.DISTINGUISHED_FOLDER_ID:
        raise ValueError(f"Class {cls} must have a DISTINGUISHED_FOLDER_ID value")
    try:
        return cls.resolve(
            account=account, folder=cls(account=account, name=cls.DISTINGUISHED_FOLDER_ID, is_distinguished=True)
        )
    except MISSING_FOLDER_ERRORS:
        raise ErrorFolderNotFound(f"Could not find distinguished folder {cls.DISTINGUISHED_FOLDER_ID}")

Instance variables

var effective_rights

Methods

def add_folder(self, folder)
Expand source code
def add_folder(self, folder):
    if not folder.id:
        raise ValueError("'folder' must have an ID")
    self._folders_map[folder.id] = folder
def clear_cache(self)
Expand source code
def clear_cache(self):
    with self._subfolders_lock:
        self._subfolders = None
def get_children(self, folder)
Expand source code
def get_children(self, folder):
    for f in self._folders_map.values():
        if not f.parent:
            continue
        if f.parent.id == folder.id:
            yield f
def get_default_folder(self, folder_cls)

Return the distinguished folder instance of type folder_cls belonging to this account. If no distinguished folder was found, try as best we can to return the default folder of type 'folder_cls'

Expand source code
def get_default_folder(self, folder_cls):
    """Return the distinguished folder instance of type folder_cls belonging to this account. If no distinguished
    folder was found, try as best we can to return the default folder of type 'folder_cls'
    """
    if not folder_cls.DISTINGUISHED_FOLDER_ID:
        raise ValueError(f"'folder_cls' {folder_cls} must have a DISTINGUISHED_FOLDER_ID value")
    # Use cached distinguished folder instance, but only if cache has already been prepped. This is an optimization
    # for accessing e.g. 'account.contacts' without fetching all folders of the account.
    if self._subfolders is not None:
        for f in self._folders_map.values():
            # Require exact class, to not match subclasses, e.g. RecipientCache instead of Contacts
            if f.__class__ == folder_cls and f.is_distinguished:
                log.debug("Found cached distinguished %s folder", folder_cls)
                return f
    try:
        log.debug("Requesting distinguished %s folder explicitly", folder_cls)
        return folder_cls.get_distinguished(root=self)
    except ErrorAccessDenied:
        # Maybe we just don't have GetFolder access? Try FindItems instead
        log.debug("Testing default %s folder with FindItem", folder_cls)
        fld = folder_cls(root=self, name=folder_cls.DISTINGUISHED_FOLDER_ID, is_distinguished=True)
        fld.test_access()
        return self._folders_map.get(fld.id, fld)  # Use cached instance if available
    except MISSING_FOLDER_ERRORS:
        # The Exchange server does not return a distinguished folder of this type
        pass
    raise ErrorFolderNotFound(f"No usable default {folder_cls} folders")
def get_folder(self, folder)
Expand source code
def get_folder(self, folder):
    if not folder.id:
        raise ValueError("'folder' must have an ID")
    return self._folders_map.get(folder.id)
def remove_folder(self, folder)
Expand source code
def remove_folder(self, folder):
    if not folder.id:
        raise ValueError("'folder' must have an ID")
    try:
        del self._folders_map[folder.id]
    except KeyError:
        pass
def update_folder(self, folder)
Expand source code
def update_folder(self, folder):
    if not folder.id:
        raise ValueError("'folder' must have an ID")
    self._folders_map[folder.id] = folder

Inherited members

class Schedule (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class Schedule(NonDeletableFolderMixin, Folder):
    LOCALIZED_NAMES = {
        None: ("Schedule",),
    }

Ancestors

Class variables

var LOCALIZED_NAMES

Inherited members

class SearchFolders (**kwargs)

Base class to use until we have a more specific folder implementation for this folder.

Expand source code
class SearchFolders(WellknownFolder):
    DISTINGUISHED_FOLDER_ID = "searchfolders"

Ancestors

Class variables

var DISTINGUISHED_FOLDER_ID

Inherited members

class SentItems (**kwargs)
Expand source code
class SentItems(Messages):
    DISTINGUISHED_FOLDER_ID = "sentitems"

    LOCALIZED_NAMES = {
        "da_DK": ("Sendt post",),
        "de_DE": ("Gesendete Elemente",),
        "en_US": ("Sent Items",),
        "es_ES": ("Elementos enviados",),
        "fr_CA": ("Éléments envoyés",),
        "nl_NL": ("Verzonden items",),
        "ru_RU": ("Отправленные",),
        "sv_SE": ("Skickat",),
        "zh_CN": ("已发送邮件",),
    }

Ancestors

Class variables

var DISTINGUISHED_FOLDER_ID
var LOCALIZED_NAMES

Inherited members

class ServerFailures (**kwargs)

Base class to use until we have a more specific folder implementation for this folder.

Expand source code
class ServerFailures(WellknownFolder):
    DISTINGUISHED_FOLDER_ID = "serverfailures"
    supported_from = EXCHANGE_2013

Ancestors

Class variables

var DISTINGUISHED_FOLDER_ID
var supported_from

Inherited members

class Sharing (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class Sharing(NonDeletableFolderMixin, Folder):
    CONTAINER_CLASS = "IPF.Note"
    LOCALIZED_NAMES = {
        None: ("Sharing",),
    }

Ancestors

Class variables

var CONTAINER_CLASS
var LOCALIZED_NAMES

Inherited members

class Shortcuts (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class Shortcuts(NonDeletableFolderMixin, Folder):
    LOCALIZED_NAMES = {
        None: ("Shortcuts",),
    }

Ancestors

Class variables

var LOCALIZED_NAMES

Inherited members

class Signal (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class Signal(NonDeletableFolderMixin, Folder):
    CONTAINER_CLASS = "IPF.StoreItem.Signal"
    LOCALIZED_NAMES = {
        None: ("Signal",),
    }

Ancestors

Class variables

var CONTAINER_CLASS
var LOCALIZED_NAMES

Inherited members

class SingleFolderQuerySet (account, folder)

A helper class with simpler argument types.

Expand source code
class SingleFolderQuerySet(FolderQuerySet):
    """A helper class with simpler argument types."""

    def __init__(self, account, folder):
        from .collections import FolderCollection

        folder_collection = FolderCollection(account=account, folders=[folder])
        super().__init__(folder_collection=folder_collection)

    def _copy_cls(self):
        return self.__class__(account=self.folder_collection.account, folder=self.folder_collection.folders[0])

    def resolve(self):
        return list(self.folder_collection.resolve())[0]

Ancestors

Methods

def resolve(self)
Expand source code
def resolve(self):
    return list(self.folder_collection.resolve())[0]

Inherited members

class SkypeTeamsMessages (**kwargs)
Expand source code
class SkypeTeamsMessages(Folder):
    CONTAINER_CLASS = "IPF.SkypeTeams.Message"
    LOCALIZED_NAMES = {
        None: ("Team-chat",),
    }

Ancestors

Class variables

var CONTAINER_CLASS
var LOCALIZED_NAMES

Inherited members

class SmsAndChatsSync (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class SmsAndChatsSync(NonDeletableFolderMixin, Folder):
    CONTAINER_CLASS = "IPF.SmsAndChatsSync"
    LOCALIZED_NAMES = {
        None: ("SmsAndChatsSync",),
    }

Ancestors

Class variables

var CONTAINER_CLASS
var LOCALIZED_NAMES

Inherited members

class SpoolerQueue (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class SpoolerQueue(NonDeletableFolderMixin, Folder):
    LOCALIZED_NAMES = {
        None: ("Spooler Queue",),
    }

Ancestors

Class variables

var LOCALIZED_NAMES

Inherited members

class SwssItems (**kwargs)
Expand source code
class SwssItems(Folder):
    CONTAINER_CLASS = "IPF.StoreItem.SwssItems"

Ancestors

Class variables

var CONTAINER_CLASS

Inherited members

class SyncIssues (**kwargs)

Base class to use until we have a more specific folder implementation for this folder.

Expand source code
class SyncIssues(WellknownFolder):
    CONTAINER_CLASS = "IPF.Note"
    DISTINGUISHED_FOLDER_ID = "syncissues"
    supported_from = EXCHANGE_2013

Ancestors

Class variables

var CONTAINER_CLASS
var DISTINGUISHED_FOLDER_ID
var supported_from

Inherited members

class System (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class System(NonDeletableFolderMixin, Folder):
    LOCALIZED_NAMES = {
        None: ("System",),
    }
    get_folder_allowed = False

Ancestors

Class variables

var LOCALIZED_NAMES
var get_folder_allowed

Inherited members

class Tasks (**kwargs)
Expand source code
class Tasks(Folder):
    DISTINGUISHED_FOLDER_ID = "tasks"
    CONTAINER_CLASS = "IPF.Task"
    supported_item_models = (Task,)

    LOCALIZED_NAMES = {
        "da_DK": ("Opgaver",),
        "de_DE": ("Aufgaben",),
        "en_US": ("Tasks",),
        "es_ES": ("Tareas",),
        "fr_CA": ("Tâches",),
        "nl_NL": ("Taken",),
        "ru_RU": ("Задачи",),
        "sv_SE": ("Uppgifter",),
        "zh_CN": ("任务",),
    }

Ancestors

Class variables

var CONTAINER_CLASS
var DISTINGUISHED_FOLDER_ID
var LOCALIZED_NAMES
var supported_item_models

Inherited members

class TemporarySaves (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class TemporarySaves(NonDeletableFolderMixin, Folder):
    LOCALIZED_NAMES = {
        None: ("TemporarySaves",),
    }

Ancestors

Class variables

var LOCALIZED_NAMES

Inherited members

class ToDoSearch (**kwargs)

Base class to use until we have a more specific folder implementation for this folder.

Expand source code
class ToDoSearch(WellknownFolder):
    CONTAINER_CLASS = "IPF.Task"
    DISTINGUISHED_FOLDER_ID = "todosearch"
    supported_from = EXCHANGE_2013

    LOCALIZED_NAMES = {
        None: ("To-Do Search",),
    }

Ancestors

Class variables

var CONTAINER_CLASS
var DISTINGUISHED_FOLDER_ID
var LOCALIZED_NAMES
var supported_from

Inherited members

class Views (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class Views(NonDeletableFolderMixin, Folder):
    LOCALIZED_NAMES = {
        None: ("Views",),
    }

Ancestors

Class variables

var LOCALIZED_NAMES

Inherited members

class VoiceMail (**kwargs)

Base class to use until we have a more specific folder implementation for this folder.

Expand source code
class VoiceMail(WellknownFolder):
    DISTINGUISHED_FOLDER_ID = "voicemail"
    CONTAINER_CLASS = "IPF.Note.Microsoft.Voicemail"
    LOCALIZED_NAMES = {
        None: ("Voice Mail",),
    }

Ancestors

Class variables

var CONTAINER_CLASS
var DISTINGUISHED_FOLDER_ID
var LOCALIZED_NAMES

Inherited members

class WellknownFolder (**kwargs)

Base class to use until we have a more specific folder implementation for this folder.

Expand source code
class WellknownFolder(Folder, metaclass=EWSMeta):
    """Base class to use until we have a more specific folder implementation for this folder."""

    supported_item_models = ITEM_CLASSES

Ancestors

Subclasses

Class variables

var supported_item_models

Inherited members

class WorkingSet (**kwargs)

A mixin for non-wellknown folders than that are not deletable.

Expand source code
class WorkingSet(NonDeletableFolderMixin, Folder):
    LOCALIZED_NAMES = {
        None: ("Working Set",),
    }

Ancestors

Class variables

var LOCALIZED_NAMES

Inherited members