ddp_utils.notifications¶

import ddp_utils.notifications

Provide fail-safe email, desktop, mobile, and Slack notifications.

Each channel is optional and independent. Delivery failures are converted to False results so notification problems do not terminate business logic.

Examples

Configure channels and send a notification:

from ddp_utils.notifications import Notifier

notifier = Notifier()
notifier.configure_telegram(token="123:ABC", chat_id="987654321")
notifier.configure_ntfy(topic="my-project-alerts")
results = notifier.notify("VPN connected", "Server: US-West")

class ddp_utils.notifications.EmailConfig(smtp_host: str = '', smtp_port: int = 587, smtp_user: str = '', smtp_password: str = '', from_addr: str = '', to: List[str] = <factory>, use_tls: bool = True)¶

Bases: object

SMTP settings for sending email notifications.

Examples

Use this public operation:

instance = EmailConfig(...)

class ddp_utils.notifications.TelegramConfig(token: str = '', chat_id: str = '', parse_mode: str = 'HTML')¶

Bases: object

Bot token and chat settings for sending Telegram notifications.

Examples

Use this public operation:

instance = TelegramConfig(...)

class ddp_utils.notifications.NtfyConfig(topic: str = '', server: str = 'https://ntfy.sh', priority: str = 'default', tags: List[str] = <factory>, allow_insecure_http: bool = False)¶

Bases: object

Topic and server settings for sending ntfy.sh push notifications.

Examples

Use this public operation:

instance = NtfyConfig(...)

class ddp_utils.notifications.PushoverConfig(app_token: str = '', user_key: str = '')¶

Bases: object

App and user key settings for sending Pushover notifications.

Examples

Use this public operation:

instance = PushoverConfig(...)

class ddp_utils.notifications.DesktopConfig(app_name: str = 'ddp_utils', timeout: int = 10)¶

Bases: object

App name and timeout settings for native desktop notifications.

Examples

Use this public operation:

instance = DesktopConfig(...)

class ddp_utils.notifications.Notifier(*, async_send: bool = False)¶

Bases: object

Send fail-safe notifications through configured channels.

A channel failure is reported as False and does not stop the other channels. With async_send=True, notify() starts one daemon thread per channel and waits up to 30 seconds for each thread.

Parameters

Name	Type	Description
async_send	bool	Whether multi-channel sends should use daemon threads.

Examples

Configure two channels and send through both:

notifier = Notifier()
notifier.configure_telegram(token="token", chat_id="123")
notifier.configure_ntfy(topic="my-project")
results = notifier.notify("Deploy complete", "Version 2.5.1")

Initialize an empty notification facade.

Parameters

Name	Type	Description
async_send	bool	Whether multi-channel sends should use daemon threads.

Examples

Create a notifier that dispatches channels concurrently:

notifier = Notifier(async_send=True)

configure_email(*, smtp_host: str, smtp_user: str, smtp_password: str, to: str | List[str], from_addr: str = '', smtp_port: int = 587, use_tls: bool = True) → Notifier¶

Configure the SMTP email channel.

Parameters

Name	Type	Description
smtp_host	str	SMTP server hostname.
smtp_user	str	Username used to authenticate with the SMTP server.
smtp_password	str	Password used to authenticate with the SMTP server.
to	str \| List[str]	One recipient address or a list of recipient addresses.
from_addr	str	Sender address. Defaults to `smtp_user` when empty.
smtp_port	int	SMTP server port.
use_tls	bool	Whether the email notifier should use TLS.

Returns

Type	Description
Notifier	This notifier, allowing configuration calls to be chained.

Examples

Configure a TLS-enabled SMTP channel:

notifier.configure_email(
    smtp_host="smtp.example.com",
    smtp_user="bot@example.com",
    smtp_password="secret",
    to=["ops@example.com"],
)

configure_desktop(*, app_name: str = 'ddp_utils', timeout: int = 10) → Notifier¶

Configure operating-system desktop notifications.

Parameters

Name	Type	Description
app_name	str	Application name displayed by the operating system.
timeout	int	Requested notification display time in seconds.

Returns

Type	Description
Notifier	This notifier, allowing configuration calls to be chained.

Examples

Enable desktop notifications for an application:

notifier.configure_desktop(app_name="Data Importer", timeout=8)

configure_telegram(*, token: str, chat_id: str, parse_mode: str = 'HTML') → Notifier¶

Configure a Telegram bot notification channel.

Parameters

Name	Type	Description
token	str	Telegram bot token.
chat_id	str	Destination chat identifier.
parse_mode	str	Telegram message parse mode.

Returns

Type	Description
Notifier	This notifier, allowing configuration calls to be chained.

Examples

Configure HTML-formatted Telegram messages:

notifier.configure_telegram(token="token", chat_id="123")

configure_ntfy(*, topic: str, server: str = 'https://ntfy.sh', priority: str = 'default', tags: List[str] | None = None, allow_insecure_http: bool = False) → Notifier¶

Configure an ntfy push-notification channel.

Parameters

Name	Type	Description
topic	str	Destination ntfy topic.
server	str	Base URL of the public or self-hosted ntfy server.
priority	str	ntfy priority such as `max`, `high`, `default`, `low`, or `min`.
tags	List[str] \| None	Optional ntfy tags included in the request headers.
allow_insecure_http	bool	Whether to permit plain HTTP for a self-hosted server. HTTPS remains required when this is `False`.

Returns

Type	Description
Notifier	This notifier, allowing configuration calls to be chained.

Examples

Configure a topic on the public ntfy service:

notifier.configure_ntfy(
    topic="project-alerts",
    priority="high",
    tags=["warning"],
)

configure_pushover(*, app_token: str, user_key: str) → Notifier¶

Configure a Pushover notification channel.

Parameters

Name	Type	Description
app_token	str	Pushover application API token.
user_key	str	Pushover destination user or group key.

Returns

Type	Description
Notifier	This notifier, allowing configuration calls to be chained.

Examples

Configure Pushover delivery:

notifier.configure_pushover(
    app_token="application-token",
    user_key="user-key",
)

set_logger(logger: Any) → Notifier¶

Attach a logger used for channel error reporting.

Parameters

Name	Type	Description
logger	Any	Object exposing an `error(message)` method.

Returns

Type	Description
Notifier	This notifier, allowing configuration calls to be chained.

Examples

Route suppressed channel exceptions to a standard logger:

notifier.set_logger(logging.getLogger("notifications"))

notify(title: str, message: str = '', *, html: bool = False) → Dict[str, bool]¶

Send through every configured notification channel.

Parameters

Name	Type	Description
title	str	Notification title or email subject.
message	str	Notification body.
html	bool	Whether the email body contains HTML. Other channels ignore this flag.

Returns

Type	Description
Dict[str, bool]	A mapping from each configured channel name to its delivery status. A channel that is not configured is omitted.

Examples

Inspect the status of every attempted channel:

results = notifier.notify("Import complete", "42 rows loaded")
email_ok = results.get("email", False)

email(subject: str, body: str = '', *, html: bool = False) → bool¶

Send through the configured email channel only.

Parameters

Name	Type	Description
subject	str	Email subject.
body	str	Email body.
html	bool	Whether `body` contains HTML.

Returns

Type	Description
bool	`True` when the channel reports success; `False` when the channel is missing or delivery fails.

Examples

Send an HTML report:

sent = notifier.email("Daily report", "<b>Ready</b>", html=True)

desktop(title: str, message: str = '') → bool¶

Show a notification through the configured desktop channel.

Parameters

Name	Type	Description
title	str	Notification title.
message	str	Notification body.

Returns

Type	Description
bool	`True` when the channel reports success; `False` when the channel is missing or delivery fails.

Examples

Show a local completion message:

shown = notifier.desktop("Complete", "Parsing finished")

phone(title: str, message: str = '') → Dict[str, bool]¶

Send through all configured mobile notification channels.

Parameters

Name	Type	Description
title	str	Notification title.
message	str	Notification body.

Returns

Type	Description
Dict[str, bool]	A status mapping for configured Telegram, ntfy, and Pushover channels. Unconfigured channels are omitted.

Examples

Send an alert to every configured mobile channel:

results = notifier.phone("Alert", "CPU usage is high")

slack_webhook(webhook_url: str, text: str, *, title: str = '') → bool¶

Send a message through a Slack Incoming Webhook.

Parameters

Name	Type	Description
webhook_url	str	HTTPS Slack Incoming Webhook URL.
text	str	Message text. Slack markdown is preserved.
title	str	Optional title prepended in bold markdown.

Returns

Type	Description
bool	`True` when Slack accepts the request; otherwise `False`.

Examples

Send a titled deployment message:

sent = notifier.slack_webhook(
    "https://hooks.slack.com/services/...",
    "Version 2.5.1 is live",
    title="Deploy complete",
)

ddp_utils.notifications.get_notifier() → Notifier¶

Return the process-wide notifier, creating it on first access.

Returns

Type	Description
Notifier	The current process-wide `Notifier` instance.

Examples

Configure the lazily created global notifier:

notifier = get_notifier()
notifier.configure_desktop()

ddp_utils.notifications.configure_notifier(notifier: Notifier) → None¶

Replace the process-wide notifier.

Parameters

Name	Type	Description
notifier	Notifier	Fully or partially configured notifier to expose globally.

Examples

Install a preconfigured asynchronous notifier:

configure_notifier(Notifier(async_send=True))

ddp_utils.notifications.notify(title: str, message: str = '', *, html: bool = False) → Dict[str, bool]¶

Send through every channel configured on the global notifier.

Parameters

Name	Type	Description
title	str	Notification title or email subject.
message	str	Notification body.
html	bool	Whether the email body contains HTML.

Returns

Type	Description
Dict[str, bool]	A mapping from each configured channel name to its delivery status.

Examples

Configure once and notify from any module:

get_notifier().configure_ntfy(topic="my-app")
results = notify("Error", "Worker crashed")