Email

Note

It’s required to install email from Optional dependencies section.

Outlook OAuth2

Note

In case you’ll be using Outlook OAuth2.0 authorization, you need to create & configure new application with the Microsoft identity platform. By the following steps:
Register an application
(Optional) In case of using Public(Delegated) Authorization option, you have to enable Allow public client flows in Authentication section
Add credentials, we currently support only Client’s Secret
Add corresponding permissions for Mail, in section API Permissions.

Application permissions - App runs as a background service or daemon without a signed-in user.

Delegated permissions - App needs to access the API as the signed-in user.

Required Permissions based on Authorization option
Permission	Application	(Optional) Delegated
`Mail.ReadWrite`	✅	✅
`Mail.Send`	✅	✅
`Mail.ReadWrite.Shared`	❌	(for shared mailbox) ✅
`Mail.Send.Shared`	❌	(for shared mailbox) ✅

../_images/outlook_permissions.png — Correct configuration in Azure portal.

Warning

Important, as Application permissions allow a user to access any mailbox, Administrators can configure application access policy to limit app access to specific mailboxes and not to all the mailboxes in the organization.

If you need access to a shared mailbox (e.g. orders@customer.com), the application access policy needs to be applied to a security-group into which the shared mailbox and user belong to. More info can be found here.

Client

class aiviro.modules.email.EmailClient

Email client which uses IMAP and SMTP protocols to communicate with server.

Example:

>>> from aiviro.modules.email import EmailClient
>>> from datetime import datetime
>>> client = EmailClient()
>>> # To receive emails you need to add imap server.
>>> client.setup_imap_basic_auth("<IMAP_SERVER>", "<EMAIL_ADDRESS>", "<EMAIL_PASSWORD>")

>>> # Go through all messages
>>> for email in client.inbox.all():
...     print(email.sender.full, email.subject, email.datetime)
...     email.set_seen(True)

>>> # Go through unseen messages
>>> for email in client.inbox.query(seen=False):
...     print(email.sender.full, email.subject, email.datetime)
...     print("Has attachments:", len(email.file_attachments) != 0)
...
...     # Download all attachments to this mail
...     for attachment in email.file_attachments:
...         with open(attachment.filename, "wb") as f:
...             f.write(attachment.content)

>>> # To send messages you need to connect to SMTP.
>>> client.setup_smtp_basic_auth("<SMTP_SERVER>")

>>> # Basic sending of message
>>> client.send_mail(
...     "recipient@foo.bar",
...     "Hello from Python",
...     "This message was sent by Aiviro",
... )

setup_imap_basic_auth(server: str, username: str, password: str, port: int | None = None, starttls: bool = False, oauth2: bool = False, no_verify: bool = False, unencrypted: bool = False, plain_login: bool = False) → EmailClient

Setups the IMAP connection with a basic username-password authorization for future use.

Parameters:

server – server IP or hostname
port – port for IMAP service, if None, default port will be used - 993(TLS), 143(Unencrypted)
username – username to this server
password – password to this server
starttls – set True if connection should be use starttls
oauth2 – If True, ‘password’ argument is used as access_token for oauth2 authentication
no_verify – If True, SSL certificate will not be verified
unencrypted – If True, connection will be unencrypted, with no SSL/TLS
plain_login – If True, username & password are encoded by utf-8 and PLAIN authentication is used

setup_smtp_basic_auth(server: str, username: str, password: str, port: int | None = None, starttls: bool = True, sender_name: str | None = None, email_address: str | None = None, oauth2: bool = False, no_verify: bool = False, unencrypted: bool = False) → EmailClient

Setups the SMTP connection with a basic username-password authorization for future use.

Parameters:

server – server IP or hostname
username – username to this server
password – password to this server
port – port for SMTP service, if None, default port will be used - 587(TLS), 465(SSL), 25(Unencrypted)
starttls – set True if connection should be use starttls
sender_name – set name of the sender. Is shown on sent emails
email_address – Send emails as this email address. Set this if you need to send emails from a shared or alias email address you have access to via username.
oauth2 – If True, ‘password’ argument is used as access_token for oauth2 authentication
no_verify – If True, SSL certificate will not be verified
unencrypted – If True, connection will be unencrypted, with no SSL/TLS

setup_imap_outlook_oauth2_public(username: str, client_id: str, authority_url: str, shared_username: str | None = None) → EmailClient

Setups the Graph-API connection (functionality as IMAP protocol) with an OAuth2.0 authorization.

Note

This authorization type requires manual action, during the set-up process, to allow the app to access user’s data. By default, Mail.ReadWrite scope/permission is necessary.

Parameters:

username – Username to outlook server
client_id – Application (client) id of the registered App in Azure Portal
authority_url – OAuth 2.0 authorization endpoint, it contains Directory (tenant) ID
shared_username – Specify this argument if you want to access a shared mailbox of different user

setup_imap_outlook_oauth2_confidential(username: str, client_id: str, authority_url: str, secret: str) → EmailClient

Setups the Graph-API connection (functionality as IMAP protocol) with an OAuth2.0 authorization.

Note

This authorization type doesn’t require manual action, during the set-up process, but the permissions require an Admin consent in Azure Portal. By default, Mail.ReadWrite scope/permission is necessary.

Parameters:

username – Username to outlook server
client_id – Application (client) id of the registered App in Azure Portal
authority_url – OAuth 2.0 authorization endpoint, it contains Directory (tenant) ID
secret – A secret string that the application uses to prove its identity when requesting a token. Also, can be referred to as application password

setup_smtp_outlook_oauth2_public(username: str, client_id: str, authority_url: str, sender_name: str | None = None, email_address: str | None = None) → EmailClient

Setups the Graph-API connection (functionality as SMTP protocol) with an OAuth2.0 authorization.

Note

This authorization type requires manual action, during the set-up process, to allow the app to access user’s data. By default, Mail.Send scope/permission is necessary.

Parameters:

username – Username to outlook server
client_id – Application (client) id of the registered App in Azure Portal
authority_url – OAuth 2.0 authorization endpoint, it contains Directory (tenant) ID
sender_name – set name of the sender. Is shown on sent emails
email_address – Send emails as this email address. Set this if you need to send emails from a shared or alias email address you have access to via username.

setup_smtp_outlook_oauth2_confidential(username: str, client_id: str, authority_url: str, secret: str, sender_name: str | None = None, email_address: str | None = None) → EmailClient

Setups the Graph-API connection (functionality as SMTP protocol) with an OAuth2.0 authorization.

Note

This authorization type doesn’t require manual action, during the set-up process, but the permissions require an Admin consent in Azure Portal. By default, Mail.Send scope/permission is necessary.

Parameters:

username – Username to outlook server
client_id – Application (client) id of the registered App in Azure Portal
authority_url – OAuth 2.0 authorization endpoint, it contains Directory (tenant) ID
secret – A secret string that the application uses to prove its identity when requesting a token. Also, can be referred to as application password
sender_name – set name of the sender. Is shown on sent emails
email_address – Send emails as this email address. Set this if you need to send emails from a shared or alias email address you have access to via username.

setup_imap_exchangelib(server: str, username: str, password: str, email_address: str, service_endpoint: bool = False, oauth2: BaseOAuth2Credentials | None = None, no_verify: bool = False) → EmailClient

Setups the IMAP connection with Exchangelib library.

Parameters:

server – Server address
username – Username to server
password – Password to server
email_address – Email address of the user
service_endpoint – If True, server is a service endpoint, otherwise it’s a hostname
oauth2 – OAuth2 exchangelib credentials
no_verify – If True, disables SSL verification

setup_smtp_exchangelib(server: str, username: str, password: str, email_address: str, service_endpoint: bool = False, sender_name: str | None = None, oauth2: BaseOAuth2Credentials | None = None, no_verify: bool = False) → EmailClient

Setups the SMTP connection with Exchangelib library.

Parameters:

server – Server address
username – Username to server
password – Password to server
email_address – Email address of the user
service_endpoint – If True, server is a service endpoint, otherwise it’s a hostname
sender_name – set name of the sender. Is shown on sent emails
oauth2 – OAuth2 exchangelib credentials
no_verify – If True, disables SSL verification

property root: IMAPPath: Root folder on server

property inbox: IMAPPath: Inbox folder on server

list_folders_raw() → List[IMAPFolder]

Lists folders on the IMAP mailbox

Returns:: List of folders and paths

get_current_folder() → str

Get current folder

Returns:: Current folder path

set_current_folder(path: EMAIL_PATH_TYPE) → None

Sets current folder

Parameters:: path – Sets this path as current folder path

folder_exists(path: EMAIL_PATH_TYPE) → bool

Checks if this folder exists on server

Parameters:: path – Folder path to check
Returns:: True if folder exists, False otherwise.

create_folder(path: EMAIL_PATH_TYPE) → None

Create folder on server

Parameters:: path – Folder path to create

delete_folder(path: EMAIL_PATH_TYPE) → None

Delete folder on server

Parameters:: path – Folder path to delete

all(folder: EMAIL_PATH_TYPE | None = None, limit: int | None = None) → Iterator[IMAPMessage]

Returns all emails

Parameters:

folder – Folder from which to extract message. If not set, currently set folder is used
limit – Limits the number of emails which to get. Default None = No Limit

Returns:

Emails in supplied folder

Example:

>>> from aiviro.modules.email import EmailClient
>>> from datetime import datetime
>>> client = EmailClient()
>>> # To receive emails you need to add imap server.
>>> client.setup_imap_basic_auth("<IMAP_SERVER>", "<EMAIL_ADDRESS>", "<EMAIL_PASSWORD>")

>>> # Go through all messages
>>> for email in client.all():
...     print(email.sender.full, email.subject, email.datetime)
...     email.set_seen(True)

Returns all emails following set constraints

Parameters:

seen – True = seen emails, False = unseen emails, None = all emails. Default None
subject – Emails containing this subject. Default None
sender – Emails from this sender. Default None
recipient – Emails to this recipient. Default None
text – Emails containing this text. Default None
cc – CC field containing this recipient. Default None
bcc – BCC field containing this recipient. Default None
date – Email received on this date. Default None
date_from – Emails received after this date. Default None
date_to – Emails received before this date. Default None
folder – Folder which to search. Default is Inbox
limit – Limits the number of emails which to get. Default None = No Limit

Returns:

Emails in supplied folder

Example:

>>> from aiviro.modules.email import EmailClient
>>> from datetime import datetime
>>> client = EmailClient()
>>> # To receive emails you need to add imap server.
>>> client.setup_imap_basic_auth("<IMAP_SERVER>", "<EMAIL_ADDRESS>", "<EMAIL_PASSWORD>")

>>> # Go through all unseen messages
>>> for email in client.query(seen=False):
...     print(email.sender.full, email.subject, email.datetime)
...     email.set_seen(True)

execute_query(query: IMAPQuery) → Iterator[IMAPMessage]

Executes the query.

Parameters:: query – Query to execute

copy_mail(ref: EMAIL_ID_TYPE, path: EMAIL_PATH_TYPE, working_folder: EMAIL_PATH_TYPE | None = None) → None

Copies email to a new folder.

Parameters:

ref – UUID of the email
path – Path to new location of the email
working_folder – Folder which contains email with specified UUID

move_mail(ref: EMAIL_ID_TYPE, path: EMAIL_PATH_TYPE, working_folder: EMAIL_PATH_TYPE | None = None) → None

Moves email to a new folder.

Parameters:

ref – UUID of the email
path – Path to new location of the email
working_folder – Folder which contains email with specified UUID

mail_set_seen(ref: EMAIL_ID_TYPE, seen: bool, working_folder: EMAIL_PATH_TYPE | None = None) → None

Sets the email as seen or unseen.

Parameters:

ref – UUID of the email
seen – Set seen flag to either True or False
working_folder – Folder which contains email with specified UUID

mail_delete(ref: EMAIL_ID_TYPE, working_folder: EMAIL_PATH_TYPE | None = None) → None

Delete the mail from server.

Parameters:

ref – UUID of the email
working_folder – Folder which contains email with specified UUID

Sends email via SMTP server

Parameters:

recipients – One or more recipients in format email or (fullname, email)
subject – Subject of the message
message – Content of the message
attachments – List of paths to attach to the email.
cc – Carbon copy recipients, same format as recipients.
bcc – Blind carbon copy recipients, same format as recipients.
reply_msg – If this message is a reply to email use this attribute to include the reference.
sender_email – Send emails as this email address. Set this if you need to send emails from a shared or alias email address.
message_as_html – If True, content of the ‘messsage’ argument is sent as html

Example:

>>> from aiviro.modules.email import EmailClient
>>> from datetime import datetime
>>> client = EmailClient()
>>> # To send messages you need to connect to SMTP.
>>> client.setup_smtp_basic_auth("<SMTP_SERVER>", "<EMAIL_ADDRESS>", "<EMAIL_PASSWORD>")

>>> # Basic sending of message
>>> client.send_mail(
...     "recipient@foo.bar",
...     "Hello from Python",
...     "This message was sent by Aiviro",
... )

class aiviro.modules.email.IMAPMessage(_id: Optional[str], subject: str, recipients: List[aiviro.modules.email.client.base.Address], sender: aiviro.modules.email.client.base.Address, cc: List[aiviro.modules.email.client.base.Address], bcc: List[aiviro.modules.email.client.base.Address], datetime: datetime.datetime, html: str, body: str, file_attachments: List[~_A] = <factory>, _connector: Optional[ForwardRef('EmailClient')] = None)

property text: str: Returns message’s body

as_file_attachment(attachment_name: str) → IMAPAttachment

Converts message into attachment object, therefore e-mail can be saved or sent as an attachment to another: email.

Parameters:: attachment_name – Name of the attachment
Example:

>>> from aiviro.modules.email import EmailClient
>>> client = EmailClient()
>>> # set-up IMAP & SMTP
>>> attachment_to_send = None
>>> for msg in client.all(limit=1):
...     # get first email from INBOX & convert it into an attachment object
...     attachment_to_send = msg.as_file_attachment(
...         attachment_name=msg.subject.replace(" ", "_")
...     )
...
>>> # send email with an email-attachment
>>> client.send_mail(
...     "recipient@gmail.com",
...     "Subject",
...     "See email in the attachment",
...     [attachment_to_send]
... )

move(path: EMAIL_PATH_TYPE) → None

Moves email to a new folder

Parameters:: path – Path to new location of the email.

copy(path: EMAIL_PATH_TYPE) → None

Copies email to a new folder

Parameters:: path – Path to new location of the email.

set_seen(seen_value: bool) → None

Sets the email as seen or unseen

Parameters:: seen_value – Set seen flag to either True or False

delete() → None: Delete the mail from server

Sends new email as a reply

Parameters:

message – Content of the message
subject – Subject of the message
to_all – Send email to all recipients
attachments – List of paths to attach to the email.
cc – Carbon copy recipients, same format as recipients.
bcc – Blind carbon copy recipients, same format as recipients.

forward(recipients: ADDR_TYPE, message: str = '', subject: str | None = None) → None

Sends new email as a forward

Parameters:

recipients – One or more recipients in format email or (fullname, email)
message – Content of the message
subject – Subject of the message

class aiviro.modules.email.IMAPAttachment(filename: str, content_type: str, content_id: str, content_disposition: str, _size: int, _payload: bytes, is_inline: bool)

as_imap_message() → IMAPMessage: Converts attachment into a message object, only works for message/rfc822 content type.

property content: bytes: Returns attachment’s content

Extractor

class aiviro.modules.email.extractor.EmailExtractor(client: EmailClient, source_dir: EMAIL_DIR_TYPE, processed_dir: EMAIL_DIR_TYPE, unprocessed_dir: EMAIL_DIR_TYPE | None = None, attachment_conditions: List[BaseCondition] | None = None, max_valid_emails: int = 0)

Email extractor provides a standardized way to extract emails and their attachments, based on the provided configuration.

Parameters:

client – EmailClient object with configured IMAP server credentials
source_dir – Email directory from which emails are extracted
processed_dir – Email directory into which valid or processed emails can be moved, see EmailExtractor.move_to_processed()
unprocessed_dir – Email directory into which invalid emails are automatically moved, if set to None, emails are not automatically moved, see EmailExtractor.move_to_unprocessed()
attachment_conditions – Conditions for separating valid/invalid emails, based on their attachments
max_valid_emails – Maximum number of valid emails to extract, if set to 0, all valid emails are extracted

Example:

>>> from aiviro.modules.email import EmailClient
>>> from aiviro.modules.email.extractor import (
...     EmailExtractor,
...     HasAttachmentsCondition,
...     FileExtensionCondition
... )
>>> client = EmailClient()
>>> client.setup_imap_basic_auth("<SMTP_SERVER>", "<EMAIL_ADDRESS>", "<EMAIL_PASSWORD>")
>>> e_extractor = EmailExtractor(
...     client,
...     "INBOX",
...     "INBOX/Processed",
...     "INBOX/Manual",
...     [HasAttachmentsCondition(), FileExtensionCondition("pdf")],
...     15
... )
>>> for email, attachments in e_extractor.extract_all():
...     # process email & attachments

>>> # extract exactly one valid email
>>> email, attachments = e_extractor.extract_one()

move_to_processed(email: IMAPMessage) → None

Moves email into processed email directory, see processed_dir argument in EmailExtractor.

Parameters:: email – Email to move

move_to_unprocessed(email: IMAPMessage) → None

Moves email into unprocessed email directory, see unprocessed_dir argument in EmailExtractor.

Parameters:: email – Email to move

property email_query: IMAPQuery | None

The custom query for extracting emails, set it if you want emails from a specific recipient, with certain subjects, or others. See query() for possible options.

Warning

By setting this option, you override source_dir argument from EmailExtractor

Example:

>>> from aiviro.modules.email import EmailClient
>>> from aiviro.modules.email.extractor import EmailExtractor
>>> client = EmailClient()
>>> client.setup_imap_basic_auth("<SMTP_SERVER>", "<EMAIL_ADDRESS>", "<EMAIL_PASSWORD>")
>>> e_extractor = EmailExtractor(
...     client,
...     "INBOX",
...     "INBOX/Processed",
...     "INBOX/Manual",
... )
>>> e_extractor.email_query = client.query(seen=False)
>>> for email, attachments in e_extractor.extract_all():
...     # process email & attachments

extract_all(auto_move: bool = True) → Generator[EXTRACT_TYPE, None, None]

Extracts all valid emails, based on provided configuration.

Parameters:: auto_move – if True, it automatically moves invalid emails into unprocessed folder

extract_one(auto_move: bool = True) → EXTRACT_TYPE

Extracts one valid email, based on provided configuration.

Parameters:: auto_move – if True, it automatically moves invalid emails into unprocessed folder
Raises:: NoEmailToExtract – If no valid email can be extracted

class aiviro.modules.email.extractor.BaseCallbackStrategy

Base class to inherit from for creating custom callback-strategy. Callback strategies can be added to attachment-conditions, they are called in case the condition is not satisfied. The callback logic is implemented in the __call__ method.

Example:

>>> from aiviro.modules.email.extractor import (
...     EmailExtractor,
...     HasAttachmentsCondition,
...     InvalidEmailNotificationCallback
... )
>>> e_extractor = EmailExtractor(
...     "<client-object>",
...     "INBOX",
...     "INBOX/Processed",
...     "INBOX/Manual",
...     [
...         FileExtensionCondition(
...             "pdf",
...             InvalidEmailNotificationCallback("<email-notifier>", "do not contain PDF attachment")
...         )
...     ],
... )

class aiviro.modules.email.extractor.InvalidEmailNotificationCallback(email_notifier: EmailNotifier, message: str)

Strategy creates unable_to_process() report-message, containing information about invalid email and custom message.

Parameters:

email_notifier – Standardized email-notifier, see EmailNotifier
message – Custom text added at the end of the report-message

class aiviro.modules.email.extractor.BaseCondition(callback: BaseCallbackStrategy | None)

Base class provides an interface to create a custom attachment validator. The validation logic must be implemented in the __call__ method.

Parameters:: callback – Callback that is called, in case the attachments are not valid, see BaseCallbackStrategy, InvalidEmailNotificationCallback

class aiviro.modules.email.extractor.OrCondition(*conditions: BaseCondition, callback: BaseCallbackStrategy | None = None)

Validates if at least one input condition is satisfied.

Parameters:: conditions – Conditions to check

class aiviro.modules.email.extractor.HasAttachmentsCondition(do_not_contain: bool = False, callback: BaseCallbackStrategy | None = None)

Validates if there’s at least one attachment, or none, based on the do_not_contain argument.

Parameters:: do_not_contain – Reverse the logic of the condition

class aiviro.modules.email.extractor.FileExtensionCondition(file_extension: str, callback: BaseCallbackStrategy | None = None)

Validates if at least one attachment has specified file-extension.

Parameters:: file_extension – Extension to check

class aiviro.modules.email.extractor.FilenameCondition(regex: str, callback: BaseCallbackStrategy | None = None)

Validates if at least one attachment’s filename satisfies provided regex.

Parameters:: regex – Regex used for filename validation

class aiviro.modules.email.extractor.ReplaceFilenameCondition(from_char: List[str], to_char: List[str])

Renames attachments filename.

Parameters:

from_char – List of characters to replace
to_char – List of characters to use for the replacement

class aiviro.modules.email.extractor.EmailSubjectCondition(regex: str, callback: BaseCallbackStrategy | None = None)

Validates if email’s subject satisfies provided regex.

Parameters:: regex – Regex used for validation

class aiviro.modules.email.extractor.EmailDatetimeCondition(delta_time: timedelta, timezone: str | None = None, callback: BaseCallbackStrategy | None = None)

Validates if email is younger than specified datetime.

Parameters:

delta_time – The amount of time for which email is still valid
timezone – Pytz timezone for calculating current time, if not set timezone from aiviro-configuration is used

exception aiviro.modules.email.extractor.NoEmailToExtract