No project description provided
Project description
Keble Email
A Python package providing asynchronous email handling capabilities, including sending emails via SMTP and retrieving emails via IMAP.
Features
- Async SMTP: Send emails asynchronously with support for HTML and MJML templates
- Async IMAP: Retrieve emails asynchronously with filtering by date
- Regular IMAP: Non-async IMAP client for synchronous operations
- Template Support: Use HTML or MJML templates with Jinja2 rendering
- Attachments: Easily add single or multiple attachments to emails
- Email Parsing: Parse email bodies and HTML content
- Environment Variables: Configure via environment variables for easy deployment
Installation
pip install keble-email
Or via Poetry:
poetry add keble-email
Configuration
Create a settings class that implements the EmailSettingABC interface:
from keble_email import EmailSettingABC
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(EmailSettingABC, BaseSettings):
model_config = SettingsConfigDict(
env_file=".env", env_file_encoding="utf-8", case_sensitive=True
)
SMTP_SENDER_EMAIL: str
SMTP_SENDER_NAME: str
SMTP_USER: str
SMTP_PASSWORD: str
SMTP_HOST: str
SMTP_PORT: int
SMTP_TLS: bool
SMTP_SSL: bool
TEST_RECIPIENT_EMAIL: str = "test@example.com" # For testing purposes
@property
def sender_email(self) -> str:
return self.SMTP_SENDER_EMAIL
@property
def sender_name(self) -> str:
return self.SMTP_SENDER_NAME
@property
def smtp_user(self) -> str:
return self.SMTP_USER
@property
def smtp_password(self) -> str:
return self.SMTP_PASSWORD
@property
def smtp_host(self) -> str:
return self.SMTP_HOST
@property
def smtp_port(self) -> int:
return self.SMTP_PORT
@property
def smtp_tls(self) -> bool:
return self.SMTP_TLS
@property
def smtp_ssl(self) -> bool:
return self.SMTP_SSL
# Create instance
settings = Settings()
Example .env file:
SMTP_TLS=False
SMTP_SSL=True
SMTP_PORT=465
SMTP_HOST=smtp.example.com
SMTP_SENDER_EMAIL=admin@example.com
SMTP_SENDER_NAME=Example Admin
SMTP_USER=your_username
SMTP_PASSWORD=your_password
TEST_RECIPIENT_EMAIL=test@example.com
# Optional IMAP settings
IMAP_HOST=imap.example.com
IMAP_PORT=993
Usage Examples
Sending Emails with AsyncEmailSender
Using HTML Content Directly
import asyncio
from keble_email import AsyncEmailSender
async def send_email_example():
html_content = """
<!DOCTYPE html>
<html>
<head>
<title>Test Email</title>
</head>
<body>
<h1>Hello World</h1>
<p>This is a test email sent from Keble Email.</p>
</body>
</html>
"""
result = await AsyncEmailSender.asend(
subject="Test Email",
recipient_email="recipient@example.com",
html=html_content,
settings=settings, # Your settings instance
)
print(f"Email sent successfully: {result['success']}")
print(f"Status code: {result['status_code']}")
if __name__ == "__main__":
asyncio.run(send_email_example())
Using HTML Templates with Jinja2
import asyncio
from pathlib import Path
from keble_email import AsyncEmailSender
async def send_template_email():
# Path to your HTML template
template_path = Path("templates/welcome.html")
# Template variables
environment = {
"name": "John Doe",
"verification_url": "https://example.com/verify?token=123456",
"company_name": "Example Company"
}
result = await AsyncEmailSender.asend(
subject="Welcome to Our Service",
recipient_email="user@example.com",
html_template=template_path,
environment=environment,
settings=settings,
cc="support@example.com",
)
print(f"Email sent successfully: {result['success']}")
if __name__ == "__main__":
asyncio.run(send_template_email())
Using MJML Templates
import asyncio
from pathlib import Path
from keble_email import AsyncEmailSender
async def send_mjml_email():
# Path to your MJML template
template_path = Path("templates/newsletter.mjml")
# Template variables
environment = {
"user_name": "Jane Smith",
"articles": [
{"title": "New Feature Announcement", "url": "https://example.com/blog/new-feature"},
{"title": "Tips and Tricks", "url": "https://example.com/blog/tips"}
]
}
result = await AsyncEmailSender.asend(
subject="This Month's Newsletter",
recipient_email="subscriber@example.com",
mjml_template=template_path,
environment=environment,
settings=settings,
)
print(f"Email sent successfully: {result['success']}")
if __name__ == "__main__":
asyncio.run(send_mjml_email())
Sending Emails with Attachments
import asyncio
from pathlib import Path
from keble_email import AsyncEmailSender
async def send_email_with_attachments():
html_content = "<html><body><h1>Please find attached files</h1></body></html>"
# Single attachment
pdf_path = Path("documents/report.pdf")
# Multiple attachments
attachment_paths = [
Path("documents/data.xlsx"),
Path("documents/image.jpg")
]
result = await AsyncEmailSender.asend(
subject="Monthly Report",
recipient_email="manager@example.com",
html=html_content,
settings=settings,
attachment_path=pdf_path, # Main attachment
attachment_paths=attachment_paths, # Additional attachments
)
print(f"Email with attachments sent: {result['success']}")
if __name__ == "__main__":
asyncio.run(send_email_with_attachments())
Retrieving Emails with AsyncImap
Connect and List Folders
import asyncio
from keble_email import AsyncImap
async def list_folders_example():
# Initialize IMAP client
imap = AsyncImap(
host="imap.example.com", # Your IMAP server
port=993, # Usually 993 for IMAPS
user="your_username",
password="your_password",
)
# Connect to server
await imap.aconnect()
# List all folders
folders = await imap.alist_folders()
print("Available folders:")
for folder in folders:
print(f"- {folder}")
# Logout
if imap.client:
await imap.client.logout()
if __name__ == "__main__":
asyncio.run(list_folders_example())
Fetch Emails from a Specific Folder with Date Filtering
import asyncio
from datetime import date, timedelta
from keble_email import AsyncImap
async def fetch_recent_emails():
# Initialize IMAP client
imap = AsyncImap(
host="imap.example.com",
port=993,
user="your_username",
password="your_password",
)
# Connect
await imap.aconnect()
# Get emails from INBOX for the last 7 days
week_ago = date.today() - timedelta(days=7)
today = date.today()
msgs_in_folder = await imap.afetch_folder(
folder="INBOX",
d=[week_ago, today] # Date range
)
print(f"Found {msgs_in_folder.get_total_msgs()} messages in INBOX")
# Process emails
for i, msg_bytes in enumerate(msgs_in_folder.msgs):
print(f"Email {i+1} size: {len(msg_bytes)} bytes")
# Here you would typically parse the email using a library like email.parser
# Logout
if imap.client:
await imap.client.logout()
if __name__ == "__main__":
asyncio.run(fetch_recent_emails())
Get Emails from All Folders
import asyncio
from datetime import date
from keble_email import AsyncImap
async def get_all_emails_from_today():
# Initialize IMAP client
imap = AsyncImap(
host="imap.example.com",
port=993,
user="your_username",
password="your_password",
)
# Get emails from all folders from today
today = date.today()
all_msgs = await imap.aget_emails(d=today)
print(f"Total folders: {len(all_msgs)}")
print(f"Total messages: {sum(m.get_total_msgs() for m in all_msgs)}")
# Process each folder
for folder_msgs in all_msgs:
folder_name = folder_msgs.folder
msg_count = folder_msgs.get_total_msgs()
print(f"Folder '{folder_name}' has {msg_count} messages")
if __name__ == "__main__":
asyncio.run(get_all_emails_from_today())
Using the Synchronous IMAP Client
from datetime import date, timedelta
from keble_email import Imap
def get_emails_example():
# Create IMAP config
imap_config = Imap(
imap_host="imap.example.com",
imap_port=993,
imap_user="your_username",
imap_password="your_password",
)
# Get emails from the last 7 days
week_ago = date.today() - timedelta(days=7)
today = date.today()
# Get emails within date range
email_folders = Imap.get_emails_by_dates(
imap=imap_config,
d=[week_ago, today]
)
# Process results
total_folders = len(email_folders)
total_emails = sum(folder.get_total_msgs() for folder in email_folders)
print(f"Found {total_emails} emails in {total_folders} folders")
# Process each email
for folder in email_folders:
print(f"Folder: {folder.folder} - {folder.get_total_msgs()} emails")
for msg in folder.msgs:
print(f"Subject: {msg.subject}")
print(f"From: {msg.from_}")
print(f"Date: {msg.date}")
print("-" * 50)
if __name__ == "__main__":
get_emails_example()
Parsing Email Content
from keble_email import parse_email_body, HTMLStripperParser
# Parse email body to separate current email from past forwarded/replied content
def parse_email_example():
email_text = """
Hello,
This is my response to your inquiry.
Best regards,
John
> On Monday, June 1, 2023, Jane <jane@example.com> wrote:
> Hello John,
> I have a question about your product.
> Regards,
> Jane
"""
current_email, past_emails = parse_email_body(email_text)
print("Current email content:")
print(current_email)
print("\nPast email content:")
for line in past_emails:
print(line)
# Strip HTML from an email body
html_content = "<html><body><h1>Hello</h1><p>This is a test</p></body></html>"
parser = HTMLStripperParser()
parser.feed(html_content)
plain_text = parser.get_content()
print("\nHTML stripped content:")
print(plain_text)
if __name__ == "__main__":
parse_email_example()
Testing
Run tests using pytest:
pip install pytest pytest-asyncio
pytest -xvs tests/
Make sure to set up a proper .env file with test credentials before running the tests.
License
MIT
Project details
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file keble_email-2.0.0.tar.gz.
File metadata
- Download URL: keble_email-2.0.0.tar.gz
- Upload date:
- Size: 14.1 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: poetry/1.8.2 CPython/3.12.2 Darwin/25.1.0
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
a52c427307ca043f85aeb3b0896cff7a7d143770fa2ab316075d84d11e2cefe5
|
|
| MD5 |
ffbff3cbabcf83696a5ba615bf46c301
|
|
| BLAKE2b-256 |
f5cfea58978677f29b43f4c54b2f0da10261503554bcadf9ad7f74f12294c769
|
File details
Details for the file keble_email-2.0.0-py3-none-any.whl.
File metadata
- Download URL: keble_email-2.0.0-py3-none-any.whl
- Upload date:
- Size: 13.2 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: poetry/1.8.2 CPython/3.12.2 Darwin/25.1.0
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
6cc4de6ecc728397e6fba668dc1d98678297aedf0c7ee9cbfe9bb76f39c8da78
|
|
| MD5 |
86b77d2402bdee51972621a3b9b73c6c
|
|
| BLAKE2b-256 |
10a6928de3b8ea94aee746b4f4f979e596d82c39425356269d4f04d1ab6c1153
|