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:
1. Register an application
2. (Optional) In case of using Public(Delegated) Authorization option, you have to enable Allow public client flows in Authentication section
3. Add credentials, we currently support only Client’s Secret
4. 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.

Parameters

  • username – Username for both services

  • password – Password for both services

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: Optional[int] = None, starttls: bool = False, oauth2: bool = False, no_verify: bool = False, unencrypted: bool = False) → aiviro.modules.email.client.email.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

setup_smtp_basic_auth(server: str, username: str, password: str, port: Optional[int] = None, starttls: bool = True, sender_name: Optional[str] = None, email_address: Optional[str] = None, oauth2: bool = False, no_verify: bool = False) → aiviro.modules.email.client.email.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)

  • 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

setup_imap_outlook_oauth2_public(username: str, client_id: str, authority_url: str, shared_username: Optional[str] = None) → aiviro.modules.email.client.email.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) → aiviro.modules.email.client.email.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: Optional[str] = None, email_address: Optional[str] = None) → aiviro.modules.email.client.email.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: Optional[str] = None, email_address: Optional[str] = None) → aiviro.modules.email.client.email.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.

property root: aiviro.modules.email.client.imap.IMAPPath

Root folder on server

property inbox: aiviro.modules.email.client.imap.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: Optional[EMAIL_PATH_TYPE] = None, limit: Optional[int] = 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)
query(seen: Optional[bool] = None, subject: Optional[str] = None, sender: Optional[str] = None, recipient: Optional[str] = None, text: Optional[str] = None, cc: Optional[str] = None, bcc: Optional[str] = None, date: Optional[datetime.date] = None, date_from: Optional[datetime.date] = None, date_to: Optional[datetime.date] = None, folder: Optional[EMAIL_PATH_TYPE] = None, limit: Optional[int] = None) → IMAPQuery

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: Optional[EMAIL_PATH_TYPE] = 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: Optional[EMAIL_PATH_TYPE] = 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: Optional[EMAIL_PATH_TYPE] = 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: Optional[EMAIL_PATH_TYPE] = None) → None

Delete the mail from server.

Parameters

  • ref – UUID of the email

  • working_folder – Folder which contains email with specified UUID

send_mail(recipients: ADDR_TYPE, subject: str, message: str, attachments: Optional[Sequence[Union[str, pathlib.Path, BaseAttachment]]] = None, cc: Optional[ADDR_TYPE] = None, bcc: Optional[ADDR_TYPE] = None, reply_msg: Optional[IMAPMessage] = None, sender_email: Optional[str] = None, message_as_html: bool = False) → None

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) → aiviro.modules.email.client.imap.IMAPAttachment

Converts message into attachment object, therefore e-mail can be 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

reply(message: str, subject: Optional[str] = None, to_all: bool = True, attachments: Optional[List[Union[str, pathlib.Path, aiviro.modules.email.client.base.BaseAttachment]]] = None, cc: Optional[ADDR_TYPE] = None, bcc: Optional[ADDR_TYPE] = None) → None

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: Optional[str] = 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

Extractor

class aiviro.modules.email.extractor.EmailExtractor(client: EmailClient, source_dir: EMAIL_DIR_TYPE, processed_dir: EMAIL_DIR_TYPE, unprocessed_dir: Optional[EMAIL_DIR_TYPE] = None, attachment_conditions: Optional[List[aiviro.modules.email.extractor.conditions.BaseCondition]] = 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: Optional[IMAPQuery]

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: Optional[aiviro.modules.email.extractor.conditions.BaseCallbackStrategy])

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: aiviro.modules.email.extractor.conditions.BaseCondition, callback: Optional[aiviro.modules.email.extractor.conditions.BaseCallbackStrategy] = 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: Optional[aiviro.modules.email.extractor.conditions.BaseCallbackStrategy] = 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: Optional[aiviro.modules.email.extractor.conditions.BaseCallbackStrategy] = 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: Optional[aiviro.modules.email.extractor.conditions.BaseCallbackStrategy] = 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: Optional[aiviro.modules.email.extractor.conditions.BaseCallbackStrategy] = None)

Validates if email’s subject satisfies provided regex.

Parameters

regex – Regex used for validation

class aiviro.modules.email.extractor.EmailDatetimeCondition(delta_time: datetime.timedelta, timezone: Optional[str] = None, callback: Optional[aiviro.modules.email.extractor.conditions.BaseCallbackStrategy] = 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