Skip to content

notify

azure_bootstrap.notify

Sender-facing throttling + two-tier notification builders.

Modules:

Name Description
templates

Two-tier (sender / dev) notification body builders.
throttle

Sliding-window per-sender notification throttle.

Functions:

Name Description
build_failure_alert_body

Build an HTML body for a pipeline-failure notification.
build_unprocessable_notification

Returns (sender_subject, sender_body, dev_subject, dev_body).
build_validation_notice_body

Soft-fail observation body. Report generated; these are advisory.
reset_sender_notification_throttle

Test-only. Refuses unless AZURE_BOOTSTRAP_ALLOW_RESET=1.
should_notify_sender

Return True when the sender is allowed another notification.

build_failure_alert_body 

build_failure_alert_body(*, attachment_name: str, correlation_id: str, sender: str, error_summary: str, audience: str, product_name: str = '', sender_send_failed: bool = False, action_steps: tuple[str, ...] | None = None, **extra_context: Any) -> str

Build an HTML body for a pipeline-failure notification.

Sender audience: generic acknowledgement, no forensic leakage. Dev audience: full traceback / blob path / exception type rendered.

Source code in azure_bootstrap/notify/templates.py 
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def build_failure_alert_body(
    *,
    attachment_name: str,
    correlation_id: str,
    sender: str,
    error_summary: str,
    audience: str,
    product_name: str = "",
    sender_send_failed: bool = False,
    action_steps: tuple[str, ...] | None = None,
    **extra_context: Any,
) -> str:
    """Build an HTML body for a pipeline-failure notification.

    Sender audience: generic acknowledgement, no forensic leakage.
    Dev audience: full traceback / blob path / exception type rendered.
    """
    if audience == "sender":
        steps = action_steps or _DEFAULT_SENDER_ACTION_STEPS
        steps_html = "<ul>" + "".join(f"<li>{escape(s)}</li>" for s in steps) + "</ul>"
        product_clause = (
            f"the {escape(product_name)} you submitted" if product_name else "your submission"
        )
        rows = [
            _row("Attachment", attachment_name),
        ]
        return (
            '<html><body style="font-family:Helvetica,Arial,sans-serif;color:#333;">'
            f"<p>We hit a problem processing {product_clause}. "
            "Our team has been notified and is looking into it.</p>"
            + _table(rows)
            + "<h3 style='margin-top:16px;'>What you can do</h3>"
            + steps_html
            + "</body></html>"
        )

    # dev audience — full forensic detail
    warning = ""
    if sender_send_failed:
        warning = (
            '<p style="background:#fee;padding:8px;border:1px solid #c00;color:#c00;">'
            "⚠ Sender-facing notification ALSO failed to send — manual follow-up needed."
            "</p>"
        )
    product_clause = f"a {escape(product_name)} submission" if product_name else "a submission"
    rows = [
        _row("Attachment", attachment_name),
        _row("Sender", sender),
        _row("Correlation ID", correlation_id),
        _row("Error", error_summary),
    ]
    for key, value in extra_context.items():
        rows.append(_row(key, str(value)))
    return (
        '<html><body style="font-family:Helvetica,Arial,sans-serif;color:#333;">'
        + warning
        + f"<p>Pipeline failure: {product_clause} could not be processed "
        "and has been dead-lettered.</p>" + _table(rows) + "</body></html>"
    )

build_unprocessable_notification 

build_unprocessable_notification(*, failure_reason: UnprocessableReason, sender: str, attachment_summary: list[dict[str, Any]], correlation_id: str, product_name: str = '', email_subject: str | None = None, email_id: str | None = None) -> tuple[str, str, str, str]

Returns (sender_subject, sender_body, dev_subject, dev_body).

Sender body is template-driven, no internal jargon, no correlation id. Dev body has the full forensic table.

Source code in azure_bootstrap/notify/templates.py 
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
def build_unprocessable_notification(
    *,
    failure_reason: UnprocessableReason,
    sender: str,
    attachment_summary: list[dict[str, Any]],
    correlation_id: str,
    product_name: str = "",
    email_subject: str | None = None,
    email_id: str | None = None,
) -> tuple[str, str, str, str]:
    """Returns ``(sender_subject, sender_body, dev_subject, dev_body)``.

    Sender body is template-driven, no internal jargon, no correlation id.
    Dev body has the full forensic table.
    """
    product = product_name or "submission"
    sender_subject = f"Unable to process your {product} submission"
    dev_subject = (
        f"[{product_name}] Unprocessable submission: {failure_reason.value}"
        if product_name
        else f"Unprocessable submission: {failure_reason.value}"
    )

    body_copy = _UNPROCESSABLE_TEMPLATES[failure_reason]
    sender_body = (
        '<html><body style="font-family:Helvetica,Arial,sans-serif;color:#333;">'
        f"<p>{escape(body_copy)}</p>"
        "</body></html>"
    )

    summary_rows = []
    for item in attachment_summary:
        name = escape(str(item.get("name", "")))
        size = escape(str(item.get("size_bytes", "")))
        mime = escape(str(item.get("content_type", "")))
        classification = escape(str(item.get("classification", "")))
        reject = escape(str(item.get("reject_reason", "")))
        summary_rows.append(
            "<tr>"
            f'<td style="padding:4px 8px;border:1px solid #ddd;">{name}</td>'
            f'<td style="padding:4px 8px;border:1px solid #ddd;">{size}</td>'
            f'<td style="padding:4px 8px;border:1px solid #ddd;">{mime}</td>'
            f'<td style="padding:4px 8px;border:1px solid #ddd;">{classification}</td>'
            f'<td style="padding:4px 8px;border:1px solid #ddd;">{reject}</td>'
            "</tr>"
        )
    summary_table = (
        '<table style="border-collapse:collapse;border:1px solid #ddd;width:100%;">'
        "<thead><tr>"
        '<th style="padding:4px 8px;border:1px solid #ddd;">Name</th>'
        '<th style="padding:4px 8px;border:1px solid #ddd;">Size</th>'
        '<th style="padding:4px 8px;border:1px solid #ddd;">MIME</th>'
        '<th style="padding:4px 8px;border:1px solid #ddd;">Classified</th>'
        '<th style="padding:4px 8px;border:1px solid #ddd;">Reject reason</th>'
        "</tr></thead><tbody>" + "".join(summary_rows) + "</tbody></table>"
    )
    dev_rows = [
        _row("Failure reason", failure_reason.value),
        _row("Sender", sender),
        _row("Correlation ID", correlation_id),
    ]
    if email_subject:
        dev_rows.append(_row("Email subject", email_subject))
    if email_id:
        dev_rows.append(_row("Email ID", email_id))

    dev_body = (
        '<html><body style="font-family:Helvetica,Arial,sans-serif;color:#333;">'
        "<p>Inbound submission rejected by the ingress gates.</p>"
        + _table(dev_rows)
        + "<h3>Attachments seen</h3>"
        + summary_table
        + "</body></html>"
    )
    return sender_subject, sender_body, dev_subject, dev_body

build_validation_notice_body 

build_validation_notice_body(*, attachment_name: str, correlation_id: str, sender: str, issues: list[Any], audience: str, product_name: str = '', sender_send_failed: bool = False, **extra_context: Any) -> str

Soft-fail observation body. Report generated; these are advisory.

Source code in azure_bootstrap/notify/templates.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def build_validation_notice_body(
    *,
    attachment_name: str,
    correlation_id: str,
    sender: str,
    issues: list[Any],
    audience: str,
    product_name: str = "",
    sender_send_failed: bool = False,
    **extra_context: Any,
) -> str:
    """Soft-fail observation body. Report generated; these are advisory."""
    issue_rows = []
    for issue in issues:
        rule = escape(str(getattr(issue, "rule", "") or ""))
        page = escape(str(getattr(issue, "page_number", "") or ""))
        equipment = escape(str(getattr(issue, "equipment_id", "") or ""))
        message = escape(str(getattr(issue, "message", "") or ""))
        issue_rows.append(
            "<tr>"
            f'<td style="padding:4px 8px;border:1px solid #ddd;">{rule}</td>'
            f'<td style="padding:4px 8px;border:1px solid #ddd;">{page}</td>'
            f'<td style="padding:4px 8px;border:1px solid #ddd;">{equipment}</td>'
            f'<td style="padding:4px 8px;border:1px solid #ddd;">{message}</td>'
            "</tr>"
        )
    issue_table = (
        '<table style="border-collapse:collapse;border:1px solid #ddd;margin-top:8px;width:100%;">'
        "<thead><tr>"
        '<th style="padding:4px 8px;border:1px solid #ddd;">Rule</th>'
        '<th style="padding:4px 8px;border:1px solid #ddd;">Page</th>'
        '<th style="padding:4px 8px;border:1px solid #ddd;">Equipment</th>'
        '<th style="padding:4px 8px;border:1px solid #ddd;">Message</th>'
        "</tr></thead><tbody>" + "".join(issue_rows) + "</tbody></table>"
    )

    product_clause = product_name or "report"
    if audience == "sender":
        return (
            '<html><body style="font-family:Helvetica,Arial,sans-serif;color:#333;">'
            f"<p>Your {escape(product_clause)} has been generated and uploaded. "
            "Data quality observations were noted; please review at your "
            "convenience.</p>" + issue_table + "</body></html>"
        )

    warning = ""
    if sender_send_failed:
        warning = (
            '<p style="background:#fee;padding:8px;border:1px solid #c00;color:#c00;">'
            "⚠ Sender-facing notice ALSO failed to send."
            "</p>"
        )
    triage_rows = [_row(k, str(v)) for k, v in extra_context.items()]
    return (
        '<html><body style="font-family:Helvetica,Arial,sans-serif;color:#333;">'
        + warning
        + f"<p>Validation observations: {escape(product_clause)} generated and "
        "uploaded successfully. Listed rules are advisory.</p>"
        + issue_table
        + _table(
            [
                _row("Sender", sender),
                _row("Correlation ID", correlation_id),
                _row("Attachment", attachment_name),
                *triage_rows,
            ]
        )
        + "</body></html>"
    )

reset_sender_notification_throttle 

reset_sender_notification_throttle() -> None

Test-only. Refuses unless AZURE_BOOTSTRAP_ALLOW_RESET=1.

Source code in azure_bootstrap/notify/throttle.py 
53
54
55
56
57
58
59
60
def reset_sender_notification_throttle() -> None:
    """Test-only. Refuses unless AZURE_BOOTSTRAP_ALLOW_RESET=1."""
    if os.environ.get("AZURE_BOOTSTRAP_ALLOW_RESET") != "1":
        raise RuntimeError(
            "reset_sender_notification_throttle is test-only — " "set AZURE_BOOTSTRAP_ALLOW_RESET=1"
        )
    with _throttle_lock:
        _throttle_state.clear()

should_notify_sender 

should_notify_sender(sender_email: str, *, max_per_hour: int = NOTIFY_SENDER_MAX_PER_HOUR_DEFAULT, window_seconds: float = NOTIFY_SENDER_WINDOW_SECONDS_DEFAULT, counter_namespace: str = 'sender') -> bool

Return True when the sender is allowed another notification.

Source code in azure_bootstrap/notify/throttle.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def should_notify_sender(
    sender_email: str,
    *,
    max_per_hour: int = NOTIFY_SENDER_MAX_PER_HOUR_DEFAULT,
    window_seconds: float = NOTIFY_SENDER_WINDOW_SECONDS_DEFAULT,
    counter_namespace: str = "sender",
) -> bool:
    """Return True when the sender is allowed another notification."""
    key = (sender_email or "").strip().lower()
    if not key:
        bump_counter(f"{counter_namespace}.notification.throttled")
        return False
    now = time.monotonic()
    cutoff = now - window_seconds
    with _throttle_lock:
        bucket = _throttle_state.setdefault(key, deque())
        while bucket and bucket[0] < cutoff:
            bucket.popleft()
        if len(bucket) >= max_per_hour:
            bump_counter(f"{counter_namespace}.notification.throttled")
            return False
        bucket.append(now)
    bump_counter(f"{counter_namespace}.notified")
    return True