Client for Microsoft Exchange Web Services (EWS) providing Django-style ORM interface for Exchange mailboxes.
—
Advanced functionality including bulk operations, extended properties, impersonation, and low-level EWS service access for complex Exchange operations.
def bulk_create(account: Account, items: list, chunk_size: int = None) -> list:
"""Create multiple items efficiently."""
def bulk_update(account: Account, items: list, chunk_size: int = None) -> list:
"""Update multiple items efficiently."""
def bulk_delete(account: Account, items: list, chunk_size: int = None, delete_type: str = 'MoveToDeletedItems') -> None:
"""Delete multiple items efficiently."""
def export_items(account: Account, items: list) -> list:
"""Export items to their native EWS format."""
def upload_items(account: Account, data: list, folder: Folder) -> list:
"""Upload items from their native EWS format."""class ExtendedProperty:
def __init__(
self,
distinguished_property_set_id: str = None,
property_set_id: str = None,
property_tag: int = None,
property_name: str = None,
property_id: int = None,
property_type: str = None
):
"""Create custom Exchange property."""
value: any
# Property set constants
COMMON: str = 'Common'
PUBLIC_STRINGS: str = 'PublicStrings'
APPOINTMENT: str = 'Appointment'
MEETING: str = 'Meeting'
TASK: str = 'Task'
ADDRESS: str = 'Address'class BaseProtocol:
def __init__(
self,
config: Configuration,
retry_policy: RetryPolicy = None
):
"""Low-level EWS protocol handler."""
class FailFast:
"""Retry policy that fails immediately on errors."""
class FaultTolerance:
def __init__(self, max_wait: int = 3600):
"""Retry policy with fault tolerance."""
class NoVerifyHTTPAdapter:
"""HTTP adapter that skips SSL verification."""
class TLSClientAuth:
def __init__(self, ca_cert: str, cert_file: str, key_file: str):
"""TLS client certificate authentication."""class Version:
def __init__(self, build=None):
"""Exchange server version information."""
build: Build
api_version: str
class Build:
def __init__(self, major_version: int, minor_version: int, major_build_number: int, minor_build_number: int = 0):
"""Exchange server build information."""
major_version: int
minor_version: int
major_build_number: int
minor_build_number: int
def __str__(self) -> str:
"""String representation of build version."""def get_streaming_events(
account,
subscription_id: str,
connection_timeout: int = 30
):
"""Get streaming notifications for folder changes."""
def subscribe_to_notifications(
account,
folders,
event_types
) -> str:
"""Subscribe to folder change notifications."""# Base exception classes
class EWSError(Exception):
"""Base EWS error."""
class TransportError(EWSError):
"""Transport/connection error."""
class ProtocolError(EWSError):
"""EWS protocol error."""
class ResponseMessageError(EWSError):
"""EWS response message error."""
class UnauthorizedError(EWSError):
"""Authentication/authorization error."""
class RateLimitError(EWSError):
"""Rate limiting error."""Usage examples:
from exchangelib import ExtendedProperty, FailFast, bulk_create
# Use extended properties
custom_prop = ExtendedProperty(
distinguished_property_set_id=ExtendedProperty.PUBLIC_STRINGS,
property_name='CustomField',
property_type='String'
)
message = Message(
account=account,
subject='Custom Property Test'
)
message.extended_properties = [custom_prop]
custom_prop.value = 'Custom Value'
message.save()
# Bulk operations
messages = [
Message(account=account, subject=f'Bulk {i}')
for i in range(100)
]
created = bulk_create(account, messages)
# Error handling
try:
account.inbox.all().count()
except UnauthorizedError:
print("Authentication failed")
except RateLimitError:
print("Rate limited, retry later")Install with Tessl CLI
npx tessl i tessl/pypi-exchangelib