Processing Email Attachments Automatically with Python

For many small teams, email is the integration layer. Suppliers send invoices as PDF attachments to invoices@, a bank emails a daily CSV statement, a partner's system mails an Excel stock report every morning, customers reply with signed forms. Someone opens each message, saves the attachment to a shared folder, renames it, and starts whatever comes next. It works until volume grows or that person goes on holiday — and it is exactly the kind of repetitive intake a script should own.

Scripts that "download attachments" are easy to write and easy to get wrong in production. They process the same message twice after a crash, or never because they marked it read before the download failed. They save invoice.pdf over yesterday's invoice.pdf. They trip over filenames encoded as =?UTF-8?B?UmVjaG51bmcgTsKwIDQ1LnBkZg==?=. They write attachments with names like ../../etc/cron.d/x. They stop working the day the mail provider turns off password login. This guide builds an intake step that is safe, idempotent and observable, using IMAP for standard mailboxes and Microsoft Graph for Microsoft 365, and hands clean files to the rest of your document and data pipeline.

Prerequisites

python -m venv .venv && source .venv/bin/activate
pip install "pandas>=2.2" msal requests python-dotenv
mkdir -p inbox/incoming inbox/processed state

The IMAP path uses only the standard library (imaplib, email). msal and requests are for the Microsoft Graph variant. python-dotenv keeps credentials out of code: put them in a .env file that is excluded from version control, or better, in your scheduler's secret store. Use a dedicated mailbox for automation rather than a person's inbox — it keeps permissions narrow, avoids a script moving someone's personal mail, and makes audit trails clear.

# .env — never commit this file
MAIL_HOST=imap.example.com
MAIL_USER=[email protected]
MAIL_PASSWORD=app-specific-password-here

Diagnostic: Inspect the Mailbox Before Automating

Connect read-only and summarise what arrives: how many unread messages, from whom, with what attachment types and sizes. This tells you which filters and limits the job needs — and confirms authentication works before anything else is built.

# stdlib only
import email
import imaplib
import os
from collections import Counter
from email import policy
from pathlib import Path

def mailbox_profile(limit: int = 50) -> None:
    host, user, password = os.environ["MAIL_HOST"], os.environ["MAIL_USER"], os.environ["MAIL_PASSWORD"]
    try:
        conn = imaplib.IMAP4_SSL(host, 993, timeout=30)
        conn.login(user, password)
    except (imaplib.IMAP4.error, OSError) as exc:
        raise SystemExit(f"cannot log in to {host} as {user}: {exc}")
    try:
        conn.select("INBOX", readonly=True)                     # readonly: nothing gets marked seen
        status, data = conn.uid("SEARCH", None, "UNSEEN")
        uids = data[0].split()[-limit:]
        senders, types = Counter(), Counter()
        for uid in uids:
            _, parts = conn.uid("FETCH", uid, "(BODY.PEEK[])")   # PEEK: do not set \Seen
            msg = email.message_from_bytes(parts[0][1], policy=policy.default)
            senders[msg["From"].addresses[0].addr_spec if msg["From"] else "?"] += 1
            for att in msg.iter_attachments():
                name = att.get_filename() or "(no name)"
                size = len(att.get_payload(decode=True) or b"")
                types[(Path(name).suffix.lower() or att.get_content_type(), size // 100_000 * 100)] += 1
        print(f"{len(data[0].split())} unseen; sampled {len(uids)}")
        print("top senders:", senders.most_common(5))
        print("attachment (type, size bucket KB):", types.most_common(8))
    finally:
        conn.logout()

if __name__ == "__main__":
    from dotenv import load_dotenv
    load_dotenv()
    mailbox_profile()

37 unseen; sampled 37
top senders: [('[email protected]', 14), ('[email protected]', 9), ('[email protected]', 8)]
attachment (type, size bucket KB): [(('.pdf', 0), 22), (('.csv', 0), 9), (('.xlsx', 0), 8), (('image/png', 0), 31)]

Thirty-one inline PNGs are email signature logos — the job must ignore those. If login fails here, stop and fix authentication first; fix IMAP authentication failed error covers app passwords, OAuth and disabled basic authentication.

Core Implementation

Step 1: Search and Fetch by UID

IMAP sequence numbers change when messages are moved or deleted; UIDs do not. Use conn.uid(...) for every command so a message processed in the middle of a run is still identified correctly at the end.

# stdlib only
import email
import imaplib
from email import policy
from email.message import EmailMessage

ALLOWED_SENDERS = {"[email protected]", "[email protected]", "[email protected]"}

def candidate_uids(conn: imaplib.IMAP4_SSL, folder: str = "INBOX") -> list[bytes]:
    status, _ = conn.select(folder)
    if status != "OK":
        raise RuntimeError(f"cannot select {folder}")
    status, data = conn.uid("SEARCH", None, "UNSEEN")
    if status != "OK":
        raise RuntimeError("IMAP search failed")
    return data[0].split()

def fetch_message(conn: imaplib.IMAP4_SSL, uid: bytes) -> EmailMessage:
    status, parts = conn.uid("FETCH", uid, "(BODY.PEEK[])")
    if status != "OK" or not parts or parts[0] is None:
        raise RuntimeError(f"cannot fetch UID {uid!r}")
    return email.message_from_bytes(parts[0][1], policy=policy.default)

policy=policy.default is not optional. The legacy default policy returns raw header strings, so encoded subjects and filenames come back as =?UTF-8?...?=; the modern policy decodes them and exposes structured headers such as msg["From"].addresses. The details are in fix encoded attachment filenames in Python.

Step 2: Filter Attachments Worth Keeping

# stdlib only
from email.message import EmailMessage
from pathlib import Path

ALLOWED_EXT = {".pdf", ".csv", ".xlsx", ".xls", ".docx"}
MAX_BYTES = 25 * 1024 * 1024

def wanted_attachments(msg: EmailMessage) -> list[tuple[str, bytes]]:
    sender = msg["From"].addresses[0].addr_spec.lower() if msg["From"] else ""
    if sender not in ALLOWED_SENDERS:
        return []
    out = []
    for part in msg.iter_attachments():
        name = part.get_filename() or ""
        ext = Path(name).suffix.lower()
        if ext not in ALLOWED_EXT:
            continue                                           # signature images, calendar invites
        if part.get_content_disposition() == "inline" and ext not in {".pdf", ".csv", ".xlsx"}:
            continue
        payload = part.get_payload(decode=True) or b""
        if not payload or len(payload) > MAX_BYTES:
            continue
        out.append((name, payload))
    return out

An allowlist of senders and extensions is a security control, not only a noise filter. A mailbox that accepts mail from anyone will eventually receive a malicious .docm or an archive with a misleading name; the pipeline should only ever touch file types it knows how to process. Checking a PDF's first bytes for %PDF- before handing it on catches renamed executables cheaply.

Step 3: Save Under Safe, Unique Names

# stdlib only
import hashlib
import re
import unicodedata
from datetime import datetime, timezone
from pathlib import Path

INCOMING = Path("inbox/incoming")

def safe_filename(name: str, max_len: int = 80) -> str:
    name = unicodedata.normalize("NFKC", name)
    name = Path(name.replace("\\", "/")).name                  # strip any directory components
    stem, ext = Path(name).stem, Path(name).suffix.lower()
    stem = re.sub(r"[^\w.\-]+", "_", stem).strip("._") or "attachment"
    return f"{stem[:max_len]}{ext}"

def save_attachment(name: str, payload: bytes, received: datetime, sender: str) -> Path:
    digest = hashlib.sha256(payload).hexdigest()[:12]
    stamp = received.astimezone(timezone.utc).strftime("%Y%m%dT%H%M%S")
    domain = sender.split("@")[-1].replace(".", "-")
    final = INCOMING / f"{stamp}_{domain}_{digest}_{safe_filename(name)}"
    if final.exists():
        return final                                           # identical content already saved
    tmp = final.with_suffix(final.suffix + ".part")
    tmp.write_bytes(payload)
    tmp.replace(final)                                          # watchers never see a partial file
    return final

Including a content hash in the name makes saves idempotent: re-processing the same message produces the same filename and is skipped, while two different invoices both called invoice.pdf never collide. Writing to .part and renaming means a folder watcher downstream only ever sees complete files, which is the contract Watching Folders for Incoming Documents relies on.

Step 4: Record State and Move the Message Last

Process each message in a fixed order — save files, record state, then move the message out of the inbox. A crash at any point before the move leaves the message unread in the inbox, so the next run retries; the state store prevents duplicates if the crash happened after saving.

# stdlib only
import json
import sqlite3
from pathlib import Path

STATE = Path("state/mail-intake.sqlite")

def open_state() -> sqlite3.Connection:
    STATE.parent.mkdir(parents=True, exist_ok=True)
    db = sqlite3.connect(STATE)
    db.execute("""CREATE TABLE IF NOT EXISTS processed (
                    message_id TEXT PRIMARY KEY, received TEXT, sender TEXT, files TEXT)""")
    return db

def already_done(db: sqlite3.Connection, message_id: str) -> bool:
    return db.execute("SELECT 1 FROM processed WHERE message_id = ?", (message_id,)).fetchone() is not None

def mark_done(db, message_id: str, received: str, sender: str, files: list[Path]) -> None:
    db.execute("INSERT OR IGNORE INTO processed VALUES (?, ?, ?, ?)",
               (message_id, received, sender, json.dumps([f.name for f in files])))
    db.commit()

def move_message(conn, uid: bytes, target: str = "Processed") -> None:
    caps = conn.capabilities
    if "MOVE" in caps:
        status, _ = conn.uid("MOVE", uid, target)
    else:
        status, _ = conn.uid("COPY", uid, target)
        if status == "OK":
            conn.uid("STORE", uid, "+FLAGS", r"(\Deleted)")
            conn.expunge()
    if status != "OK":
        raise RuntimeError(f"could not move UID {uid!r} to {target}")

This ordering gives at-least-once delivery from the mailbox and exactly-once saving through the state check — the combination most intake jobs actually need. Reversing any two steps reintroduces either lost messages or duplicate files.

Keying state on the Message-ID header rather than the IMAP UID survives mailbox migrations and UID validity resets. Messages without a Message-ID are rare but exist; fall back to a hash of the date, sender and subject.

Step 5: Put the Run Together

# stdlib only + python-dotenv
import imaplib
import logging
import os
from email.utils import parsedate_to_datetime

log = logging.getLogger("mail-intake")

def run_once() -> int:
    conn = imaplib.IMAP4_SSL(os.environ["MAIL_HOST"], 993, timeout=60)
    conn.login(os.environ["MAIL_USER"], os.environ["MAIL_PASSWORD"])
    db, saved_total = open_state(), 0
    try:
        for uid in candidate_uids(conn):
            msg = fetch_message(conn, uid)
            message_id = (msg["Message-ID"] or f"{msg['Date']}|{msg['From']}|{msg['Subject']}").strip()
            sender = msg["From"].addresses[0].addr_spec.lower() if msg["From"] else ""
            received = parsedate_to_datetime(msg["Date"]) if msg["Date"] else None
            if already_done(db, message_id):
                move_message(conn, uid)
                continue
            files = [save_attachment(n, p, received, sender) for n, p in wanted_attachments(msg)]
            mark_done(db, message_id, received.isoformat() if received else "", sender, files)
            move_message(conn, uid, "Processed" if files else "Ignored")
            saved_total += len(files)
            log.info("%s from %s: %d file(s)", message_id, sender, len(files))
    finally:
        db.close()
        conn.logout()
    return saved_total

Messages with no wanted attachments move to Ignored, which keeps the inbox a queue of unprocessed work and gives a person an easy place to check for anything the filters rejected incorrectly.

Edge Cases and Variants

Microsoft 365 Mailboxes

Microsoft has disabled basic authentication for IMAP in Exchange Online, so password logins fail regardless of how correct the password is. Use the Microsoft Graph API with an app registration instead — the full setup, permissions scoping and large-attachment handling are in download attachments from Microsoft 365 with the Graph API.

Attachments Inside Forwarded Messages

A forwarded email arrives as a message/rfc822 attachment that contains its own attachments. iter_attachments() returns the embedded message as one part; recurse into it:

# stdlib only
from email.message import EmailMessage

def iter_all_attachments(msg: EmailMessage):
    for part in msg.iter_attachments():
        if part.get_content_type() == "message/rfc822":
            inner = part.get_content()                          # an EmailMessage
            yield from iter_all_attachments(inner)
        else:
            yield part

Zipped Attachments

Some systems zip reports. Extract only allowed file types, cap the total uncompressed size to defend against zip bombs, and apply the same safe naming to each member.

Validation

Verify each run by reconciling three counts — messages fetched, messages recorded in state, messages moved — and check saved files are what they claim to be.

# stdlib only
from pathlib import Path

MAGIC = {".pdf": b"%PDF-", ".xlsx": b"PK\x03\x04", ".docx": b"PK\x03\x04"}

def verify_saved(folder: Path) -> list[str]:
    problems = []
    for path in folder.iterdir():
        if path.suffix == ".part":
            problems.append(f"leftover partial file {path.name}")
            continue
        expected = MAGIC.get(path.suffix.lower())
        if expected and not path.read_bytes()[:len(expected)] == expected:
            problems.append(f"{path.name}: content does not match its extension")
        if path.stat().st_size == 0:
            problems.append(f"{path.name}: empty file")
    return problems

if __name__ == "__main__":
    issues = verify_saved(Path("inbox/incoming"))
    print("\n".join(issues) or "incoming folder verified")

Leftover .part files mean a run died mid-write; the next run rewrites them safely. A PDF whose first bytes are not %PDF- is either corrupt or not a PDF — quarantine it rather than letting an extraction job fail on it, following the quarantine pattern in quarantine invalid rows in a pipeline.

Performance and Scale Notes

IMAP is slow per round trip, not per byte. Fetch message bodies only for candidates the search already narrowed down, and search server-side by sender and date where possible (FROM, SINCE) rather than downloading everything. A mailbox receiving a few hundred messages a day processes in seconds with one connection. Run the job every few minutes from a scheduler instead of holding an IDLE connection open, unless near-real-time intake matters — long-lived connections drop and need reconnection logic that a short scheduled run avoids. Large attachments dominate transfer time; the size limit also protects the job's memory, since email.message_from_bytes holds the whole message in memory. For very high volumes, the Graph API's delta queries or provider webhooks scale better than polling IMAP.

Troubleshooting

Complete Working Script

#!/usr/bin/env python3
# pip install python-dotenv
"""Fetch allowed attachments from an IMAP mailbox into inbox/incoming, exactly once per message."""
import argparse
import email
import hashlib
import imaplib
import logging
import os
import re
import sqlite3
import sys
from email import policy
from email.utils import parsedate_to_datetime
from pathlib import Path

from dotenv import load_dotenv

ALLOWED_EXT = {".pdf", ".csv", ".xlsx", ".docx"}
MAX_BYTES = 25 * 1024 * 1024
log = logging.getLogger("mail-intake")


def safe_name(name: str) -> str:
    base = Path(name.replace("\\", "/")).name
    stem = re.sub(r"[^\w.\-]+", "_", Path(base).stem).strip("._") or "attachment"
    return stem[:80] + Path(base).suffix.lower()


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--out", type=Path, default=Path("inbox/incoming"))
    ap.add_argument("--state", type=Path, default=Path("state/mail-intake.sqlite"))
    ap.add_argument("--senders", nargs="+", required=True)
    args = ap.parse_args()
    load_dotenv()
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
    args.out.mkdir(parents=True, exist_ok=True)
    args.state.parent.mkdir(parents=True, exist_ok=True)
    allowed = {s.lower() for s in args.senders}

    db = sqlite3.connect(args.state)
    db.execute("CREATE TABLE IF NOT EXISTS processed (message_id TEXT PRIMARY KEY, files INTEGER)")
    try:
        conn = imaplib.IMAP4_SSL(os.environ["MAIL_HOST"], 993, timeout=60)
        conn.login(os.environ["MAIL_USER"], os.environ["MAIL_PASSWORD"])
    except (KeyError, imaplib.IMAP4.error, OSError) as exc:
        log.error("login failed: %s", exc)
        return 2
    saved = 0
    try:
        conn.select("INBOX")
        _, data = conn.uid("SEARCH", None, "UNSEEN")
        for uid in data[0].split():
            _, parts = conn.uid("FETCH", uid, "(BODY.PEEK[])")
            msg = email.message_from_bytes(parts[0][1], policy=policy.default)
            mid = (msg["Message-ID"] or f"{msg['Date']}|{msg['From']}").strip()
            sender = msg["From"].addresses[0].addr_spec.lower() if msg["From"] else ""
            done = db.execute("SELECT 1 FROM processed WHERE message_id=?", (mid,)).fetchone()
            files = 0
            if not done and sender in allowed:
                stamp = parsedate_to_datetime(msg["Date"]).strftime("%Y%m%dT%H%M%S") if msg["Date"] else "nodate"
                for part in msg.iter_attachments():
                    name = part.get_filename() or ""
                    payload = part.get_payload(decode=True) or b""
                    if Path(name).suffix.lower() not in ALLOWED_EXT or not payload or len(payload) > MAX_BYTES:
                        continue
                    digest = hashlib.sha256(payload).hexdigest()[:12]
                    final = args.out / f"{stamp}_{digest}_{safe_name(name)}"
                    if not final.exists():
                        tmp = final.with_suffix(final.suffix + ".part")
                        tmp.write_bytes(payload)
                        tmp.replace(final)
                    files += 1
                db.execute("INSERT OR IGNORE INTO processed VALUES (?, ?)", (mid, files))
                db.commit()
            target = "Processed" if (files or done) else "Ignored"
            conn.uid("COPY", uid, target)
            conn.uid("STORE", uid, "+FLAGS", r"(\Deleted \Seen)")
            saved += files
        conn.expunge()
    finally:
        db.close()
        conn.logout()
    log.info("saved %d attachment(s)", saved)
    return 0


if __name__ == "__main__":
    sys.exit(main())

Create the Processed and Ignored folders in the mailbox once before the first run; COPY to a missing folder fails with [TRYCREATE].

Frequently Asked Questions

IMAP or POP3? IMAP. It supports folders, flags, server-side search and moving messages, which the exactly-once pattern depends on. POP3 downloads and optionally deletes, with no way to mark progress safely.

Is the imap-tools library worth using? It wraps imaplib with a friendlier API for searching, fetching and moving by UID. The concepts in this guide — PEEK, state, move last — apply unchanged; it simply removes boilerplate.

How do I process only today's mail? Combine search criteria: conn.uid("SEARCH", None, "UNSEEN", "SINCE", "17-Sep-2026"). IMAP dates use the DD-Mon-YYYY format and ignore time and timezone.

How do I keep credentials safe on a shared server? Store them in the scheduler's secret store or the operating system keyring and inject them as environment variables at run time. Restrict the automation mailbox so the password cannot send mail or access other mailboxes, and rotate app passwords when people with server access leave the team.

What should happen to attachments the job cannot use? Move the whole message to Ignored and include the reason in the log line — wrong sender, disallowed type, too large. A weekly glance at that folder catches suppliers who changed their sending address long before month-end.

Should the job reply to senders? Only for rejected files, and only from a monitored address. Automatic confirmations for every message create mail loops with other automated senders.

Related

• Fix IMAP Authentication Failed Error — app passwords, OAuth and disabled basic auth
• Fix Encoded Attachment Filenames in Python — RFC 2047 and RFC 2231 names
• Download Attachments from Microsoft 365 with the Graph API — Exchange Online mailboxes
• Process Invoice PDFs from Email into Excel — an end-to-end intake-to-report pipeline
• Scheduling and Logging Automation Jobs — running the intake every few minutes reliably

Part of Automating Document & Data Pipelines.