pynotifyhub 0.1.0

Unified multi-channel notification library: Slack, Discord, Telegram, WeCom, DingTalk, Lark/Feishu, and voice calls with level-based routing.

notify-hub

Unified multi-channel notification library for Python.

Send alerts to Slack, Discord, Telegram, WeCom (企业微信), DingTalk (钉钉), Lark/Feishu (飞书), and phone voice calls (Twilio) through one API, with level-based routing configured per application — so you never write notification glue code again.

from notify_hub import Hub

hub = Hub.from_config("notify-hub.toml")
hub.notify("database is down", level="CRITICAL")   # returns in microseconds

What that one call does is decided entirely by config: which channels fire for which level, how attachments degrade per channel, retries, dedup, rate limits, and where the audit log goes.

📖 中文使用说明（30 秒跑通 + 全渠道接入指南）：docs/usage.md

Features

• 7 channels, one interface — Slack, Discord, Telegram, WeCom, DingTalk, Lark/Feishu, phone voice calls (Twilio, pluggable for other providers).
• Level-based routing — builtin DEBUG..CRITICAL plus your own custom levels (TRADE_HALT = 45); each app maps levels to channels in config, matching rules are unioned.
• Never blocks your app — notify() returns immediately in both sync and asyncio hosts; fan-out runs concurrently on a lazily-started background event loop. Await the handle (async) or wait() on it (sync) only when you need the outcome.
• Failure isolation & retries — one channel failing never affects the others and never raises into your code; transient errors (429/5xx, network) retry with exponential backoff + jitter.
• Attachments with graceful degradation — images/files from a path, bytes, or URL; channels declare capabilities and unsupported content is dropped / linked / failed per your policy (phone calls read text via TTS).
• Anti-storm guards — fingerprint dedup window (default 60 s) and per-channel token-bucket rate limits, so a crash loop can't ring your phone 1000 times.
• Audit log — every send attempt recorded (per-channel status, attempts, latency, errors) as JSON Lines or text, to a file and/or your own logging.Logger.
• A well-behaved library — stdlib dataclasses, one hard dependency (httpx), zero threads until first use, nothing written or configured unless you ask.

Installation

pip install pynotifyhub

Requires Python 3.10+. The distribution is named pynotifyhub (PyPI rejects notify-hub as too similar to an existing project); the import stays notify_hub.

Quickstart

1. Configure once (TOML or a dict)

# notify-hub.toml
[hub]
default_channels = ["slack-ops"]

[channels.slack-ops]
type = "slack"
webhook_url = "${SLACK_WEBHOOK_URL}"      # expanded from the environment

[channels.oncall-phone]
type = "phone"
provider = "twilio"
account_sid = "${TWILIO_ACCOUNT_SID}"
auth_token = "${TWILIO_AUTH_TOKEN}"
from_number = "[phone]"
to_numbers = ["[phone]"]
rate_limit = { per_minute = 2, burst = 1 }

[[routes]]
min_level = "WARNING"
channels = ["slack-ops"]

[[routes]]
levels = ["CRITICAL"]
channels = ["oncall-phone"]               # CRITICAL also matches the rule above

See examples/notify-hub.example.toml for every channel and option.

2. Send

from notify_hub import Attachment, Hub

with Hub.from_config("notify-hub.toml") as hub:
    hub.warning("disk at 85%")                          # fire-and-forget
    hub.critical("db down")                             # slack + phone call

    handle = hub.notify(
        "deploy finished",
        level="INFO",
        title="Release v2.1",
        attachments=[Attachment.image("https://example.com/chart.png")],
    )
    result = handle.wait(timeout=30)                    # only if you need it
    print(result.ok)

Async hosts use the same hub:

async with Hub.from_config("notify-hub.toml") as hub:
    hub.error("worker crashed")                  # still instant
    result = await hub.notify_async("done")      # or await the outcome

Custom levels:

hub.levels.register("TRADE_HALT", 45)            # or [levels.custom] in TOML
hub.notify("strategy halted", level="TRADE_HALT")

3. Observe

With [log] enabled, every attempt is recorded:

{"ts": "2026-06-11T12:00:00+00:00", "level": "CRITICAL", "fingerprint": "9f3a...",
 "preview": "db down", "suppressed_by_dedup": false,
 "channels": [{"id": "slack-ops", "ok": true, "attempts": 1, "latency_ms": 212.4,
               "error": null, "degraded": [], "skipped_reason": null}]}

Channel capability matrix

Channel	Text	Markdown	Image	File	Notes
Slack	✅	✅ (mrkdwn)	✅*	✅*	*URL images free via blocks; uploads need bot_token + file_channel
Discord	✅	✅	✅	✅	multipart, ≤10 files
Telegram	✅	✅ (HTML)	✅	✅	single image + short text sent as captioned photo
WeCom 企业微信	✅	✅	✅	✅	image ≤2 MB jpg/png; file via upload_media
DingTalk 钉钉	✅	✅	—	—	HMAC signing or keyword prefix
Lark/Feishu 飞书	✅	✅ (post)	✅*	—	*images need app_id+app_secret (custom app); both feishu.cn and larksuite.com
Phone (Twilio)	✅ (TTS)	—	—	—	text stripped of markdown/URLs, read aloud

Unsupported content degrades per [degrade] policy — by default it is dropped quietly and noted in the result, never failing the notification.

Writing your own channel

from notify_hub import Capability, Channel, Message, register_channel

@register_channel
class MyChannel(Channel):
    name = "mychannel"
    capabilities = Capability.TEXT

    async def send(self, message: Message) -> None:
        ...  # raise ChannelSendError on failure; hub handles retry/logging

Or publish it as an entry point in the notify_hub.channels group. Voice providers (e.g. Vonage) subclass VoiceProvider the same way.

Design notes

• The TOML schema and routing semantics are language-neutral by design — they are the contract for planned Go and Rust ports.
• v2 reserves escalation (ack + auto-escalate chains) and digest (aggregation windows) fields in routing rules; they parse today and warn, but do nothing yet.

Development

pip install -e ".[dev]"
pytest                 # tests
ruff check . && ruff format --check .
mypy

License

MIT