Python’s email.generator module serialises email.message.Message and EmailMessage objects to text or bytes. from email.generator import BytesGenerator, Generator, DecodedGenerator. Three classes: Generator(outfp, mangle_from_=False, maxheaderlen=78, *, policy=None) — writes text (str) to outfp; mangle_from_=True prepends > to lines starting with From (mbox safety); maxheaderlen controls folding. BytesGenerator(outfp, mangle_from_=False, maxheaderlen=78, *, policy=None) — writes bytes to a binary stream; essential for SMTP sendmail() calls. DecodedGenerator(outfp, mangle_from_=False, maxheaderlen=78, fmt=None, *, policy=None) — writes decoded (human-readable) payloads, stripping transfer encoding; fmt is a format string "%(type)s %(maintype)s %(subtype)s %(filename)s %(desc)s %(encoding)s" for non-text parts. Flatten a message: gen.flatten(msg). Or use msg.as_string(unixfrom=False, maxheaderlen=78, policy=None) / msg.as_bytes() which delegate to Generator/BytesGenerator internally. Policy pairing: BytesGenerator(buf, policy=email.policy.SMTP) produces CRLF-terminated output; Generator(buf, policy=email.policy.default) uses \n. Claude Code generates SMTP wire-format serializers, mbox file writers, policy-aware message renderers, multipart flatteners, and email archive pipelines.
CLAUDE.md for email.generator
## email.generator Stack
- Stdlib: from email.generator import Generator, BytesGenerator, DecodedGenerator
- from email import policy
- Text: buf = io.StringIO()
- Generator(buf, mangle_from_=True).flatten(msg)
- text = buf.getvalue()
- Bytes: buf = io.BytesIO()
- BytesGenerator(buf, policy=policy.SMTP).flatten(msg)
- raw = buf.getvalue() # CRLF wire-format bytes
- Decoded: buf = io.StringIO()
- DecodedGenerator(buf).flatten(msg) # no base64/QP encoding
- Short: msg.as_string() # → str via Generator
- msg.as_bytes() # → bytes via BytesGenerator
- Policy: BytesGenerator(buf, policy=policy.SMTPUTF8) for UTF-8 SMTPemail.generator Serialization Pipeline
# app/emailgeneratorutil.py — text, bytes, decoded, mbox, smtp, diff
from __future__ import annotations
import io
import mailbox
import os
from email import policy as _policy
from email.generator import BytesGenerator, DecodedGenerator, Generator
from email.message import EmailMessage, Message
from email.parser import BytesParser
from typing import Any
# ─────────────────────────────────────────────────────────────────────────────
# 1. Core serialisation helpers
# ─────────────────────────────────────────────────────────────────────────────
def to_string(msg: "Message | EmailMessage",
*,
mangle_from_: bool = False,
maxheaderlen: int = 78,
pol: Any = None) -> str:
"""
Serialise a message to a str using Generator.
Example:
text = to_string(msg)
print(text[:200])
"""
buf = io.StringIO()
gen = Generator(buf, mangle_from_=mangle_from_,
maxheaderlen=maxheaderlen, policy=pol)
gen.flatten(msg)
return buf.getvalue()
def to_bytes(msg: "Message | EmailMessage",
*,
mangle_from_: bool = False,
maxheaderlen: int = 78,
pol: Any = _policy.SMTP) -> bytes:
"""
Serialise a message to bytes using BytesGenerator with SMTP policy by default.
Produces CRLF line endings suitable for SMTP sendmail().
Example:
raw = to_bytes(msg)
smtp.sendmail(from_addr, to_addrs, raw)
"""
buf = io.BytesIO()
gen = BytesGenerator(buf, mangle_from_=mangle_from_,
maxheaderlen=maxheaderlen, policy=pol)
gen.flatten(msg)
return buf.getvalue()
def to_decoded_text(msg: "Message | EmailMessage",
fmt: str | None = None) -> str:
"""
Serialise a message to text with transfer-encoded payloads decoded.
Non-text parts are replaced by a summary line controlled by fmt.
Example:
text = to_decoded_text(msg)
print(text) # base64/QP payloads appear as plain text
"""
buf = io.StringIO()
gen = DecodedGenerator(buf, fmt=fmt)
gen.flatten(msg)
return buf.getvalue()
# ─────────────────────────────────────────────────────────────────────────────
# 2. File output helpers
# ─────────────────────────────────────────────────────────────────────────────
def save_to_eml(msg: "Message | EmailMessage", path: str) -> None:
"""
Save a message to an .eml file (bytes, SMTP policy).
Example:
save_to_eml(msg, "/tmp/outgoing.eml")
"""
with open(path, "wb") as fp:
BytesGenerator(fp, policy=_policy.SMTP).flatten(msg)
def load_from_eml(path: str,
pol: Any = _policy.default) -> EmailMessage:
"""
Load a message from an .eml file.
Example:
msg = load_from_eml("/tmp/outgoing.eml")
print(msg["Subject"])
"""
with open(path, "rb") as fp:
return BytesParser(policy=pol).parse(fp) # type: ignore[return-value]
def append_to_mbox(msg: "Message | EmailMessage", mbox_path: str) -> None:
"""
Append a message to an mbox file using mangle_from_=True for mbox safety.
Example:
append_to_mbox(msg, "/var/mail/archive.mbox")
"""
with open(mbox_path, "ab") as fp:
BytesGenerator(fp, mangle_from_=True,
policy=_policy.compat32).flatten(msg)
fp.write(b"\n")
# ─────────────────────────────────────────────────────────────────────────────
# 3. Policy-aware rendering
# ─────────────────────────────────────────────────────────────────────────────
def render_for_smtp(msg: "Message | EmailMessage") -> bytes:
"""
Render to bytes with policy.SMTP (CRLF, max_line=998).
Example:
raw = render_for_smtp(msg)
with smtplib.SMTP("localhost", 25) as smtp:
smtp.sendmail(from_addr, to_addrs, raw)
"""
return to_bytes(msg, pol=_policy.SMTP)
def render_for_smtputf8(msg: "Message | EmailMessage") -> bytes:
"""
Render to bytes with policy.SMTPUTF8 (UTF-8 headers, CRLF).
Requires an SMTPUTF8-capable server.
Example:
raw = render_for_smtputf8(msg)
"""
return to_bytes(msg, pol=_policy.SMTPUTF8)
def render_for_display(msg: "Message | EmailMessage",
max_line: int = 78) -> str:
"""
Render to human-readable str with configurable line folding.
Example:
print(render_for_display(msg))
"""
pol = _policy.default.clone(max_line_length=max_line)
return to_string(msg, pol=pol)
# ─────────────────────────────────────────────────────────────────────────────
# 4. Multipart inspection via generation
# ─────────────────────────────────────────────────────────────────────────────
def part_sizes(msg: "Message | EmailMessage") -> list[dict[str, Any]]:
"""
Return the serialised byte size of each MIME part in the message tree.
Example:
for info in part_sizes(msg):
print(info["content_type"], info["size_bytes"])
"""
result: list[dict[str, Any]] = []
for part in msg.walk():
buf = io.BytesIO()
BytesGenerator(buf, policy=_policy.SMTP).flatten(part)
result.append({
"content_type": part.get_content_type(),
"size_bytes": len(buf.getvalue()),
"encoding": part.get("Content-Transfer-Encoding", "none"),
"filename": part.get_filename(""),
})
return result
# ─────────────────────────────────────────────────────────────────────────────
# 5. Round-trip verifier
# ─────────────────────────────────────────────────────────────────────────────
def round_trip_check(raw: bytes,
pol: Any = _policy.default) -> dict[str, Any]:
"""
Parse raw bytes, re-serialise, re-parse, and compare Subject/From/To.
Returns a dict with 'match' (bool) and any 'diffs' found.
Example:
report = round_trip_check(raw_email_bytes)
print(report["match"], report["diffs"])
"""
msg1: EmailMessage = BytesParser(policy=pol).parsebytes(raw) # type: ignore
re_serialised = to_bytes(msg1, pol=pol)
msg2: EmailMessage = BytesParser(policy=pol).parsebytes(re_serialised) # type: ignore
diffs: list[str] = []
for header in ("Subject", "From", "To", "Message-ID"):
v1 = msg1.get(header, "")
v2 = msg2.get(header, "")
if str(v1) != str(v2):
diffs.append(f"{header}: {v1!r} → {v2!r}")
return {
"match": len(diffs) == 0,
"diffs": diffs,
"original_bytes": len(raw),
"serialised_bytes": len(re_serialised),
}
# ─────────────────────────────────────────────────────────────────────────────
# Demo
# ─────────────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
print("=== email.generator demo ===")
# Build a sample message
outer = MIMEMultipart("alternative")
outer["Subject"] = "Generator demo"
outer["From"] = "[email protected]"
outer["To"] = "[email protected]"
outer.attach(MIMEText("Hello plain!", "plain", "utf-8"))
outer.attach(MIMEText("<b>Hello HTML!</b>", "html", "utf-8"))
# ── to_string ─────────────────────────────────────────────────────────
print("\n--- to_string ---")
text = to_string(outer)
print(f" length: {len(text)} chars")
print(f" first linesep: {repr(text[:text.index(chr(10))+1])}")
# ── to_bytes (SMTP policy) ─────────────────────────────────────────────
print("\n--- render_for_smtp ---")
raw = render_for_smtp(outer)
print(f" length: {len(raw)} bytes")
first_crlf = raw.find(b"\r\n")
print(f" CRLF present: {first_crlf >= 0} at offset {first_crlf}")
# ── to_decoded_text ────────────────────────────────────────────────────
print("\n--- to_decoded_text ---")
decoded = to_decoded_text(outer)
print(f" length: {len(decoded)} chars")
for line in decoded.splitlines()[:5]:
print(f" {line!r}")
# ── part_sizes ────────────────────────────────────────────────────────
print("\n--- part_sizes ---")
for info in part_sizes(outer):
print(f" {info['content_type']:30s} {info['size_bytes']:5d} bytes"
f" enc={info['encoding']}")
# ── round_trip_check ─────────────────────────────────────────────────
print("\n--- round_trip_check ---")
result = round_trip_check(raw)
print(f" match: {result['match']}")
print(f" orig={result['original_bytes']} → re-serialised={result['serialised_bytes']} bytes")
if result["diffs"]:
for d in result["diffs"]:
print(f" diff: {d}")
# ── maxheaderlen comparison ───────────────────────────────────────────
print("\n--- maxheaderlen 78 vs 200 ---")
long_msg = MIMEText("body")
long_msg["Subject"] = "This is a very long subject line that should get folded at 78 characters by default"
long_msg["From"] = "[email protected]"
long_msg["To"] = "[email protected]"
text78 = to_string(long_msg, maxheaderlen=78)
text200 = to_string(long_msg, maxheaderlen=200)
subj_lines_78 = sum(1 for ln in text78.splitlines() if ln.startswith(("Subject", " ", "\t")))
subj_lines_200 = sum(1 for ln in text200.splitlines() if ln.startswith(("Subject", " ", "\t")))
print(f" maxheaderlen=78 : subject spans {subj_lines_78} line(s)")
print(f" maxheaderlen=200: subject spans {subj_lines_200} line(s)")
print("\n=== done ===")For the msg.as_string() / msg.as_bytes() shorthand — these methods on email.message.Message call Generator/BytesGenerator internally and are convenient for simple cases, but they don’t expose the policy parameter before Python 3.6; passing policy=email.policy.SMTP directly to BytesGenerator gives explicit control over CRLF and line length when targeting wire format output. For the aiofiles (PyPI) alternative — async with aiofiles.open("msg.eml", "wb") as f: await f.write(raw) handles the I/O layer when email.generator serialisation happens in an asyncio context — use aiofiles to pair async I/O with the synchronous BytesGenerator by serialising to a BytesIO buffer first, then writing the buffer with aiofiles.open. The Claude Skills 360 bundle includes email.generator skill sets covering to_string()/to_bytes()/to_decoded_text() core serialisers, save_to_eml()/load_from_eml()/append_to_mbox() file helpers, render_for_smtp()/render_for_smtputf8()/render_for_display() policy renderers, part_sizes() multipart inspector, and round_trip_check() parse-serialise-reparse verifier. Start with the free tier to try message serialisation patterns and email.generator pipeline code generation.