feat: send transactional email from posthog templates

.sampo/changesets/transactional-emails.md

@@ -0,0 +1,5 @@
+---
+pypi/posthog: minor
+---
+
+Send transactional email from PostHog email templates with `client.emails.send()` on `Posthog` and `AsyncPosthog`, or with `Emails` and `AsyncEmails` from `posthog.emails` in a process that only sends email. A send needs a project secret API key, goes to the app host, sends an idempotency key, and retries network errors, rate limits and an unavailable service with the same key. Set `app_host`, `email_request_timeout_seconds` and `email_request_max_retries` on the client to configure sends. Errors raise `EmailError` with PostHog's error `code` and a Resend-style `name`.

posthog/async_client.py

@@ -10,7 +10,7 @@
 import warnings
 import weakref
 from datetime import datetime, timezone
-from typing import Any, Dict, Mapping, Optional, Union
+from typing import Any, Callable, Dict, Mapping, Optional, Union
 from uuid import UUID, uuid4
 
 from typing_extensions import Unpack
@@ -70,6 +70,7 @@
     try_attach_code_variables_to_frames,
 )
 from .consumer import MAX_MSG_SIZE
+from .emails import AsyncEmails, _client_emails, _validate_options
 from .feature_flag_evaluations import (
     FeatureFlagEvaluations,
     _EvaluatedFlagRecord,
@@ -129,6 +130,9 @@ def __init__(
         personal_api_key: Optional[str] = None,
         feature_flags_request_timeout_seconds: float = 3,
         feature_flags_request_max_retries: int = 1,
+        app_host: Optional[str] = None,
+        email_request_timeout_seconds: float = 15,
+        email_request_max_retries: int = 2,
     ) -> None:
         if flush_at <= 0:
             raise ValueError("flush_at must be greater than zero")
@@ -173,6 +177,10 @@ def __init__(
             else resolved_secret_key
         ) or None
         self.personal_api_key = self.secret_key
+        _validate_options(email_request_timeout_seconds, email_request_max_retries)
+        self.app_host = app_host
+        self.email_request_timeout_seconds = email_request_timeout_seconds
+        self.email_request_max_retries = email_request_max_retries
         self.feature_flags_request_timeout_seconds = (
             feature_flags_request_timeout_seconds
         )
@@ -247,6 +255,48 @@ def __init__(
         self._duplicate_client_registry_key: Optional[tuple[str, str]] = None
         self._register_duplicate_client()
 
+    _emails: Optional[AsyncEmails] = None
+
+    @property
+    def emails(self) -> AsyncEmails:
+        """
+        Send transactional email from PostHog email templates.
+
+        Needs ``secret_key``, a project secret API key with the ``hog_flow:write``
+        scope, and ``posthog[async]``. Sends go to ``app_host``, or to the PostHog
+        Cloud app host of ``host``. ``email_request_timeout_seconds`` and
+        ``email_request_max_retries`` configure each send. A disabled or shut-down
+        client sends nothing. See ``posthog.emails`` for setup.
+
+        Examples:
+            ```python
+            await client.emails.send(
+                to="user@example.com",
+                template_id="0190f6d2-7c1e-7b3a-9f5e-1c2d3e4f5a6b",
+                distinct_id="user_123",
+            )
+            ```
+        """
+        if self._emails is None:
+            self._emails = _client_emails(
+                AsyncEmails,
+                self.secret_key,
+                self.host,
+                lambda: self.disabled,
+                self._admit_email_send,
+                app_host=self.app_host,
+                timeout=self.email_request_timeout_seconds,
+                max_retries=self.email_request_max_retries,
+            )
+        return self._emails
+
+    def _admit_email_send(self) -> Optional[Callable[[], None]]:
+        # Shutdown waits for in-flight operations before it closes the email transport.
+        if not self._accepting or self._closed:
+            return None
+        completion = self._start_inflight_operation()
+        return lambda: self._finish_inflight_operation(completion)
+
     async def __aenter__(self) -> AsyncClient:
         self._ensure_workers_started()
         return self
@@ -1217,6 +1267,13 @@ async def shutdown(self) -> None:
                 self.log.exception("Failed to close async capture transport")
                 errors.append(error)
 
+            if self._emails is not None:
+                try:
+                    await self._emails.aclose()
+                except Exception as error:
+                    self.log.exception("Failed to close the async email transport")
+                    errors.append(error)
+
             try:
                 self._unregister_duplicate_client()
             except Exception as error:

posthog/client.py

@@ -53,6 +53,7 @@
     set_context_session as _context_set_context_session,
     tag as _context_tag,
 )
+from posthog.emails import Emails, _client_emails, _no_op, _validate_options
 from posthog.exception_capture import ExceptionCapture
 from posthog._logging import _configure_posthog_logging
 from posthog.exception_utils import (
@@ -722,6 +723,9 @@ def __init__(
         _use_ai_lane=False,
         _enable_multimodal_capture=False,
         traces: Optional[dict] = None,
+        app_host: Optional[str] = None,
+        email_request_timeout_seconds: float = 15,
+        email_request_max_retries: int = 2,
     ):
         """
         Initialize a new PostHog client instance.
@@ -773,6 +777,15 @@ def __init__(
             feature_flags_request_max_retries: Number of retries for feature flag
                 requests after network, transport, or timeout failures. Defaults
                 to 1. Set to 0 to disable retries.
+            app_host: Your PostHog app host, such as ``https://posthog.example.com``.
+                ``emails`` sends there. Needed only when ``host`` is not PostHog
+                Cloud, for example a reverse proxy, which must not receive the
+                secret key.
+            email_request_timeout_seconds: Timeout in seconds for each email send
+                request. Defaults to 15.
+            email_request_max_retries: Number of retries for an email send after a
+                network error, a rate limit, or an unavailable service. Defaults to 2.
+                Set to 0 when a job runner already retries the send.
             super_properties: Properties merged into every captured event.
             enable_exception_autocapture: Automatically capture uncaught
                 exceptions.
@@ -1037,6 +1050,10 @@ def __init__(
             else resolved_secret_key
         ) or None
         self.personal_api_key = self.secret_key
+        _validate_options(email_request_timeout_seconds, email_request_max_retries)
+        self.app_host = app_host
+        self.email_request_timeout_seconds = email_request_timeout_seconds
+        self.email_request_max_retries = email_request_max_retries
         if debug:
             # Ensures that debug level messages are logged when debug mode is on.
             # Otherwise, defaults to WARNING level. See https://docs.python.org/3/howto/logging.html#what-happens-if-no-configuration-is-provided
@@ -1131,6 +1148,42 @@ def _set_library_identity(self, library_id: str, library_version: str) -> None:
     def _request_identity_kwargs(self) -> Dict[str, str]:
         return {"_user_agent": self._sdk_info} if self._sdk_info != _USER_AGENT else {}
 
+    _emails: Optional[Emails] = None
+
+    @property
+    def emails(self) -> Emails:
+        """
+        Send transactional email from PostHog email templates.
+
+        Needs ``secret_key``, a project secret API key with the ``hog_flow:write``
+        scope. Sends go to ``app_host``, or to the PostHog Cloud app host of
+        ``host``. ``email_request_timeout_seconds`` and ``email_request_max_retries``
+        configure each send. A disabled or shut-down client sends nothing. See
+        ``posthog.emails`` for setup.
+
+        Examples:
+            ```python
+            client.emails.send(
+                to="user@example.com",
+                template_id="0190f6d2-7c1e-7b3a-9f5e-1c2d3e4f5a6b",
+                template_data={"name": "Ada"},
+                distinct_id="user_123",
+            )
+            ```
+        """
+        if self._emails is None:
+            self._emails = _client_emails(
+                Emails,
+                self.secret_key,
+                self.host,
+                lambda: self.disabled,
+                lambda: None if self._shutdown_requested else _no_op,
+                app_host=self.app_host,
+                timeout=self.email_request_timeout_seconds,
+                max_retries=self.email_request_max_retries,
+            )
+        return self._emails
+
     @property
     def queue(self) -> Queue:
         """The analytics lane's queue (kept for backwards compatibility)."""
@@ -2320,6 +2373,8 @@ def _reinit_after_fork(self):
         self._lifecycle_lock = threading.Lock()
         self._lifecycle_condition = threading.Condition(self._lifecycle_lock)
         self._lifecycle_owner = None
+        # The parent's email session and its lock are not safe to share with the child.
+        self._emails = None
         self._deferred_lifecycle_thread_pending = False
         self._deferred_lifecycle_dirty = False
         self._lifecycle_callback_context = ContextVar(
@@ -2958,6 +3013,12 @@ def _join_once(
                 self._shutdown_flag_definition_cache_provider,
                 errors,
             )
+            if self._emails is not None:
+                self._run_lifecycle_cleanup(
+                    "Failed to close the email transport during lifecycle cleanup",
+                    self._emails.close,
+                    errors,
+                )
             self._run_lifecycle_cleanup(
                 "Failed to unregister client during lifecycle cleanup",
                 self._unregister_duplicate_client,