ctfy.sdk.admin_resources.email

 1"""``client.admin.email`` — preview and test-send the mail templates."""
 2
 3from __future__ import annotations
 4
 5from ctfy.sdk._helpers import _raise_for_status
 6from ctfy.sdk.base import BaseHttpClient
 7from ctfy.server.models import (
 8    EmailSuppressionInfo,
 9    EmailSuppressionListResponse,
10    EmailTemplateListResponse,
11    TestEmailResult,
12)
13
14
15class AdminEmailResource:
16    """The only path that puts a rendered template into a real inbox.
17
18    A template's failure mode is a rendering one that only a specific
19    client exhibits, so judging one means mailing it somewhere and
20    looking. Sends are synchronous (the caller wants the verdict, not a
21    queue receipt), skip the recipient's notification preferences (an
22    admin asking to see a template is not the platform contacting them),
23    and are audited server-side with the address.
24    """
25
26    def __init__(self, http: BaseHttpClient) -> None:
27        self._http = http
28
29    def templates(self) -> EmailTemplateListResponse:
30        """Every template with its preference category, whether that
31        category is forced, and the subject its sample data renders —
32        which doubles as a check that template and sample still agree.
33        """
34        resp = self._http.request("GET", "/admin/email/templates")
35        _raise_for_status(resp)
36        return EmailTemplateListResponse.model_validate(resp.json())
37
38    def send_test(self, template: str, to: str = "") -> TestEmailResult:
39        """Mail one template's sample render to *to* (default: the
40        caller's own address).
41
42        ``sent`` is the provider's verdict, not an enqueue receipt — a
43        false with a populated ``error`` means the message was rejected.
44        """
45        resp = self._http.request(
46            "POST",
47            "/admin/email/test",
48            json={"template": template, "to": to},
49        )
50        _raise_for_status(resp)
51        return TestEmailResult.model_validate(resp.json())
52
53    def suppressions(self, *, limit: int = 100, offset: int = 0) -> EmailSuppressionListResponse:
54        """Addresses the platform will not mail, newest first.
55
56        ``webhook_configured`` is the field to read before concluding
57        anything from an empty list: without a Svix secret nothing is
58        ever recorded automatically, so "no bounces" and "nobody
59        watching" look identical.
60        """
61        resp = self._http.request(
62            "GET",
63            "/admin/email/suppressions",
64            params={"limit": limit, "offset": offset},
65        )
66        _raise_for_status(resp)
67        return EmailSuppressionListResponse.model_validate(resp.json())
68
69    def suppress(self, email: str, detail: str = "") -> EmailSuppressionInfo:
70        """Stop mailing an address, by operator decision.
71
72        Recorded as ``manual`` so it stays distinguishable from the
73        provider's own verdicts — the first question about a suppression
74        that turns out to be wrong is whether a human or a bounce put it
75        there.
76        """
77        resp = self._http.request(
78            "POST",
79            "/admin/email/suppressions",
80            json={"email": email, "detail": detail},
81        )
82        _raise_for_status(resp)
83        return EmailSuppressionInfo.model_validate(resp.json())
84
85    def unsuppress(self, email: str) -> bool:
86        """Let an address receive platform mail again.
87
88        The un-trap. A suppression is invisible to the person it affects
89        — their sign-in codes simply stop arriving — so an entry made by
90        a misconfigured receiving server has to be removable.
91        """
92        resp = self._http.request("DELETE", f"/admin/email/suppressions/{email}")
93        _raise_for_status(resp)
94        removed = resp.json().get("removed")
95        return bool(removed)