alert-infra 0.1.2

Reusable alert detection infrastructure for Django and Python projects

alert_infra

alert_infra is a reusable alerting infrastructure package for Python applications. It gives teams one consistent way to build, sanitize, and deliver operational alerts from both plain Python services and Django projects.

The package is intentionally small: the core alert model and dispatcher have no mandatory third-party runtime dependencies, while Django support is isolated under alert_infra.django.

Table of contents

• What this project provides
• Architecture
• Installation
• Core concepts
• Non-Django / plain Python usage
• Django usage
• Transport configuration
• Security and redaction
• Error handling and delivery results
• Custom transports
• Testing patterns
• Compatibility namespace
• API reference
• Development

What this project provides

Use alert_infra when an application needs to notify operations, security, support, or engineering teams about important events such as failed payments, blocked security actions, background-job failures, webhook failures, or critical business-state changes.

Key capabilities:

• Framework-agnostic Alert objects with title, message, severity, source, tags, metadata, timestamps, correlation IDs, and request IDs.
• Built-in sensitive metadata redaction before transport delivery.
• Multi-transport dispatching through AlertDispatcher.
• Safe partial-failure handling: one broken transport does not prevent other transports from receiving the alert.
• Built-in transports for:
  • No-op/in-memory delivery.
  • Resend email API.
  • SendGrid Mail Send API.
  • SMTP email fallback.
  • Django email backend for projects that explicitly opt into Django's mail backend.
  • Slack incoming webhooks.
  • Telegram Bot API messages.
• Django settings adapter and send_alert helper.
• Django request metadata extraction with defensive redaction of sensitive headers.
• Compatibility exports for projects that still import feature_flag_infra.

Architecture

The codebase is split into framework-neutral modules and optional integration modules:

Module	Purpose
alert_infra.alert	Core Alert dataclass and severity validation.
alert_infra.security	Recursive metadata redaction helpers.
alert_infra.transports	Transport protocol, delivery result, no-op transport, and dispatcher.
alert_infra.email	Framework-agnostic Resend, SendGrid, SMTP email transports and email formatting helpers.
alert_infra.apps	Application/webhook transports such as Slack and Telegram.
alert_infra.django	Django settings loader, Django email transport, request context helper, and send_alert.
feature_flag_infra	Compatibility namespace that re-exports the same public API.

Typical flow:

1. Your application creates an Alert directly, or calls the Django send_alert helper.
2. Sensitive metadata is redacted during Alert initialization unless explicitly disabled.
3. An AlertDispatcher sends the alert to one or more transports.
4. Each transport receives the same sanitized Alert object.
5. The dispatcher returns a DeliveryResult containing successful and failed transport names.

Installation

Install the base package for plain Python, Resend, SendGrid, SMTP, Slack, and Telegram usage:

pip install alert-infra

Install Django support when using the alert_infra.django adapter:

pip install "alert-infra[django]"

Install the legacy template-email provider chain only if an older application still imports alert_infra.email.tasks, alert_infra.email.services, or alert_infra.email.providers directly:

pip install "alert-infra[legacy-email]"

The active Resend and SendGrid alert transports use the providers' HTTP APIs through Python's standard library, so no provider SDK is required. The legacy extra installs requests and sendgrid only for older code that imports the legacy template-email path directly.

For local development from this repository:

git clone <repository-url>
cd alert_infra
python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"
pytest

Core concepts

Alert

Alert is the domain object delivered to all transports.

from alert_infra import Alert

alert = Alert(
    title="Payment provider timeout",
    message="The payment provider did not respond within 10 seconds.",
    severity="critical",
    source="billing-service",
    tags=("payments", "provider"),
    metadata={
        "invoice_id": "INV-1001",
        "provider": "stripe",
        "api_key": "will-be-redacted",
    },
    request_id="req-01HZY...",
)

Supported severities are:

• info
• warning
• error
• critical

title and message are required. Invalid severities raise AlertValidationError.

Every alert receives a generated correlation_id when one is not provided. Naive created_at datetimes are treated as UTC.

Dispatcher

AlertDispatcher sends an alert to a list of transports.

from alert_infra import Alert, AlertDispatcher, NoOpTransport

transport = NoOpTransport()
dispatcher = AlertDispatcher([transport])

result = dispatcher.send(Alert(title="Smoke test", message="Alert pipeline is reachable."))

assert result.ok is True
assert result.sent == ("noop",)

DeliveryResult

dispatcher.send(...) returns a DeliveryResult:

if result.ok:
    print("alert delivered to", result.sent)
else:
    print("successful transports:", result.sent)
    print("failed transports:", result.failed)

• sent is a tuple of transport names that succeeded.
• failed is a dictionary of {transport_name: exception_class_name}.
• ok is True when no transport failed.

Non-Django / plain Python usage

Plain Python projects should import from the framework-agnostic modules only. Do not import alert_infra.django unless Django is installed and configured.

Minimal no-op example

This is useful for local development or unit tests.

from alert_infra import Alert, AlertDispatcher, NoOpTransport

dispatcher = AlertDispatcher([NoOpTransport()])

result = dispatcher.send(
    Alert(
        title="Local alert",
        message="This alert is stored in memory only.",
        severity="info",
        source="local-script",
    )
)

print(result.sent)  # ("noop",)

Resend email from environment variables

Set environment variables:

export ALERT_FROM_EMAIL=[e-mail]
export ALERT_TO_EMAILS=[e-mail],[e-mail]
export ALERT_RESEND_API_KEY='from-secret-manager'

Send an email alert through Resend without installing the Resend SDK:

from alert_infra import Alert, AlertDispatcher
from alert_infra.email import ResendEmailTransport

email_transport = ResendEmailTransport.from_env()
dispatcher = AlertDispatcher([email_transport])

dispatcher.send(Alert(title="Payment failure", message="Payment provider returned HTTP 500."))

SendGrid email from environment variables

Set environment variables:

export ALERT_FROM_EMAIL=[e-mail]
export ALERT_TO_EMAILS=[e-mail],[e-mail]
export ALERT_SENDGRID_API_KEY='from-secret-manager'

Send an email alert through SendGrid without installing the SendGrid SDK:

from alert_infra import Alert, AlertDispatcher
from alert_infra.email import SendGridEmailTransport

email_transport = SendGridEmailTransport.from_env()
dispatcher = AlertDispatcher([email_transport])

dispatcher.send(Alert(title="Payment failure", message="Payment provider returned HTTP 500."))

SMTP email from environment variables

Set environment variables:

export ALERT_FROM_EMAIL=[e-mail]
export ALERT_TO_EMAILS=[e-mail],[e-mail]
export ALERT_SMTP_HOST=smtp.example.com
export ALERT_SMTP_PORT=587
export ALERT_SMTP_USERNAME=[e-mail]
export ALERT_SMTP_PASSWORD='from-secret-manager'
export ALERT_SMTP_USE_TLS=true

Send an email alert:

from alert_infra import Alert, AlertDispatcher
from alert_infra.email import SMTPEmailTransport

email_transport = SMTPEmailTransport.from_env()
dispatcher = AlertDispatcher([email_transport])

result = dispatcher.send(
    Alert(
        title="Payment failure",
        message="Payment provider returned HTTP 500.",
        severity="error",
        source="billing-service",
        tags=("payments", "provider"),
        metadata={"invoice_id": "INV-001", "customer_id": 42},
    )
)

if not result.ok:
    # Log or report the failed transport names. Do not log sensitive metadata.
    print(result.failed)

SMTP email with explicit constructor settings

from alert_infra import Alert, AlertDispatcher
from alert_infra.email import SMTPEmailTransport

transport = SMTPEmailTransport(
    host="smtp.example.com",
    port=587,
    from_email="[e-mail]",
    to_emails=["[e-mail]"],
    username="[e-mail]",
    password="from-secret-manager",
    use_tls=True,
    timeout=8.0,
)

dispatcher = AlertDispatcher([transport])
dispatcher.send(Alert(title="Disk usage high", message="/var is above 90%.", severity="warning"))

Slack webhook alerts

Set an environment variable:

export ALERT_SLACK_WEBHOOK_URL='https://hooks.slack.com/services/...'

Create and use the transport:

from alert_infra import Alert, AlertDispatcher
from alert_infra.apps import SlackWebhookTransport

slack = SlackWebhookTransport.from_env()
dispatcher = AlertDispatcher([slack])

dispatcher.send(
    Alert(
        title="Webhook delivery failed",
        message="Customer callback endpoint returned HTTP 503.",
        severity="warning",
        source="webhook-worker",
        metadata={"endpoint_id": "ep_123", "authorization": "Bearer secret"},
    )
)

Slack webhook URLs must be absolute https:// URLs.

Telegram bot alerts

Set environment variables:

export ALERT_TELEGRAM_BOT_TOKEN='123456:telegram-token'
export ALERT_TELEGRAM_CHAT_ID='-1001234567890'

Create and use the transport:

from alert_infra import Alert, AlertDispatcher
from alert_infra.apps import TelegramBotTransport

telegram = TelegramBotTransport.from_env()
dispatcher = AlertDispatcher([telegram])

dispatcher.send(
    Alert(
        title="Nightly import failed",
        message="The supplier CSV import exited with status 1.",
        severity="critical",
        source="import-job",
    )
)

Multiple transports in one dispatcher

from alert_infra import Alert, AlertDispatcher
from alert_infra.apps import SlackWebhookTransport, TelegramBotTransport
from alert_infra.email import SMTPEmailTransport

transports = [
    SMTPEmailTransport.from_env(),
    SlackWebhookTransport.from_env(),
    TelegramBotTransport.from_env(),
]

dispatcher = AlertDispatcher(transports)
result = dispatcher.send(Alert(title="API outage", message="Health checks are failing.", severity="critical"))

# Example partial-failure result:
# result.sent == ("email.smtp", "telegram")
# result.failed == {"slack": "AlertDeliveryError"}

Raise when any transport fails

By default the dispatcher returns failures without raising. If your workflow should fail after all transports have been attempted, set raise_on_failure=True.

from alert_infra import Alert, AlertDispatcher

result = AlertDispatcher(transports, raise_on_failure=True).send(
    Alert(title="Deployment failed", message="Release job failed after migration step.")
)

If any transport fails, AlertDeliveryError is raised after the dispatcher attempts every transport.

Background jobs, CLIs, and scripts

Use one process-wide dispatcher factory so every script uses the same configuration.

# alerts.py
from functools import lru_cache

from alert_infra import AlertDispatcher
from alert_infra.apps import SlackWebhookTransport
from alert_infra.email import SMTPEmailTransport

@lru_cache(maxsize=1)
def get_dispatcher() -> AlertDispatcher:
    return AlertDispatcher([
        SMTPEmailTransport.from_env(),
        SlackWebhookTransport.from_env(),
    ])

# nightly_import.py
from alert_infra import Alert
from alerts import get_dispatcher

try:
    run_import()
except Exception as exc:
    get_dispatcher().send(
        Alert(
            title="Nightly import failed",
            message=str(exc.__class__.__name__),
            severity="critical",
            source="nightly-import",
            metadata={"job": "supplier_csv"},
        )
    )
    raise

Prefer storing exception class names or safe error summaries in alert metadata. Avoid sending raw exception messages if they may contain credentials or customer data.

Django usage

Django integration is available from alert_infra.django and reads configuration from settings.ALERT_INFRA.

Basic Django settings

# settings.py
ALERT_INFRA = {
    "ENABLED": True,
    "DEFAULT_SEVERITY": "error",
    "REDACT_SENSITIVE_DATA": True,
    "EMAIL": {
        "ENABLED": True,
        # BACKEND defaults to "auto": Resend when configured, then SendGrid, then SMTP.
        "FROM_EMAIL": env("ALERT_FROM_EMAIL"),
        "TO_EMAILS": env.list("ALERT_TO_EMAILS"),
        "RESEND_API_KEY": env("ALERT_RESEND_API_KEY", default=""),
        "SENDGRID_API_KEY": env("ALERT_SENDGRID_API_KEY", default=""),
        "SMTP_HOST": env("ALERT_SMTP_HOST", default=""),
    },
    "SLACK": {
        "ENABLED": True,
        "WEBHOOK_URL": env("ALERT_SLACK_WEBHOOK_URL"),
    },
    "TELEGRAM": {
        "ENABLED": False,
        "BOT_TOKEN": env("ALERT_TELEGRAM_BOT_TOKEN", default=""),
        "CHAT_ID": env("ALERT_TELEGRAM_CHAT_ID", default=""),
    },
}

If no transports are enabled, Django builds a dispatcher with NoOpTransport. If ENABLED is False, dispatching is skipped and returns an empty DeliveryResult.

Send an alert from a Django view

from django.http import JsonResponse
from alert_infra.django import send_alert


def update_invoice(request, invoice_id):
    invoice = get_invoice(invoice_id)

    if not invoice.can_transition_to("paid"):
        result = send_alert(
            title="Suspicious invoice update",
            message="Invoice status transition was blocked.",
            severity="warning",
            source="invoice-view",
            tags=("invoice", "security"),
            metadata={
                "invoice_id": invoice.id,
                "user_id": request.user.id,
                "authorization": request.headers.get("Authorization"),
            },
            request=request,
        )
        return JsonResponse({"ok": False, "alert_sent": result.ok}, status=409)

    invoice.mark_paid()
    return JsonResponse({"ok": True})

When request is provided, the Django helper attaches safe request metadata such as method, path, request ID, user ID, and selected sensitive headers after redaction.

Send an alert from a Django management command

from django.core.management.base import BaseCommand
from alert_infra.django import send_alert


class Command(BaseCommand):
    help = "Run supplier synchronization"

    def handle(self, *args, **options):
        try:
            synchronize_supplier_data()
        except Exception as exc:
            send_alert(
                title="Supplier sync failed",
                message=exc.__class__.__name__,
                severity="critical",
                source="management-command:supplier_sync",
                metadata={"command": "supplier_sync"},
            )
            raise

Send an alert from a Celery task in a Django project

alert_infra does not require Celery, but it can be called from any task once Django settings are loaded.

from celery import shared_task
from alert_infra.django import send_alert


@shared_task(bind=True)
def process_invoice(self, invoice_id):
    try:
        process(invoice_id)
    except Exception as exc:
        send_alert(
            title="Invoice task failed",
            message=exc.__class__.__name__,
            severity="error",
            source="celery:process_invoice",
            metadata={"invoice_id": invoice_id, "task_id": self.request.id},
        )
        raise

Django email provider selection

When EMAIL["BACKEND"] is omitted or set to "auto", alert_infra chooses the first configured provider in this order:

1. Resend when RESEND_API_KEY or ALERT_RESEND_API_KEY is present.
2. SendGrid when SENDGRID_API_KEY or ALERT_SENDGRID_API_KEY is present.
3. Direct SMTP as the fallback when neither provider API key is configured.

Use BACKEND="resend", BACKEND="sendgrid", BACKEND="smtp", or BACKEND="django" to force a provider.

ALERT_INFRA = {
    "EMAIL": {
        "ENABLED": True,
        "BACKEND": "auto",
        "FROM_EMAIL": "[e-mail]",
        "TO_EMAILS": ["[e-mail]"],
        "RESEND_API_KEY": env("ALERT_RESEND_API_KEY", default=""),
        "SENDGRID_API_KEY": env("ALERT_SENDGRID_API_KEY", default=""),
        "SMTP_HOST": env("ALERT_SMTP_HOST", default=""),
        "SMTP_PORT": env.int("ALERT_SMTP_PORT", default=587),
        "SMTP_USERNAME": env("ALERT_SMTP_USERNAME", default=""),
        "SMTP_PASSWORD": env("ALERT_SMTP_PASSWORD", default=""),
    }
}

Django email backend configuration

Set EMAIL["BACKEND"] to "django" when you want alert_infra to use Django's configured email backend through EmailMultiAlternatives.

# settings.py
EMAIL_BACKEND = "django.core.mail.backends.smtp.EmailBackend"
EMAIL_HOST = env("EMAIL_HOST")
EMAIL_PORT = env.int("EMAIL_PORT", default=587)
EMAIL_HOST_USER = env("EMAIL_HOST_USER")
EMAIL_HOST_PASSWORD = env("EMAIL_HOST_PASSWORD")
EMAIL_USE_TLS = True

ALERT_INFRA = {
    "EMAIL": {
        "ENABLED": True,
        "BACKEND": "django",
        "FROM_EMAIL": "[e-mail]",
        "TO_EMAILS": ["[e-mail]"],
    }
}

Django with direct SMTP transport

Use the direct SMTP transport by setting EMAIL["BACKEND"] to "smtp", or let BACKEND="auto" fall back to SMTP when neither Resend nor SendGrid API keys are configured.

ALERT_INFRA = {
    "EMAIL": {
        "ENABLED": True,
        "BACKEND": "smtp",
        "SMTP_HOST": env("ALERT_SMTP_HOST"),
        "SMTP_PORT": env.int("ALERT_SMTP_PORT", default=587),
        "SMTP_USERNAME": env("ALERT_SMTP_USERNAME"),
        "SMTP_PASSWORD": env("ALERT_SMTP_PASSWORD"),
        "SMTP_USE_TLS": True,
        "FROM_EMAIL": env("ALERT_FROM_EMAIL"),
        "TO_EMAILS": env.list("ALERT_TO_EMAILS"),
        "TIMEOUT": 8.0,
    }
}

Django email templates

Django email alerts can use project templates for the subject, text body, and HTML body.

ALERT_INFRA = {
    "EMAIL": {
        "ENABLED": True,
        "BACKEND": "django",
        "FROM_EMAIL": "[e-mail]",
        "TO_EMAILS": ["[e-mail]"],
        "SUBJECT_TEMPLATE": "alerts/email_subject.txt",
        "BODY_TEMPLATE": "alerts/email_body.txt",
        "HTML_TEMPLATE": "alerts/email_body.html",
        "TEMPLATE_CONTEXT": {"product_name": "Billing Portal"},
    }
}

Template context contains:

Name	Description
alert	The Alert object.
alert_dict	Dictionary returned by alert.to_dict().
metadata	Redacted alert metadata dictionary.
tags	List of alert tags.
Values from TEMPLATE_CONTEXT	Any static values configured in settings.

Example templates:

{# templates/alerts/email_subject.txt #}
{{ product_name }} {{ alert.severity|upper }}: {{ alert.title }}

{# templates/alerts/email_body.txt #}
{{ alert.message }}

Source: {{ alert.source|default:"unknown" }}
Correlation ID: {{ alert.correlation_id }}
{% if metadata.invoice_id %}Invoice: {{ metadata.invoice_id }}{% endif %}

{# templates/alerts/email_body.html #}
<h1>{{ alert.title }}</h1>
<p>{{ alert.message }}</p>
<ul>
  <li>Severity: {{ alert.severity }}</li>
  <li>Source: {{ alert.source|default:"unknown" }}</li>
  <li>Correlation ID: {{ alert.correlation_id }}</li>
</ul>

The subject renderer collapses line breaks so email subjects remain single-line.

Resend, SendGrid, and SMTP fallback

Resend and SendGrid are first-class alert email transports. In Django auto mode, configure RESEND_API_KEY to use Resend, configure SENDGRID_API_KEY to use SendGrid, or omit both API keys to fall back to direct SMTP. The SMTP fallback still requires SMTP_HOST, sender, and recipients.

All email transports support test-safe injection points: Resend and SendGrid accept an http_client, while SMTP accepts a sender callable. Use those mocks in tests to prevent real email delivery and accidental spam.

Django settings reference

ALERT_INFRA supports the following keys:

Key	Default	Description
ENABLED	True	Global switch. When False, dispatching is skipped.
DEFAULT_SEVERITY	"error"	Severity used by send_alert when no severity is provided.
REDACT_SENSITIVE_DATA	True	Whether Alert redacts sensitive metadata.
EMAIL	see below	Email transport settings.
SLACK	{"ENABLED": False}	Slack webhook settings.
TELEGRAM	{"ENABLED": False}	Telegram bot settings.

Email settings:

Key	Description
ENABLED	Enable email delivery.
BACKEND	"auto" (default), "resend", "sendgrid", "smtp", or "django". Auto prefers Resend, then SendGrid, then SMTP fallback.
FROM_EMAIL	Sender address. Falls back to ALERT_FROM_EMAIL.
TO_EMAILS	Recipient list or comma-separated string. Falls back to ALERT_TO_EMAILS.
TIMEOUT	Delivery timeout in seconds.
RESEND_API_KEY	Resend API key. Falls back to ALERT_RESEND_API_KEY.
RESEND_API_URL	Optional Resend API URL override.
SENDGRID_API_KEY	SendGrid API key. Falls back to ALERT_SENDGRID_API_KEY.
SENDGRID_API_URL	Optional SendGrid API URL override.
SMTP_HOST	SMTP hostname for direct SMTP mode or auto fallback. Falls back to ALERT_SMTP_HOST.
SMTP_PORT	SMTP port. Falls back to ALERT_SMTP_PORT or 587.
SMTP_USERNAME	SMTP username. Falls back to ALERT_SMTP_USERNAME.
SMTP_PASSWORD	SMTP password. Falls back to ALERT_SMTP_PASSWORD.
SMTP_USE_TLS	Whether direct SMTP mode starts TLS.
SUBJECT_TEMPLATE / SUBJECT_TEMPLATE_NAME	Optional Django subject template.
BODY_TEMPLATE / BODY_TEMPLATE_NAME	Optional Django text body template.
HTML_TEMPLATE / HTML_TEMPLATE_NAME	Optional Django HTML body template.
TEMPLATE_CONTEXT	Static values merged into email template context.

Slack settings:

Key	Description
ENABLED	Enable Slack delivery.
WEBHOOK_URL	Slack incoming webhook URL. Falls back to ALERT_SLACK_WEBHOOK_URL.
TIMEOUT	HTTP timeout in seconds.

Telegram settings:

Key	Description
ENABLED	Enable Telegram delivery.
BOT_TOKEN	Telegram bot token. Falls back to ALERT_TELEGRAM_BOT_TOKEN.
CHAT_ID	Telegram chat ID. Falls back to ALERT_TELEGRAM_CHAT_ID.
TIMEOUT	HTTP timeout in seconds.

Transport configuration

Environment variables

The built-in transports understand these environment variables:

Variable	Used by	Description
ALERT_FROM_EMAIL	SMTP and Django config	Sender email address.
ALERT_TO_EMAILS	SMTP and Django config	Comma-separated recipient list.
ALERT_RESEND_API_KEY	Resend	Resend API key for Resend or auto email delivery.
ALERT_SENDGRID_API_KEY	SendGrid	SendGrid API key for SendGrid or auto email delivery.
ALERT_SMTP_HOST	SMTP	SMTP host. Required when auto email delivery falls back to SMTP.
ALERT_SMTP_PORT	SMTP	SMTP port. Defaults to 587.
ALERT_SMTP_USERNAME	SMTP	SMTP username.
ALERT_SMTP_PASSWORD	SMTP	SMTP password.
ALERT_SMTP_USE_TLS	SMTP	true, false, 1, 0, yes, or no. Defaults to enabled.
ALERT_SLACK_WEBHOOK_URL	Slack	Slack incoming webhook URL. Must be https://.
ALERT_TELEGRAM_BOT_TOKEN	Telegram	Bot token.
ALERT_TELEGRAM_CHAT_ID	Telegram	Destination chat ID.

Built-in transport names

Transport names appear in DeliveryResult.sent and DeliveryResult.failed.

Transport	Name
NoOpTransport	noop
SMTPEmailTransport	email.smtp
DjangoEmailTransport	email.django
SlackWebhookTransport	slack
TelegramBotTransport	telegram

Security and redaction

Alert redacts sensitive metadata by default before any transport receives it.

Sensitive key matching is case-insensitive and treats hyphens as underscores. A key is redacted when it contains one of these terms:

• password
• token
• secret
• api_key
• authorization
• cookie
• session
• csrf
• access
• refresh
• private_key

Example:

from alert_infra import Alert, REDACTED

alert = Alert(
    title="Authentication anomaly",
    message="Unexpected login attempt.",
    metadata={
        "user_id": 7,
        "authorization": "Bearer secret",
        "nested": {"refresh_token": "secret"},
    },
)

assert alert.metadata["authorization"] == REDACTED
assert alert.metadata["nested"]["refresh_token"] == REDACTED

Nested dictionaries, lists, tuples, sets, dataclasses, and many sequence values are traversed. Unsupported values are converted to strings to keep transport payloads serializable.

Only disable redaction when you have a controlled internal transport and have reviewed the data classification risk:

Alert(
    title="Internal debug alert",
    message="Redaction disabled for a controlled test only.",
    metadata={"token": "visible"},
    redact_sensitive_data=False,
)

Recommended security practices:

• Load credentials from environment variables, secret managers, or Django settings.
• Do not hardcode webhook URLs, bot tokens, SMTP passwords, or API keys in source control.
• Do not put raw request bodies, raw exception strings, cookies, or authorization headers in metadata.
• Prefer stable identifiers such as invoice IDs, user IDs, request IDs, and job IDs.
• Keep dispatcher logs free of raw alert metadata. The built-in dispatcher logs transport names and exception class names only.

Error handling and delivery results

The project defines these exceptions:

Exception	Raised when
AlertValidationError	An alert is invalid, such as missing title/message or unsupported severity.
AlertConfigurationError	A transport is configured unsafely or incompletely.
AlertDeliveryError	A transport fails while delivering an alert.

Dispatcher behavior:

result = dispatcher.send(alert)

if result.failed:
    # Example: {"slack": "AlertDeliveryError"}
    logger.warning("Some alert transports failed: %s", result.failed)

The dispatcher attempts all transports even if one fails. Set raise_on_failure=True to raise AlertDeliveryError after all transports are attempted.

Custom transports

A transport is any object with:

• a name attribute, and
• a send(alert: Alert) -> None method.

from alert_infra import Alert, AlertDispatcher
from alert_infra.exceptions import AlertDeliveryError


class PagerDutyTransport:
    name = "pagerduty"

    def __init__(self, client, service_id: str) -> None:
        self.client = client
        self.service_id = service_id

    def send(self, alert: Alert) -> None:
        payload = {
            "summary": f"[{alert.severity.upper()}] {alert.title}",
            "source": alert.source or "unknown",
            "severity": alert.severity,
            "custom_details": alert.to_dict(),
        }
        try:
            self.client.trigger_incident(self.service_id, payload)
        except Exception as exc:
            raise AlertDeliveryError("PagerDuty alert delivery failed") from exc


dispatcher = AlertDispatcher([PagerDutyTransport(client, "svc_123")])
dispatcher.send(Alert(title="Queue backlog high", message="More than 10,000 jobs are pending."))

For webhook-style custom transports, prefer injecting an HTTP client into the transport. This makes unit tests deterministic and avoids real network calls.

Testing patterns

Test alert dispatch without external services

from alert_infra import Alert, AlertDispatcher


class RecordingTransport:
    name = "recording"

    def __init__(self):
        self.alerts = []

    def send(self, alert):
        self.alerts.append(alert)


def test_dispatches_alert():
    transport = RecordingTransport()
    result = AlertDispatcher([transport]).send(Alert(title="Test", message="Body"))

    assert result.ok
    assert result.sent == ("recording",)
    assert transport.alerts[0].title == "Test"

Test Resend or SendGrid without sending email

from alert_infra import Alert
from alert_infra.email import ResendEmailTransport


class MockHttpClient:
    def __init__(self):
        self.calls = []

    def post(self, url, *, json, headers=None, timeout):
        self.calls.append({"url": url, "json": json, "headers": headers, "timeout": timeout})
        return 200


def test_resend_payload():
    client = MockHttpClient()
    transport = ResendEmailTransport(
        api_key="test-key",
        from_email="[e-mail]",
        to_emails=["[e-mail]"],
        http_client=client,
    )

    transport.send(Alert(title="Payment failed", message="Failed", metadata={"api_key": "secret"}))

    assert client.calls[0]["headers"]["Authorization"] == "Bearer test-key"
    assert "secret" not in str(client.calls[0]["json"])

Use the same http_client pattern with SendGridEmailTransport.

Test SMTP without connecting to an SMTP server

from alert_infra import Alert
from alert_infra.email import SMTPEmailTransport


def test_email_body_redacts_secret():
    calls = []

    def sender(recipients, subject, body, html):
        calls.append((recipients, subject, body, html))

    transport = SMTPEmailTransport(
        host="smtp.example.com",
        from_email="[e-mail]",
        to_emails=["[e-mail]"],
        sender=sender,
    )

    transport.send(Alert(title="Payment failed", message="Failed", metadata={"api_key": "secret"}))

    assert "secret" not in calls[0][2]
    assert "[REDACTED]" in calls[0][2]

Test webhook transports without network access

from alert_infra import Alert
from alert_infra.apps import SlackWebhookTransport


class MockHttpClient:
    def __init__(self):
        self.calls = []

    def post(self, url, *, json, headers=None, timeout):
        self.calls.append({"url": url, "json": json, "headers": headers, "timeout": timeout})
        return 200


def test_slack_payload():
    client = MockHttpClient()
    transport = SlackWebhookTransport("https://hooks.slack.com/services/test", http_client=client)

    transport.send(Alert(title="Test", message="Body", metadata={"authorization": "secret"}))

    assert client.calls[0]["json"]["metadata"]["metadata"]["authorization"] == "[REDACTED]"

Test Django integration

Use Django's in-memory email backend or locmem template loader in tests.

from django.test import override_settings
from alert_infra.django import send_alert


def test_send_alert_disabled_mode():
    with override_settings(ALERT_INFRA={"ENABLED": False, "REDACT_SENSITIVE_DATA": True}):
        result = send_alert(title="Test", message="Body", metadata={"token": "secret"})

    assert result.sent == ()
    assert result.failed == {}

Compatibility namespace

Projects that previously imported from feature_flag_infra can continue to use that namespace:

from feature_flag_infra import Alert, AlertDispatcher
from feature_flag_infra.django import send_alert

Django settings may also be supplied as FEATURE_FLAG_INFRA when ALERT_INFRA is not defined. Prefer ALERT_INFRA for new projects.

API reference

Public imports from alert_infra

from alert_infra import (
    Alert,
    AlertConfigurationError,
    AlertDeliveryError,
    AlertDispatcher,
    AlertInfraError,
    AlertTransport,
    AlertValidationError,
    DeliveryResult,
    NoOpTransport,
    REDACTED,
    SENSITIVE_KEYWORDS,
    VALID_SEVERITIES,
    redact_metadata,
)

Alert

Alert(
    title: str,
    message: str,
    severity: str = "error",
    source: str | None = None,
    tags: tuple[str, ...] = (),
    metadata: Mapping[str, Any] = {},
    created_at: datetime = <current UTC time>,
    correlation_id: str | None = None,
    request_id: str | None = None,
    redact_sensitive_data: bool = True,
)

Methods:

• to_dict() -> dict[str, Any]: returns a transport-friendly dictionary with ISO-formatted created_at.

AlertDispatcher

AlertDispatcher(
    transports: Sequence[AlertTransport] | None = None,
    *,
    enabled: bool = True,
    raise_on_failure: bool = False,
    logger_: logging.Logger | None = None,
)

Methods:

• send(alert: Alert) -> DeliveryResult

Email API

from alert_infra.email import (
    ResendEmailTransport,
    SendGridEmailTransport,
    SMTPEmailTransport,
    format_alert_body,
    format_alert_subject,
)

ResendEmailTransport.from_env(prefix="ALERT_RESEND_") reads Resend API settings, SendGridEmailTransport.from_env(prefix="ALERT_SENDGRID_") reads SendGrid API settings, and SMTPEmailTransport.from_env(prefix="ALERT_SMTP_") reads SMTP-related environment variables.

App/webhook API

from alert_infra.apps import SlackWebhookTransport, TelegramBotTransport

• SlackWebhookTransport.from_env(env_var="ALERT_SLACK_WEBHOOK_URL")
• TelegramBotTransport.from_env(token_env="ALERT_TELEGRAM_BOT_TOKEN", chat_env="ALERT_TELEGRAM_CHAT_ID")

Django API

from alert_infra.django import (
    DjangoEmailTransport,
    build_dispatcher,
    get_alert_infra_settings,
    request_metadata,
    send_alert,
)

send_alert signature:

send_alert(
    *,
    title: str,
    message: str,
    severity: str | None = None,
    source: str | None = None,
    tags: tuple[str, ...] | list[str] = (),
    metadata: Mapping[str, Any] | None = None,
    request: Any | None = None,
) -> DeliveryResult

Development

Run the test suite:

pytest

Build the package:

python -m build

The project uses setuptools and setuptools-scm for packaging and dynamic versioning.

Async alerting with Celery

alert_infra can enqueue outbound alert delivery to Celery while keeping the core package framework-independent. Celery is an optional extra; importing alert_infra or using the synchronous dispatcher does not import Celery.

Install Celery support when you want queued dispatch:

pip install "alert-infra[celery]"

The async layer serializes an Alert with Alert.to_dict(), redacts sensitive metadata before queueing, and rehydrates it with Alert.from_dict() inside the worker. Transport credentials remain in settings/environment on the worker; they are not passed as task arguments.

Django Celery setup

Add the task module to your normal Celery autodiscovery path. If your project autodiscovers installed apps, include alert_infra.django in INSTALLED_APPS; otherwise import alert_infra.django.tasks from your Celery app module.

# config/celery.py
import os
from celery import Celery

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings")

app = Celery("config")
app.config_from_object("django.conf:settings", namespace="CELERY")
app.autodiscover_tasks()

# settings.py
INSTALLED_APPS = [
    # ...
    "alert_infra.django",
]

ALERT_INFRA = {
    "ENABLED": True,
    "ASYNC": {
        "ENABLED": True,
        "BACKEND": "celery",
        "TASK_NAME": "alert_infra.dispatch_alert",
        "QUEUE": "alerts",
        "MAX_RETRIES": 3,
        "RETRY_BACKOFF": True,
        "RETRY_BACKOFF_MAX": 300,
        "RETRY_JITTER": True,
        "FAIL_SILENTLY": True,
    },
    "EMAIL": {
        "ENABLED": True,
        # BACKEND defaults to "auto": Resend when configured, then SendGrid, then SMTP.
        "FROM_EMAIL": env("ALERT_FROM_EMAIL"),
        "TO_EMAILS": env.list("ALERT_TO_EMAILS"),
        "RESEND_API_KEY": env("ALERT_RESEND_API_KEY", default=""),
        "SENDGRID_API_KEY": env("ALERT_SENDGRID_API_KEY", default=""),
        "SMTP_HOST": env("ALERT_SMTP_HOST", default=""),
    },
    "SLACK": {
        "ENABLED": True,
        "WEBHOOK_URL": env("ALERT_SLACK_WEBHOOK_URL"),
    },
    "TELEGRAM": {
        "ENABLED": True,
        "BOT_TOKEN": env("ALERT_TELEGRAM_BOT_TOKEN"),
        "CHAT_ID": env("ALERT_TELEGRAM_CHAT_ID"),
    },
}

CELERY_TASK_ROUTES = {
    "alert_infra.dispatch_alert": {"queue": "alerts"},
}

Run a worker for the alerts queue:

celery -A config worker -Q alerts -l info

When ALERT_INFRA["ASYNC"]["ENABLED"] is true, alert_infra.django.send_alert(...) enqueues alert_infra.dispatch_alert and returns a DeliveryResult with sent=("celery",). When async is disabled, it uses the synchronous AlertDispatcher exactly as before. If Celery is unavailable or not configured, FAIL_SILENTLY=True returns a failed DeliveryResult instead of crashing the request; set FAIL_SILENTLY=False to raise a clear configuration error.

Async settings reference

Key	Default	Description
ENABLED	False	Enable Celery-backed alert dispatch from the Django helper.
BACKEND	"celery"	Async backend. Currently only Celery is supported.
TASK_NAME	"alert_infra.dispatch_alert"	Celery task name to enqueue.
QUEUE	"alerts"	Queue passed to send_task/apply_async.
MAX_RETRIES	3	Maximum worker retries for retryable transport failures.
RETRY_BACKOFF	True	Use bounded exponential retry countdowns, or an integer base delay.
RETRY_BACKOFF_MAX	300	Maximum retry countdown in seconds.
RETRY_JITTER	True	Randomize retry countdowns to avoid thundering herds.
FAIL_SILENTLY	True	Do not crash web requests when enqueueing fails.

Retry and partial-failure behavior

Celery retries are explicit so successful transports are not resent unnecessarily. The task dispatches to all configured transports on the first attempt. If, for example, email succeeds and Slack has a retryable timeout, the retry is scheduled with transport_names=["slack"]; email is omitted from the retry. Non-retryable configuration/authentication errors are reported in the task result and are not retried.

Retryable examples include network errors, timeouts, SMTP connection/disconnection failures, SMTP 4xx temporary data errors, and webhook 5xx responses. Non-retryable examples include missing SMTP settings, invalid Slack webhook URLs, invalid Telegram chat IDs/tokens, invalid recipients, authentication/configuration failures, unsupported severities, and webhook 4xx responses.

Task logs and dispatcher logs contain transport names and exception class names only. They do not include webhook URLs, bot tokens, SMTP passwords, raw authorization headers, cookies, or metadata values.

Plain Python synchronous usage

Plain Python applications can continue using the synchronous dispatcher without Celery:

from alert_infra import Alert, AlertDispatcher
from alert_infra.apps import SlackWebhookTransport

sync_dispatcher = AlertDispatcher([SlackWebhookTransport.from_env()])
sync_dispatcher.send(Alert(title="API outage", message="Health checks failed."))

Plain Python Celery usage

Plain Python applications can opt into Celery by passing either a Celery app or a task-like object with apply_async to CeleryAlertDispatcher:

from celery import Celery
from alert_infra import Alert
from alert_infra.celery import CeleryAlertDispatcher

app = Celery("alerts")
async_dispatcher = CeleryAlertDispatcher(
    celery_app=app,
    config={
        "TASK_NAME": "alert_infra.dispatch_alert",
        "QUEUE": "alerts",
        "MAX_RETRIES": 3,
        "FAIL_SILENTLY": False,
    },
)

async_dispatcher.send(Alert(title="Import failed", message="Supplier import exited with status 1."))

For non-Django plain Python workers, define your own Celery task that receives the serialized payload, calls Alert.from_dict(payload), and dispatches with your application-specific AlertDispatcher.

Testing async alerting

Use the repository test suite to exercise serialization, Django settings loading, async enqueueing, task rehydration, retry behavior, partial failures, and transport error classification:

python -m pytest