class O365Client
Maturity: 37
A client class for interacting with Microsoft 365 Graph API to send emails with authentication, validation, and attachment support.
File:
/tf/active/vicechatdev/email-forwarder/src/forwarder/o365_client.py
Lines:
21 - 260
21 - 260
Complexity:
moderate
moderate
Purpose
O365Client provides a comprehensive interface for sending emails through Microsoft 365 using OAuth2 client credentials flow. It handles authentication token management, email address validation and cleaning, attachment preparation, and email sending with support for reply-to addresses and forwarding metadata. The class manages token lifecycle, retries on authentication failures, and provides connection testing capabilities.
Source Code
class O365Client:
def __init__(self):
self.tenant_id = settings.MS365_TENANT_ID
self.client_id = settings.MS365_CLIENT_ID
self.client_secret = settings.MS365_CLIENT_SECRET
self.sender_email = settings.MS365_SENDER_EMAIL
self.token = None
def get_access_token(self) -> Optional[str]:
"""Get access token from Microsoft 365"""
try:
url = f"https://login.microsoftonline.com/{self.tenant_id}/oauth2/v2.0/token"
headers = {
'Content-Type': 'application/x-www-form-urlencoded'
}
data = {
'client_id': self.client_id,
'client_secret': self.client_secret,
'scope': 'https://graph.microsoft.com/.default',
'grant_type': 'client_credentials'
}
response = requests.post(url, headers=headers, data=data, timeout=30)
if response.status_code == 200:
token_data = response.json()
self.token = token_data.get('access_token')
logger.info("Successfully acquired MS365 token")
return self.token
else:
logger.error(f"Failed to get access token. Status: {response.status_code}, Response: {response.text}")
return None
except Exception as e:
logger.error(f"Exception getting access token: {str(e)}")
return None
def validate_email_address(self, email: str) -> bool:
"""Validate email address format"""
if not email or not isinstance(email, str):
return False
# Parse the email address
name, addr = parseaddr(email)
# Check if we have a valid email address
if not addr or '@' not in addr:
return False
# Basic email validation
parts = addr.split('@')
if len(parts) != 2 or not parts[0] or not parts[1]:
return False
return True
def clean_email_address(self, email: str) -> Optional[str]:
"""Clean and extract email address from various formats"""
if not email:
return None
# Handle different email formats
email = email.strip()
# If it's already a clean email, return it
if '@' in email and '<' not in email and '>' not in email:
return email if self.validate_email_address(email) else None
# Parse email with potential name
name, addr = parseaddr(email)
if addr and self.validate_email_address(addr):
return addr
return None
def prepare_attachments(self, attachments: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Prepare attachments for Microsoft Graph API"""
prepared_attachments = []
for attachment in attachments:
try:
# Prepare attachment for Graph API
graph_attachment = {
"@odata.type": "#microsoft.graph.fileAttachment",
"name": attachment['filename'],
"contentType": attachment['content_type'],
"contentBytes": attachment['content']
}
prepared_attachments.append(graph_attachment)
logger.debug(f"Prepared attachment: {attachment['filename']}")
except Exception as e:
logger.error(f"Failed to prepare attachment {attachment.get('filename', 'unknown')}: {str(e)}")
continue
return prepared_attachments
def send_email(self, to_email: str, subject: str, body: str,
from_email: Optional[str] = None,
reply_to: Optional[str] = None,
attachments: Optional[List[Dict[str, Any]]] = None) -> bool:
"""Send email via Microsoft 365 Graph API"""
try:
# Get access token if not available
if not self.token:
if not self.get_access_token():
logger.error("Failed to get access token")
return False
# Clean and validate email addresses
clean_to = self.clean_email_address(to_email)
if not clean_to:
logger.error(f"Invalid to_email address: {to_email}")
return False
# Clean reply-to address if provided
clean_reply_to = None
if reply_to:
clean_reply_to = self.clean_email_address(reply_to)
if not clean_reply_to:
logger.warning(f"Invalid reply_to address, ignoring: {reply_to}")
clean_reply_to = None
# Clean from_email if provided
clean_from = None
if from_email:
clean_from = self.clean_email_address(from_email)
if not clean_from:
logger.warning(f"Invalid from_email address, ignoring: {from_email}")
# Prepare the email message
message = {
"message": {
"subject": subject or "Forwarded Email",
"body": {
"contentType": "HTML",
"content": body or "No content"
},
"toRecipients": [
{
"emailAddress": {
"address": clean_to
}
}
]
}
}
# Add reply-to only if we have a valid address
if clean_reply_to:
message["message"]["replyTo"] = [
{
"emailAddress": {
"address": clean_reply_to
}
}
]
# Add sender information if provided
if clean_from and clean_from != self.sender_email:
# Add as a note in the body since we can't change the sender
original_body = message["message"]["body"]["content"]
message["message"]["body"]["content"] = f"""
<div style="border-left: 3px solid #ccc; padding-left: 10px; margin: 10px 0;">
<strong>Original Sender:</strong> {from_email or clean_from}<br>
<strong>Forwarded by:</strong> Email Forwarder Service
</div>
<hr>
{original_body}
"""
# Add attachments if provided
if attachments:
prepared_attachments = self.prepare_attachments(attachments)
if prepared_attachments:
message["message"]["attachments"] = prepared_attachments
logger.info(f"Added {len(prepared_attachments)} attachments to email")
# Send the email
url = f"https://graph.microsoft.com/v1.0/users/{self.sender_email}/sendMail"
headers = {
'Authorization': f'Bearer {self.token}',
'Content-Type': 'application/json'
}
response = requests.post(url, headers=headers, json=message, timeout=60) # Increased timeout for attachments
if response.status_code == 202:
logger.info(f"Email sent successfully to {clean_to}")
if attachments:
logger.info(f"Successfully sent {len(attachments)} attachments")
return True
else:
logger.error(f"Failed to send email. Status: {response.status_code}, Response: {response.text}")
# Try to get a new token and retry once
if response.status_code == 401:
logger.info("Token might be expired, getting new token")
if self.get_access_token():
headers['Authorization'] = f'Bearer {self.token}'
response = requests.post(url, headers=headers, json=message, timeout=60)
if response.status_code == 202:
logger.info(f"Email sent successfully to {clean_to} (retry)")
return True
return False
except Exception as e:
logger.error(f"Exception sending email: {str(e)}")
return False
def test_connection(self) -> bool:
"""Test the connection to Microsoft 365"""
try:
if self.get_access_token():
# Test by getting user info
url = f"https://graph.microsoft.com/v1.0/users/{self.sender_email}"
headers = {
'Authorization': f'Bearer {self.token}',
'Content-Type': 'application/json'
}
response = requests.get(url, headers=headers, timeout=30)
if response.status_code == 200:
logger.info("MS365 connection test successful")
return True
else:
logger.error(f"MS365 connection test failed. Status: {response.status_code}")
return False
else:
return False
except Exception as e:
logger.error(f"Exception testing MS365 connection: {str(e)}")
return FalseParameters
| Name | Type | Default | Kind |
|---|---|---|---|
bases |
- | - |
Parameter Details
__init__: No parameters required. The constructor initializes the client by reading configuration from settings (tenant_id, client_id, client_secret, sender_email) and sets the token to None. All configuration is pulled from the settings module.
Return Value
Instantiation returns an O365Client object. Key method returns: get_access_token() returns Optional[str] (access token or None), validate_email_address() returns bool, clean_email_address() returns Optional[str] (cleaned email or None), prepare_attachments() returns List[Dict[str, Any]] (prepared attachments), send_email() returns bool (success status), test_connection() returns bool (connection status).
Class Interface
Methods
__init__(self)
Purpose: Initialize the O365Client with configuration from settings and set token to None
Returns: None - initializes instance attributes
get_access_token(self) -> Optional[str]
Purpose: Authenticate with Microsoft 365 using OAuth2 client credentials flow and retrieve an access token
Returns: Access token string if successful, None if authentication fails. Token is also stored in self.token
validate_email_address(self, email: str) -> bool
Purpose: Validate that an email address has correct format with @ symbol and valid parts
Parameters:
email: Email address string to validate
Returns: True if email format is valid, False otherwise
clean_email_address(self, email: str) -> Optional[str]
Purpose: Extract and clean email address from various formats including 'Name <email@domain.com>' format
Parameters:
email: Email address string in any format (with or without display name)
Returns: Cleaned email address string if valid, None if invalid or cannot be parsed
prepare_attachments(self, attachments: List[Dict[str, Any]]) -> List[Dict[str, Any]]
Purpose: Convert attachment dictionaries to Microsoft Graph API format with proper OData types
Parameters:
attachments: List of dictionaries with 'filename', 'content_type', and 'content' (base64) keys
Returns: List of dictionaries formatted for Graph API with @odata.type, name, contentType, and contentBytes
send_email(self, to_email: str, subject: str, body: str, from_email: Optional[str] = None, reply_to: Optional[str] = None, attachments: Optional[List[Dict[str, Any]]] = None) -> bool
Purpose: Send an email via Microsoft Graph API with optional reply-to, from metadata, and attachments
Parameters:
to_email: Recipient email address (required)subject: Email subject line (required)body: HTML email body content (required)from_email: Original sender email (optional, added as metadata in body since sender cannot be changed)reply_to: Reply-to email address (optional)attachments: List of attachment dictionaries with filename, content_type, and base64 content (optional)
Returns: True if email sent successfully (HTTP 202), False on any failure. Automatically retries once on 401 errors
test_connection(self) -> bool
Purpose: Test the connection to Microsoft 365 by authenticating and retrieving user information
Returns: True if connection successful and user info retrieved, False otherwise
Attributes
| Name | Type | Description | Scope |
|---|---|---|---|
tenant_id |
str | Microsoft 365 tenant ID from settings, used for OAuth2 authentication endpoint | instance |
client_id |
str | Azure AD application client ID from settings, used for OAuth2 authentication | instance |
client_secret |
str | Azure AD application client secret from settings, used for OAuth2 authentication | instance |
sender_email |
str | Email address to send from (from settings), must have SendMail permissions in Azure AD | instance |
token |
Optional[str] | Cached OAuth2 access token, initially None, set by get_access_token() and reused for subsequent API calls | instance |
Dependencies
requeststypingemail.utilsloggingconfig
Required Imports
import requests
from typing import Dict, List, Optional, Any
from email.utils import parseaddr, formataddr
import logging
from config import settings
Usage Example
# Initialize the client
client = O365Client()
# Test connection
if client.test_connection():
print("Connected successfully")
# Send a simple email
success = client.send_email(
to_email='recipient@example.com',
subject='Test Email',
body='<p>Hello from O365Client</p>'
)
# Send email with reply-to and attachments
attachments = [
{
'filename': 'document.pdf',
'content_type': 'application/pdf',
'content': 'base64_encoded_content_here'
}
]
success = client.send_email(
to_email='recipient@example.com',
subject='Email with Attachment',
body='<p>Please see attached document</p>',
reply_to='original@sender.com',
attachments=attachments
)
if success:
print("Email sent successfully")
else:
print("Failed to send email")Best Practices
- Always call test_connection() after instantiation to verify credentials are valid
- The token is automatically managed - get_access_token() is called internally by send_email() if needed
- Token is cached in self.token and reused across multiple send_email() calls
- On 401 errors, the class automatically attempts to refresh the token once
- Email addresses are automatically cleaned and validated - invalid addresses are logged and rejected
- Attachments must be provided as dictionaries with 'filename', 'content_type', and 'content' (base64 encoded) keys
- The sender email cannot be changed (enforced by MS365) - from_email is added as metadata in the body
- Use reply_to parameter to set a different reply address than the sender
- The class uses a 60-second timeout for email sending (to accommodate large attachments)
- All operations are logged using the logger - check logs for detailed error information
- The class is stateful (stores token) - reuse the same instance for multiple emails to avoid re-authentication
Tags
Similar Components
AI-powered semantic similarity - components with related functionality:
-
function get_o365_token 73.0% similar
-
function test_o365_connection 70.5% similar
-
function get_ms365_token 65.8% similar
-
function authenticate_o365 65.2% similar
-
function send_email_ms365 60.8% similar