Send transactional emails with Python

Contents

The Python SDK sends transactional emails from version 7.81.2. Before you start, create a transactional email and create a project secret API key.

Install

Terminal
pip install posthog

For async code, install posthog[async] instead.

Set up the client

Reuse the PostHog client your application already creates, and set secret_key to your project secret API key. The client still captures events with your project token. Shut the client down when your application exits, not after each send.

import os
from posthog import Posthog
posthog = Posthog(
"<ph_project_token>",
host="https://us.i.posthog.com",
secret_key=os.environ["POSTHOG_PROJECT_SECRET_API_KEY"],
)
  • Host: On PostHog Cloud, the client sends email to the app host of your region, even when host is an ingestion host such as us.i.posthog.com. For any other host, such as a reverse proxy or a self-hosted instance, set app_host to your PostHog app URL. The client refuses to send the secret key to a host that might be a proxy.
  • Local evaluation: The client loads flag definitions only when it evaluates a flag. If your application evaluates flags locally with the same client, give the key the Feature flag read scope too.
  • Email only: In a process that only sends email, you can use Emails or AsyncEmails from posthog.emails instead of a full client.

Send an email

def send_password_reset(user_id: str, email: str, first_name: str, reset_link: str, reset_request_id: str) -> str:
# Raises posthog.emails.EmailError when PostHog doesn't queue the email. Branch on error.code.
result = posthog.emails.send(
to=email,
template_id="<template_id>",
template_data={"first_name": first_name, "reset_link": reset_link},
distinct_id=user_id,
# Identifies this one email, so a retry never sends it twice
idempotency_key=f"password-reset-{reset_request_id}",
)
return result.id

send returns a SendEmailResult, and result.id is the message ID. A send that PostHog doesn't queue raises EmailError, with the code from the errors table. A wrong argument raises TypeError or ValueError before any request. After a network_error or a 503, PostHog might have queued the email. Retry with the same idempotency key rather than sending the email through another provider, or the recipient can get it twice.

Timeouts and retries

Each request times out after 15 seconds. The client retries network errors, rate limits, an in-flight idempotency key, and a 503 up to two times, with the same idempotency key. It waits about 1 and then 2 seconds, or longer if the server's Retry-After asks for it. A Retry-After over 30 seconds returns the error at once.

To change this, set email_request_timeout_seconds and email_request_max_retries on the client, or max_retries on one send. When a job queue already retries the send, set the retries to 0 and pass an idempotency_key that identifies the email.

Pass the distinct ID that you pass to identify(), usually the ID your app gives that user, as distinct_id. PostHog then records the send, and its engagement events such as sent, opened, and clicked, against that user's person, next to their product events.

  • If you leave it out, PostHog records the email against a person whose distinct ID is the recipient's email address. Unless your app identifies users by their email address, that's a separate person from your user.
  • For someone who isn't a user yet, such as an invite or an email address check before the account exists, there's no distinct ID to pass. Leave distinct_id out.

Send content from your code

From version 7.83.0, send also takes content that your code rendered, without a template in PostHog. Name the email with template_key, and pass subject and html, text, or both. PostHog sends the content as you wrote it, so render it with your template engine first, such as Jinja or Django templates.

def send_password_reset(user_id: str, email: str, html: str, reset_request_id: str) -> str:
result = posthog.emails.send(
to=email,
template_key="password-reset",
subject="Reset your password",
html=html,
from_="Hedgebox <security@example.com>",
distinct_id=user_id,
idempotency_key=f"password-reset-{reset_request_id}",
)
return result.id
  • Key: template_key names the email, such as password-reset. PostHog groups the email's metrics and sends under it, so use the same key every time you send this email.
  • Text: If you leave out text, PostHog makes it from html.
  • Sender: The sender is from_, because from is a Python keyword. It needs an address on a domain that your project verified. Without it, PostHog uses the project's default sender. reply_to takes one address or a list.
  • Engagement: PostHog captures the email's engagement events unless you pass capture_engagement=False.
  • One form per send: A send has template_id or template_key, not both, or it raises TypeError. Content from your code takes no template_data, because PostHog renders nothing.

See content from your code for each field and its checks.

Check an email's status

Python
from posthog.emails import EmailError
def email_bounced(message_id: str) -> bool | None:
try:
status = posthog.emails.get(message_id)
except EmailError as error:
if error.code == "not_found":
# The first status can take a few seconds to arrive. Try again later.
return None
raise
return status.status == "bounced"

get returns an EmailStatus with id, status, created_at, status_at, and reason. The two times are datetime objects. See statuses and reasons. The client doesn't retry a not_found, so retry it yourself a few times, with a wait in between, before you treat the email as missing. A status that isn't there yet doesn't mean the email failed.

Test without sending email

Pass your own HTTP transport to check what your code sends, without a network call. This checks your integration, not delivery. To check delivery, send a real email to your own address.

Python
import httpx
from posthog.emails import AsyncEmails
requests = []
def handler(request: httpx.Request) -> httpx.Response:
requests.append(request)
return httpx.Response(202, json={"message_id": "msg_1", "replayed": False})
async def test_send_password_reset():
emails = AsyncEmails(
"phs_test",
host="https://us.posthog.com",
http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)),
)
result = await emails.send(to="jane@example.com", template_id="<template_id>", idempotency_key="password-reset-1")
assert result.id == "msg_1"
assert requests[0].headers["Idempotency-Key"] == "password-reset-1"

For synchronous code, pass a requests.Session with a test adapter to Emails(session=...). Any library that mocks requests or httpx works too.

Still have questions?

Was this page useful?