DEV Community: Fernando Celmer

A Pythonic Way to Handle Emails (IMAP/SMTP) with Auto-Discovery and AI-Ready Design

Fernando Celmer — Fri, 17 Apr 2026 18:22:35 +0000

The Problem

If you've ever tried to read or send emails with Python, you know the pain:

# The old way
import imaplib
import email

mail = imaplib.IMAP4_SSL("imap.gmail.com")
mail.login("user@gmail.com", "password")
mail.select("inbox")
_, data = mail.search(None, "UNSEEN")
for num in data[0].split():
    _, msg_data = mail.fetch(num, "(RFC822)")
    msg = email.message_from_bytes(msg_data[0][1])
    # now manually decode headers, parse body, handle attachments...

Low-level, verbose, and error-prone. You spend more time fighting the protocol than solving your actual problem.

And this is just for reading. If you want to send an HTML email with attachments, you're deep into MIMEMultipart, MIMEText, MIMEBase, encoders.encode_base64... pages of boilerplate before you even get to your actual logic.

Python's standard library gives you the building blocks, but it never gives you the house.

Why I Built This

I was working on a project that needed to automate email workflows — read incoming messages, extract attachments, reply programmatically, and back everything up. After writing the same IMAP boilerplate for the third time across different projects, I decided to build something better.

The goals were simple:

Zero configuration for common providers — just pass an email address and password
A Django ORM-like query API — composable, readable, Pythonic
Lazy evaluation — don't fetch 10,000 emails when you only need a count
Pydantic models everywhere — type-safe, serializable, predictable
One library for everything — read, send, search, reply, forward, backup

The result is email-profile.

Meet email-profile

pip install email-profile

from email_profile import Email

with Email("user@gmail.com", "app-password") as app:
    for msg in app.unread().messages():
        print(f"{msg.from_}: {msg.subject}")

That's it. 3 lines to read your unread emails. No server configuration, no manual connection handling, no protocol-level code.

Key Features

Auto-Discovery for 35+ Providers

No more googling "what's the IMAP server for Outlook?". Just pass your email address:

# Gmail, Outlook, Yahoo, iCloud, ProtonMail, Zoho...
# Server settings are auto-discovered
app = Email("user@outlook.com", "password")

It uses a 4-tier resolution strategy:

Hard-coded provider map — Gmail, Outlook, Yahoo, iCloud, and 30+ others are mapped directly
RFC 6186 SRV DNS lookups — standard service discovery for domains that support it
MX record inference — extracts the provider from MX records and maps to known IMAP hosts
Convention fallback — tries imap.<domain> and smtp.<domain> as a last resort

This means even custom domains hosted on known providers (like a company email on Google Workspace) are resolved automatically.

Powerful Search with Q & Query

If you've used Django's ORM, this will feel familiar. The Q class provides static methods that return composable expressions:

from email_profile import Q
from datetime import date

# Combine with & (AND), | (OR), ~ (NOT)
q = Q.from_("boss@company.com") & Q.subject("urgent") & Q.since(date(2025, 1, 1))

for msg in app.inbox.where(q).messages():
    print(msg.subject)

Available Q methods include Q.from_(), Q.subject(), Q.unseen(), Q.seen(), Q.since(), Q.before(), Q.larger(), and more. They can be combined freely with &, |, and ~ operators.

For simpler cases, use kwargs-style queries with Query:

from email_profile import Query

results = app.inbox.where(Query(subject="invoice", unseen=True))
print(f"Found {results.count()} unread invoices")

Query also supports chaining with .exclude() and .or_():

q = Query(subject="report").exclude(from_who="spam@example.com").or_(subject="invoice")
messages = app.inbox.where(q).list()

Send Emails (Plain Text, HTML, Attachments)

Sending is just as clean. Plain text, HTML, attachments, CC, BCC — all in one call:

app.send(
    to="team@company.com",
    subject="Weekly Report",
    html="<h1>Report</h1><p>All systems operational.</p>",
    attachments=["report.pdf", "metrics.csv"]
)

You can also send both plain text and HTML as a multipart message:

app.send(
    to="recipient@example.com",
    subject="Update",
    body="Plain text fallback for clients that don't render HTML.",
    html="<h1>Update</h1><p>Rich version of the same content.</p>"
)

Note that html is a string parameter (the HTML content itself), not a boolean flag. This is a deliberate design choice — it keeps the API explicit and avoids ambiguity.

Reply & Forward

Reply and forward are methods on the Email object, not on the message. This keeps Message as a pure data object (Pydantic model) with no side effects:

# Reply to sender only
app.reply(msg, body="Thanks, I'll review it today!")

# Reply all
app.reply(msg, body="Acknowledged.", reply_all=True)

# Forward
app.forward(msg, to="colleague@company.com", body="FYI - please take a look")

Mailbox Management

Access standard folders via properties, or reach any custom folder by path:

# Built-in shortcuts
app.inbox
app.sent
app.spam
app.trash
app.drafts
app.archive

# Custom folders
reports = app.mailbox("Projects/Reports")

Flag operations live on the mailbox, not the message. They accept a Message, a UID string, or an integer:

app.inbox.mark_seen(msg)
app.inbox.mark_unseen(msg)
app.inbox.flag(msg)
app.inbox.unflag(msg)
app.inbox.move(msg, "Archive")
app.inbox.copy(msg, "Backup")
app.inbox.delete(msg)

This design means you can operate on messages from any context — even messages loaded from a backup — without needing an active connection reference on the message object itself.

Sync & Backup to SQLite

Back up your entire mailbox to a local SQLite database with incremental sync:

from email_profile import StorageSQLite

app = Email("user@gmail.com", "password")
app.storage = StorageSQLite("backup.db")

with app:
    result = app.sync()  # incremental — only downloads new emails
    print(f"{result.inserted} new, {result.skipped} skipped")

The SyncResult object gives you full visibility: inserted, updated, deleted, skipped, and errors. Restore is just as simple:

with app:
    count = app.restore()  # returns int (count of restored messages)
    print(f"Restored {count} messages")

Lazy Evaluation

Queries return a Where object — nothing hits the server until you explicitly ask for data. This is critical for performance when working with large mailboxes:

query = app.inbox.where(Q.unseen())

# No IMAP call yet — the query is just a description
query.count()       # hits the server, returns int
query.exists()      # bool — cheaper than count for "any unread?" checks
query.first()       # Message or None — fetches only one
query.last()        # Message or None — fetches only one
query.list()        # list[Message] — materializes everything
query.messages()    # iterator — memory-efficient for large result sets

This means app.inbox.where(Q.unseen()).count() only asks the server "how many?", without downloading a single message body.

Fetch Modes

Control how much data you pull per message. This makes a huge difference when you're scanning thousands of emails:

# Full message + attachments (default)
app.inbox.where().messages(mode="full")

# Headers + body, skip attachments (faster)
app.inbox.where().messages(mode="text")

# Headers only (cheapest — great for building an index)
app.inbox.where().messages(mode="headers")

For example, if you're building a dashboard that shows subject lines and dates, use mode="headers" and save 90%+ of bandwidth.

Message as a Pydantic Model

Every message is a Message object — a Pydantic BaseModel with typed fields:

from email_profile import Message

for msg in app.inbox.where(unseen=True).messages():
    msg.subject          # str
    msg.from_            # str
    msg.to_              # str
    msg.date             # datetime
    msg.uid              # str
    msg.message_id       # str
    msg.body_text_plain  # Optional[str]
    msg.body_text_html   # Optional[str]
    msg.cc               # Optional[str]
    msg.bcc              # Optional[str]
    msg.reply_to         # Optional[str]
    msg.in_reply_to      # Optional[str]
    msg.references        # Optional[str]
    msg.content_type     # str
    msg.attachments      # list[Attachment]
    msg.headers          # dict
    msg.list_id          # Optional[str]
    msg.list_unsubscribe # Optional[str]

Because it's Pydantic, you get .model_dump(), .model_dump_json(), validation, and serialization for free. This makes it trivial to pipe email data into APIs, databases, or AI workflows.

Real-World Recipes

Monitor inbox for new messages

with Email.from_env() as app:
    for msg in app.unread().messages():
        print(f"From: {msg.from_} | Subject: {msg.subject}")
        app.inbox.mark_seen(msg)

Download PDF attachments from invoices

with Email("user@gmail.com", "password") as app:
    for msg in app.search("invoice").messages():
        for att in msg.attachments:
            if att.content_type == "application/pdf":
                att.save("./invoices/")

Backup all emails to SQLite

from email_profile import StorageSQLite

app = Email("user@gmail.com", "password")
app.storage = StorageSQLite("full_backup.db")

with app:
    result = app.sync()
    print(f"Backed up {result.inserted} new emails")

Clean up spam folder

with Email.from_env() as app:
    for msg in app.spam.where().messages():
        app.spam.delete(msg)

Search across date ranges

from datetime import date
from email_profile import Q

with Email.from_env() as app:
    q = Q.since(date(2025, 1, 1)) & Q.before(date(2025, 4, 1)) & Q.from_("client@company.com")
    messages = app.inbox.where(q).list()
    print(f"Found {len(messages)} emails from client in Q1 2025")

Use environment variables

# .env
EMAIL_SERVER=imap.gmail.com
EMAIL_USERNAME=user@gmail.com
EMAIL_PASSWORD=app-password

with Email.from_env() as app:
    print(f"Connected as {app.user}")

You can also customize the variable names:

app = Email.from_env(
    server_var="MY_IMAP_HOST",
    user_var="MY_EMAIL",
    password_var="MY_EMAIL_PW"
)

Built-in Error Handling & Retry

The library provides specific exception classes so you can handle failures gracefully:

from email_profile import ConnectionFailure, NotConnected, QuotaExceeded, RateLimited

try:
    app.connect()
except ConnectionFailure:
    print("Could not connect to server")
except QuotaExceeded:
    print("Mailbox quota exceeded")
except RateLimited:
    print("Too many requests — back off and retry")

For automated workflows, use the built-in retry decorator:

from email_profile.retry import with_retry

@with_retry(attempts=3, initial_delay=1.0, max_delay=30.0)
def fetch_unread():
    with Email.from_env() as app:
        return app.unread().list()

It handles exponential backoff automatically, so your scripts survive transient network issues without custom retry logic.

Why email-profile?

Feature	imaplib/smtplib	email-profile
Auto-discovery	No	35+ providers
Context manager	No	Yes
Search API	Raw IMAP commands	Q objects & Query kwargs
Send HTML	Manual MIME building	`html="<h1>..."`
Attachments	Manual encoding	`attachments=["file.pdf"]`
Reply/Forward	Build from scratch	`app.reply(msg, body="...")`
Backup/Sync	DIY	`StorageSQLite` built-in
Lazy evaluation	No	Where objects
Retry logic	DIY	`@with_retry` decorator
Pydantic models	No	Message DTO
Fetch modes	Manual flags	`mode="full/text/headers"`

Design Principles

A few decisions that shape the library:

Message is data, not behavior. Message is a pure Pydantic model. It has no methods that mutate state or require a connection. Operations like mark_seen(), move(), and delete() live on the mailbox. This keeps the model serializable and testable.
Lazy by default. .where() returns a Where object, not a list. You choose when and how to materialize: .count(), .first(), .list(), or .messages(). This prevents accidentally downloading thousands of emails.
One import, one object. For most use cases, from email_profile import Email is all you need. The Email class is the entry point for connecting, reading, sending, replying, forwarding, and syncing.
Fail explicitly. Instead of generic exceptions, the library provides ConnectionFailure, NotConnected, QuotaExceeded, and RateLimited — so you can handle each case differently.

Getting Started

pip install email-profile

from email_profile import Email

with Email("your@email.com", "app-password") as app:
    # Read
    for msg in app.unread().messages():
        print(msg.subject)

    # Send
    app.send(to="friend@email.com", subject="Hello!", body="Sent with email-profile")

How to Create a Pipeline with Dotflow in Python

Fernando Celmer — Tue, 07 Apr 2026 01:47:42 +0000

In this tutorial, you'll learn how to build a complete data pipeline using Dotflow — a lightweight Python library that requires zero infrastructure.

No Redis. No RabbitMQ. No Postgres. No Docker. Just pip install dotflow.

What we'll build

A pipeline that:

Extracts user data from a source
Transforms it by filtering active users and calculating stats
Loads the results into storage

Along the way, we'll add retry with backoff, parallel execution, checkpoint/resume, and cron scheduling.

Step 1 — Install Dotflow

pip install dotflow

Step 2 — Create your first pipeline

Create a file called pipeline.py:

from dotflow import DotFlow, action


@action
def extract():
    """Simulate extracting data from a database or API."""
    return {
        "users": [
            {"name": "Alice", "age": 30, "active": True},
            {"name": "Bob", "age": 25, "active": False},
            {"name": "Charlie", "age": 35, "active": True},
            {"name": "Diana", "age": 28, "active": True},
        ]
    }


@action
def transform(previous_context):
    """Filter active users and calculate stats."""
    users = previous_context.storage["users"]
    active = [u for u in users if u["active"]]
    avg_age = sum(u["age"] for u in active) / len(active)

    return {
        "active_users": active,
        "total": len(active),
        "average_age": round(avg_age, 1),
    }


@action
def load(previous_context):
    """Print the final results."""
    data = previous_context.storage
    print(f"Active users: {data['total']}")
    print(f"Average age: {data['average_age']}")
    for user in data["active_users"]:
        print(f"  - {user['name']} ({user['age']})")
    return data


workflow = DotFlow()
workflow.task.add(step=extract)
workflow.task.add(step=transform)
workflow.task.add(step=load)

workflow.start()

Run it:

python pipeline.py

Output:

Active users: 3
Average age: 31.0
  - Alice (30)
  - Charlie (35)
  - Diana (28)

That's a working pipeline in 40 lines. Let's make it production-ready.

Step 3 — Add retry and timeout

Real data sources fail. Let's make the extract step resilient:

import requests


@action(retry=3, timeout=10, backoff=True)
def extract():
    """Fetch users from an API with automatic retry."""
    response = requests.get("https://api.example.com/users")
    response.raise_for_status()
    return {"users": response.json()}

What this does:

retry=3 — tries up to 3 times before failing
timeout=10 — each attempt has a 10-second limit
backoff=True — waits 1s, then 2s, then 4s between retries

If all retries fail, the task is marked as FAILED with full error tracking:

for task in workflow.result_task():
    if task.errors:
        for error in task.errors:
            print(f"Attempt {error.attempt}: {error.message}")

Step 4 — Run tasks in parallel

If your tasks are independent, run them simultaneously:

@action(retry=3, timeout=10, backoff=True)
def fetch_users():
    response = requests.get("https://api.example.com/users")
    response.raise_for_status()
    return {"users": response.json()}


@action(retry=3, timeout=10, backoff=True)
def fetch_orders():
    response = requests.get("https://api.example.com/orders")
    response.raise_for_status()
    return {"orders": response.json()}


@action(retry=3, timeout=10, backoff=True)
def fetch_products():
    response = requests.get("https://api.example.com/products")
    response.raise_for_status()
    return {"products": response.json()}


workflow = DotFlow()
workflow.task.add(step=fetch_users)
workflow.task.add(step=fetch_orders)
workflow.task.add(step=fetch_products)

workflow.start(mode="parallel")

All three API calls run at the same time. If one fails, the others continue (with keep_going=True):

workflow.start(mode="parallel", keep_going=True)

Using groups

Need parallel between groups but sequential within? Use group_name:

workflow = DotFlow()

# These two run sequentially (same group)
workflow.task.add(step=fetch_users, group_name="users")
workflow.task.add(step=process_users, group_name="users")

# These two run sequentially (same group)
workflow.task.add(step=fetch_orders, group_name="orders")
workflow.task.add(step=process_orders, group_name="orders")

# The two groups run in parallel
workflow.start()

Step 5 — Save results with storage providers

By default, results stay in memory. To persist them:

File storage

from dotflow import DotFlow, Config, action
from dotflow.providers import StorageFile

config = Config(storage=StorageFile(path=".output"))

workflow = DotFlow(config=config)
workflow.task.add(step=extract)
workflow.task.add(step=transform)
workflow.task.add(step=load)

workflow.start()

Results are saved as JSON files in .output/tasks/.

AWS S3

pip install dotflow[aws]

from dotflow.providers import StorageS3

config = Config(storage=StorageS3(bucket="my-bucket", prefix="pipelines/"))

Google Cloud Storage

pip install dotflow[gcp]

from dotflow.providers import StorageGCS

config = Config(storage=StorageGCS(bucket="my-bucket", project="my-project"))

Step 6 — Checkpoint and resume

Long pipeline crashed at step 47? Don't restart from zero:

config = Config(storage=StorageFile())

workflow = DotFlow(config=config, workflow_id="550e8400-e29b-41d4-a716-446655440000")
workflow.task.add(step=step_1)
workflow.task.add(step=step_2)
workflow.task.add(step=step_3)
# ... many steps

# First run — saves checkpoint after each completed task
workflow.start()

# Pipeline crashed at step_3. Fix the bug and resume:
workflow.start(resume=True)  # skips step_1 and step_2 automatically

Two requirements:

A persistent storage provider (file, S3, or GCS)
A fixed workflow_id (so it knows which checkpoints to load)

Step 7 — Schedule with cron

Run your pipeline on a schedule:

pip install dotflow[scheduler]

from dotflow import DotFlow, Config, action
from dotflow.providers import SchedulerCron, StorageFile


@action(retry=3, timeout=30, backoff=True)
def extract():
    return {"users": [...]}


@action
def transform(previous_context):
    return {"processed": True}


@action
def load(previous_context):
    print("Pipeline complete!")
    return previous_context.storage


config = Config(
    storage=StorageFile(path=".output"),
    scheduler=SchedulerCron(cron="0 */6 * * *"),  # every 6 hours
)

workflow = DotFlow(config=config, workflow_id="a1b2c3d4-e5f6-7890-abcd-ef1234567890")
workflow.task.add(step=extract)
workflow.task.add(step=transform)
workflow.task.add(step=load)

workflow.schedule(resume=True)  # runs on cron + resumes from failure

What happens when a new execution triggers while the previous one is still running? Choose an overlap strategy:

# Skip — drop the new run (default)
SchedulerCron(cron="*/5 * * * *", overlap="skip")

# Queue — buffer one pending run
SchedulerCron(cron="*/5 * * * *", overlap="queue")

# Parallel — up to 10 concurrent executions
SchedulerCron(cron="*/5 * * * *", overlap="parallel")

Step 8 — Get notified on failure

Send a Telegram message when a task fails:

from dotflow.providers import NotifyTelegram
from dotflow.core.types.status import TypeStatus

config = Config(
    storage=StorageFile(path=".output"),
    notify=NotifyTelegram(
        token="YOUR_BOT_TOKEN",
        chat_id=123456789,
        notification_type=TypeStatus.FAILED,
    ),
)

Step 9 — Inspect results

After execution, inspect everything:

workflow.start()

# Task objects with status, duration, errors
for task in workflow.result_task():
    print(f"Task {task.task_id}: {task.status} ({task.duration}s)")

# Just the return values
for storage in workflow.result_storage():
    print(storage)

# Serialized result (Pydantic model)
result = workflow.result()

Step 10 — Run from the CLI

No need to write a script. Run directly from the terminal:

# Simple execution
dotflow start --step pipeline.extract

# With execution mode
dotflow start --step pipeline.extract --mode parallel

# With file storage
dotflow start --step pipeline.extract --storage file --path .output

# Scheduled
dotflow schedule --step pipeline.extract --cron "0 */6 * * *" --storage file

Complete example

Here's everything together — a production-ready pipeline in one file:

from dotflow import DotFlow, Config, action
from dotflow.providers import StorageFile, NotifyTelegram
from dotflow.core.types.status import TypeStatus


@action(retry=3, timeout=30, backoff=True)
def extract():
    return {
        "users": [
            {"name": "Alice", "age": 30, "active": True},
            {"name": "Bob", "age": 25, "active": False},
            {"name": "Charlie", "age": 35, "active": True},
        ]
    }


@action
def transform(previous_context):
    users = previous_context.storage["users"]
    active = [u for u in users if u["active"]]
    return {"active_users": active, "count": len(active)}


@action
def load(previous_context):
    data = previous_context.storage
    print(f"Loaded {data['count']} active users")
    return data


config = Config(
    storage=StorageFile(path=".output"),
    notify=NotifyTelegram(
        token="YOUR_BOT_TOKEN",
        chat_id=123456789,
        notification_type=TypeStatus.FAILED,
    ),
)

workflow = DotFlow(config=config, workflow_id="d4e5f6a7-b8c9-0123-4567-89abcdef0123")
workflow.task.add(step=extract)
workflow.task.add(step=transform)
workflow.task.add(step=load)

workflow.start()

# Inspect results
for task in workflow.result_task():
    print(f"Task {task.task_id}: {task.status}")

What's next?

Dotflow is open source (MIT license) and actively maintained. If you found this useful, a star on GitHub helps the project grow.

Questions? Drop them in the comments.