Customer.io transactional email and messaging client for Altissimo Python projects.
- π Template emails via
send_email()β uses Customer.io transactional templates - π§ Plain text emails via
send_text()β inline content, no template required - π¨ HTML emails via
send_html()β inline content, no template required - π Region support β US and EU data centers
- ποΈ Factory method β
CustomerIOClient.from_env()readsCUSTOMERIO_API_KEYfrom the environment - π Optional dependency β
customerioSDK is lazily imported with a helpful error message - π Typed β full type annotations with
py.typedPEP 561 marker - β
Consistent return type β all methods return a
SendResultdataclass - π Retry with backoff β configurable retry for transient failures
- βοΈ Custom headers β including the RFC 8058 one-click unsubscribe pair
- π Fails loudly β send failures raise
CustomerIOSendErrorby default; opt out per client for batch sends
- Python 3.11+
pip install altissimo-customerio[customerio] # core + Customer.io SDKfrom altissimo.customerio import CustomerIOClient
client = CustomerIOClient.from_env()
# Template email (most common)
result = client.send_email(
to="[email protected]",
transactional_message_id="3",
message_data={"first_name": "Alice", "year": 2026},
)
# Plain text
result = client.send_text(
to="[email protected]",
subject="Hello",
body="Welcome aboard!",
)
# HTML
result = client.send_html(
to="[email protected]",
subject="Hello",
html="<h1>Welcome!</h1>",
reply_to="[email protected]",
)
print(result.delivery_id)By default a failed send raises CustomerIOSendError. The originating SDK
exception is preserved as __cause__, and the full SendResult is attached
as .result:
from altissimo.customerio import CustomerIOClient, CustomerIOSendError
try:
result = client.send_html(to="[email protected]", subject="Hi", html="<p>Hi</p>")
except CustomerIOSendError as exc:
logger.error("Reset email failed (status=%s): %s", exc.status_code, exc)
else:
logger.info("Reset email sent: delivery_id=%s", result.delivery_id)For batch sends where one bad recipient should not abort the run, opt out and
check result.ok yourself:
client = CustomerIOClient.from_env(raise_on_error=False)
for user in users:
result = client.send_email(to=user.email, transactional_message_id="3")
if not result.ok:
failures.append((user.email, result.status_code, result.error))SendResult.raise_for_status() converts a result to an exception on demand
(mirroring requests.Response.raise_for_status), which is handy when you want
swallow semantics in one place and raise semantics in another:
client.send_html(to=..., subject=..., html=...).raise_for_status()Note: the library does not log send failures above
DEBUGwhenraise_on_error=Trueβ it has no recipient or template context worth logging from that frame, and a log there would make the failure look handled. Reporting is the caller's job.
Gmail and Yahoo's bulk-sender rules expect notification and marketing mail to carry a
one-click unsubscribe. That needs both headers β List-Unsubscribe alone only produces
a plain link:
client.send_email(
to=user.email,
transactional_message_id="12",
message_data={"first_name": user.first_name},
headers={
"List-Unsubscribe": f"<https://api.example.com/email/unsubscribe?t={token}>",
"List-Unsubscribe-Post": "List-Unsubscribe=One-Click",
},
# The opt-out of record is our own database, so Customer.io's global
# `unsubscribed` flag (which tracks marketing) must not suppress this.
send_to_unsubscribed=True,
)Your endpoint must answer an unauthenticated POST with a 2xx and no redirect β RFC 8058
forbids redirecting the POST, because browsers have historically turned redirected POSTs into
GETs. Providers will retry, so make it idempotent. Expect GETs on the same URL too: some
clients (Apple Mail, Thunderbird) render the header as a link rather than POSTing it, and mail
gateways and link scanners fetch URLs found in mail β so a GET should never mutate anything.
Note:
send_to_unsubscribedisbool | None. Leave it unset to inherit whatever the transactional message is configured with in the Customer.io dashboard; passFalseto explicitly opt out.Falseis forwarded, not treated as unset.
| Variable | Description | Default |
|---|---|---|
CUSTOMERIO_API_KEY |
Customer.io App API key | (required) |
CUSTOMERIO_REGION |
Data center region (us or eu) |
us |
client = CustomerIOClient(
app_api_key="your-api-key",
region="us", # "us" or "eu"
default_from="[email protected]", # default sender
max_retries=3, # retry transient failures
retry_delay=1.0, # base delay (seconds)
raise_on_error=True, # raise CustomerIOSendError on failure (default)
)| Method | Description |
|---|---|
CustomerIOClient(app_api_key, region?, default_from?, raise_on_error?) |
Create a client with an explicit API key |
CustomerIOClient.from_env(env_var?, region?, default_from?, raise_on_error?) |
Create a client from an environment variable |
send_email(to, transactional_message_id, message_data?, ...) |
Send a template-based transactional email |
send_text(to, subject, body, from_email?, reply_to?, ...) |
Send a plain-text email (inline content) |
send_html(to, subject, html, from_email?, reply_to?, ...) |
Send an HTML email (inline content) |
| Field | Type | Description |
|---|---|---|
ok |
bool |
Whether the request succeeded |
delivery_id |
str |
Delivery ID from Customer.io |
status_code |
int |
HTTP status code (0 on exception) |
body |
dict[str, Any] |
Response body |
error |
str | None |
Error message on failure |
exception |
Exception | None |
Originating SDK exception on failure |
| Method | Description |
|---|---|
raise_for_status(context?) |
Raise CustomerIOSendError if ok is False; no-op otherwise |
| Exception | Raised when |
|---|---|
CustomerIOError |
Base class for all library errors |
CustomerIOImportError |
The customerio SDK is not installed |
CustomerIOSendError |
A send failed (carries status_code and result) |
altissimo.customerio
βββ __init__.py # Public API surface
βββ client.py # CustomerIOClient with lazy SDK initialization
βββ exceptions.py # CustomerIOError, CustomerIOImportError, CustomerIOSendError
βββ models.py # EmailAddress, SendResult dataclasses
βββ py.typed # PEP 561 marker
# Install all dependencies
poetry sync
# Run tests
poetry run pytest
# Run tests with coverage
poetry run pytest --cov=altissimo --cov-report=term-missing
# Run linters
poetry run ruff check .
poetry run ruff format --check .See CONTRIBUTING.md for detailed development guidelines.
See CHANGELOG.md for version history.
For reporting security vulnerabilities, see SECURITY.md.
Apache License 2.0 β see LICENSE for details.