Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions docs/_quarto.yml
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,7 @@ quartodoc:
- name: Email.write_preview_email
- name: Email.write_email_message
- name: Email.preview_send_email
- name: Email.write_quarto_json
- title: Uploading emails
desc: >
Converting emails to Emails,
Expand Down
46 changes: 30 additions & 16 deletions nbmail/ingress.py
Original file line number Diff line number Diff line change
Expand Up @@ -69,7 +69,9 @@ def mjml_to_email(
mjml_markup = processed_mjml._to_mjml()
else:
# String-based MJML, no preprocessing needed
warnings.warn("MJMLTag not detected; treating input as plaintext MJML markup", UserWarning)
warnings.warn(
"MJMLTag not detected; treating input as plaintext MJML markup", UserWarning
)
mjml_markup = mjml_content
inline_attachments = {}

Expand All @@ -78,8 +80,8 @@ def mjml_to_email(
i_email = Email(
html=email_content,
subject="",
rsc_email_supress_report_attachment=False,
rsc_email_supress_scheduled=False,
email_suppress_report_attachment=False,
email_suppress_scheduled=False,
inline_attachments=inline_attachments,
)

Expand Down Expand Up @@ -173,33 +175,45 @@ def quarto_json_to_email(path: str) -> Email:
with open(path, "r", encoding="utf-8") as f:
metadata = json.load(f)

email_html = metadata.get("rsc_email_body_html", "")
email_subject = metadata.get("rsc_email_subject", "")
email_text = metadata.get("rsc_email_body_text", "")


# Support both rsc_-prefixed (Quarto standard) and non-prefixed formats
email_html = metadata.get("rsc_email_body_html", metadata.get("email_body_html", ""))
email_subject = metadata.get("rsc_email_subject", metadata.get("email_subject", ""))
email_text = metadata.get("rsc_email_body_text", metadata.get("email_body_text", ""))

# This is a list of paths that connect dumps attached files into.
# Should be in same output directory
output_files = metadata.get("rsc_output_files", [])
output_files += metadata.get("rsc_email_attachments", [])
output_files += metadata.get("rsc_email_attachments", metadata.get("email_attachments", []))

# Get email images (dictionary: {filename: base64_string})
email_images = metadata.get("rsc_email_images", {})

supress_report_attachment = metadata.get(
"rsc_email_supress_report_attachment", False
email_images = metadata.get("rsc_email_images", metadata.get("email_images", {}))

# Support both old (suppress) and new (suppress) spellings, with both prefixes
suppress_report_attachment = metadata.get(
"rsc_email_suppress_report_attachment",
metadata.get(
"rsc_email_suppress_report_attachment",
metadata.get("email_suppress_report_attachment",
metadata.get("email_suppress_report_attachment", False))
)
)
suppress_scheduled = metadata.get(
"rsc_email_suppress_scheduled",
metadata.get(
"rsc_email_suppress_scheduled",
metadata.get("email_suppress_scheduled",
metadata.get("email_suppress_scheduled", False))
)
)
supress_scheduled = metadata.get("rsc_email_supress_scheduled", False)

i_email = Email(
html=email_html,
text=email_text,
inline_attachments=email_images,
external_attachments=output_files,
subject=email_subject,
rsc_email_supress_report_attachment=supress_report_attachment,
rsc_email_supress_scheduled=supress_scheduled,
email_suppress_report_attachment=suppress_report_attachment,
email_suppress_scheduled=suppress_scheduled,
)

return i_email
75 changes: 71 additions & 4 deletions nbmail/structs.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
from __future__ import annotations
from dataclasses import dataclass, field
import re
import json

from email.message import EmailMessage

Expand Down Expand Up @@ -38,10 +39,10 @@ class Email:
recipients
Optional list of recipient email addresses.

rsc_email_supress_report_attachment
email_suppress_report_attachment
Whether to suppress report attachments (used in some workflows).

rsc_email_supress_scheduled
email_suppress_scheduled
Whether to suppress scheduled sending (used in some workflows).

Examples
Expand All @@ -58,8 +59,8 @@ class Email:

html: str
subject: str
rsc_email_supress_report_attachment: bool | None = None
rsc_email_supress_scheduled: bool | None = None
email_suppress_report_attachment: bool | None = None
email_suppress_scheduled: bool | None = None

# is a list of files in path from current directory
external_attachments: list[str] = field(default_factory=list)
Expand Down Expand Up @@ -224,3 +225,69 @@ def preview_send_email(self):
```
"""
raise NotImplementedError

def write_quarto_json(self, out_file: str = ".output_metadata.json") -> None:
"""
Write the Email to Quarto's output metadata JSON format.

This method serializes the Email object to JSON in the format expected by Quarto,
making it compatible with Quarto's email integration workflows. This is the inverse
of the `quarto_json_to_email()` ingress function.

Parameters
----------
out_file
The file path to write the Quarto metadata JSON. Defaults to ".output_metadata.json".

Returns
-------
None

Examples
--------
```python
email = Email(
html="<p>Hello world</p>",
subject="Test Email",
)
email.write_quarto_json("email_metadata.json")
```

Notes
------
The output JSON includes:\n
- email_subject: The subject line
- email_attachments: List of attachment file paths
- email_body_html: The HTML content of the email
- email_body_text: Plain text version (if present)
- email_images: Dictionary of base64-encoded inline images (only if not empty)
- email_suppress_report_attachment: Suppression flag for report attachments
- email_suppress_scheduled: Suppression flag for scheduled sending
"""
metadata = {
"email_subject": self.subject,
"email_attachments": self.external_attachments or [],
"email_body_html": self.html,
}

# Add optional text field if present
if self.text:
metadata["email_body_text"] = self.text

# Add inline images only if not empty
if self.inline_attachments:
metadata["email_images"] = self.inline_attachments

# Add suppression flags if they are set (not None)
if self.email_suppress_report_attachment is not None:
metadata["email_suppress_report_attachment"] = (
self.email_suppress_report_attachment
)

if self.email_suppress_scheduled is not None:
metadata["email_suppress_scheduled"] = self.email_suppress_scheduled

with open(out_file, "w", encoding="utf-8") as f:
json.dump(metadata, f, indent=2)


144 changes: 144 additions & 0 deletions nbmail/tests/test_egress.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,10 @@


import pytest
import json
import tempfile
import os

from nbmail.egress import (
send_email_with_redmail,
send_email_with_yagmail,
Expand All @@ -11,6 +15,7 @@
send_quarto_email_with_gmail,
)
from nbmail.structs import Email
from nbmail.ingress import quarto_json_to_email


def make_basic_email():
Expand Down Expand Up @@ -286,3 +291,142 @@ def test_not_implemented_functions(send_func):
email = make_basic_email()
with pytest.raises(NotImplementedError):
send_func(email)


# Tests for Email.write_quarto_json() method
def test_email_write_quarto_json_basic():
email = Email(
html="<html><body><p>Test email</p></body></html>",
subject="Test Subject",
text="Plain text version",
external_attachments=["file1.pdf", "file2.csv"],
inline_attachments={"img1": "base64data123"},
email_suppress_report_attachment=True,
email_suppress_scheduled=False,
)

with tempfile.TemporaryDirectory() as tmpdir:
json_path = os.path.join(tmpdir, "test.json")
email.write_quarto_json(json_path)

with open(json_path, "r") as f:
data = json.load(f)

# Check that all expected fields are present (with prefix for Quarto compatibility)
assert data["email_subject"] == "Test Subject"
assert data["email_body_html"] == "<html><body><p>Test email</p></body></html>"
assert data["email_body_text"] == "Plain text version"
assert data["email_attachments"] == ["file1.pdf", "file2.csv"]
assert data["email_images"] == {"img1": "base64data123"}
assert data["email_suppress_report_attachment"] is True
assert data["email_suppress_scheduled"] is False


def test_email_write_quarto_json_minimal():
"""Test writing a minimal email to Quarto JSON format."""
email = Email(
html="<html><body>Minimal</body></html>",
subject="Minimal Subject",
)

with tempfile.TemporaryDirectory() as tmpdir:
json_path = os.path.join(tmpdir, "minimal.json")
email.write_quarto_json(json_path)

with open(json_path, "r") as f:
data = json.load(f)

# Check minimal fields
assert data["email_subject"] == "Minimal Subject"
assert data["email_body_html"] == "<html><body>Minimal</body></html>"
assert data["email_attachments"] == []

# Optional fields should not be present
assert "email_body_text" not in data
assert "email_images" not in data
assert "email_suppress_report_attachment" not in data
assert "email_suppress_scheduled" not in data


def test_email_write_quarto_json_round_trip():
"""Test writing and reading back a Quarto JSON email."""
original_email = Email(
html="<html><body><p>Quarto email</p></body></html>",
subject="Quarto Test",
text="Plain text version",
external_attachments=["output.pdf"],
inline_attachments={"img1": "base64encodedstring"},
email_suppress_report_attachment=True,
email_suppress_scheduled=False,
)

with tempfile.TemporaryDirectory() as tmpdir:
json_path = os.path.join(tmpdir, "roundtrip.json")
original_email.write_quarto_json(json_path)

# Read it back
read_email = quarto_json_to_email(json_path)

# Verify all fields match
assert read_email.subject == original_email.subject
assert read_email.html == original_email.html
assert read_email.text == original_email.text
assert read_email.external_attachments == original_email.external_attachments
assert read_email.inline_attachments == original_email.inline_attachments
assert read_email.email_suppress_report_attachment == original_email.email_suppress_report_attachment
assert read_email.email_suppress_scheduled == original_email.email_suppress_scheduled


def test_email_write_quarto_json_no_attachments():
"""Test writing an email without attachments or images."""
email = Email(
html="<html><body>No attachments</body></html>",
subject="No Attachments",
)

with tempfile.TemporaryDirectory() as tmpdir:
json_path = os.path.join(tmpdir, "no_attachments.json")
email.write_quarto_json(json_path)

with open(json_path, "r") as f:
data = json.load(f)

# Check that attachments and images are empty
assert data["email_attachments"] == []
assert "email_images" not in data


def test_email_write_quarto_json_no_text():
"""Test writing an email without plain text version."""
email = Email(
html="<html><body>HTML only</body></html>",
subject="HTML Only",
)

with tempfile.TemporaryDirectory() as tmpdir:
json_path = os.path.join(tmpdir, "html_only.json")
email.write_quarto_json(json_path)

with open(json_path, "r") as f:
data = json.load(f)

# Plain text should not be present
assert "email_body_text" not in data


def test_email_write_quarto_json_custom_filename():
email = Email(
html="<html><body>Custom</body></html>",
subject="Custom Filename",
)

with tempfile.TemporaryDirectory() as tmpdir:
custom_path = os.path.join(tmpdir, "my_custom_file.json")
email.write_quarto_json(custom_path)

assert os.path.exists(custom_path)

with open(custom_path, "r") as f:
data = json.load(f)

assert data["email_subject"] == "Custom Filename"
12 changes: 6 additions & 6 deletions nbmail/tests/test_ingress.py
Original file line number Diff line number Diff line change
Expand Up @@ -225,8 +225,8 @@ def test_quarto_json_to_email_basic(tmp_path):
"rsc_output_files": ["output.pdf"],
"rsc_email_attachments": ["attachment.csv"],
"rsc_email_images": {"img1": "base64encodedstring"},
"rsc_email_supress_report_attachment": True,
"rsc_email_supress_scheduled": False,
"rsc_email_suppress_report_attachment": True,
"rsc_email_suppress_scheduled": False,
}

json_file = tmp_path / "metadata.json"
Expand All @@ -240,8 +240,8 @@ def test_quarto_json_to_email_basic(tmp_path):
assert result.text == "Plain text version"
assert result.external_attachments == ["output.pdf", "attachment.csv"]
assert result.inline_attachments == {"img1": "base64encodedstring"}
assert result.rsc_email_supress_report_attachment is True
assert result.rsc_email_supress_scheduled is False
assert result.email_suppress_report_attachment is True
assert result.email_suppress_scheduled is False


def test_quarto_json_to_email_minimal(tmp_path):
Expand All @@ -261,8 +261,8 @@ def test_quarto_json_to_email_minimal(tmp_path):
assert result.text == ""
assert result.external_attachments == []
assert result.inline_attachments == {}
assert result.rsc_email_supress_report_attachment is False
assert result.rsc_email_supress_scheduled is False
assert result.email_suppress_report_attachment is False
assert result.email_suppress_scheduled is False


def test_quarto_json_to_email_empty_lists(tmp_path):
Expand Down
4 changes: 2 additions & 2 deletions nbmail/tests/test_structs.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,8 +37,8 @@ def test_subject_inserts_after_body(tmp_path):
email = Email(
html=html,
subject="Test Subject",
rsc_email_supress_report_attachment=False,
rsc_email_supress_scheduled=False,
email_suppress_report_attachment=False,
email_suppress_scheduled=False,
)
out_file = tmp_path / "preview.html"

Expand Down