Skip to content

API reference

pynotify.app

Textual user interface for posting messages to Teams webhooks.

Channel(name, webhook_env_var=None, webhook_url=None) dataclass

A channel name and one local source for its webhook URL.

ChannelConfigurationError

Bases: ValueError

Raised when the channel configuration cannot be used.

NotifyApp(channels)

Bases: App[None]

Select channels, compose a Teams card, and send it.

Source code in src/pynotify/app.py
123
124
125
def __init__(self, channels: list[Channel]) -> None:
    super().__init__()
    self.channels = {channel.name: channel for channel in channels}

build_adaptive_card(heading, message)

Build the Adaptive Card payload accepted by Teams workflows.

Source code in src/pynotify/app.py
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def build_adaptive_card(heading: str, message: str) -> dict[str, object]:
    """Build the Adaptive Card payload accepted by Teams workflows."""
    return {
        "type": "AdaptiveCard",
        "$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
        "version": "1.4",
        "body": [
            {
                "type": "TextBlock",
                "size": "Small",
                "weight": "Bolder",
                "color": "Accent",
                "text": heading,
                "wrap": True,
            },
            {"type": "TextBlock", "text": message, "wrap": True},
        ],
    }

default_channels_path()

Return the user-level default rather than a file inside the project.

Source code in src/pynotify/app.py
33
34
35
36
def default_channels_path() -> Path:
    """Return the user-level default rather than a file inside the project."""
    config_home = Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config"))
    return config_home / "pynotify" / "channels.csv"

load_channels(path)

Load channels using either an environment variable or local webhook URL.

Source code in src/pynotify/app.py
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def load_channels(path: Path) -> list[Channel]:
    """Load channels using either an environment variable or local webhook URL."""
    try:
        with path.open(newline="", encoding="utf-8") as csv_file:
            reader = csv.DictReader(csv_file)
            if reader.fieldnames is None or "channel_name" not in reader.fieldnames:
                raise ChannelConfigurationError(
                    f"{path} must have a channel_name column and either webhook_env_var or webhook_url."
                )
            has_env_var = "webhook_env_var" in reader.fieldnames
            has_url = "webhook_url" in reader.fieldnames
            if has_env_var == has_url:
                raise ChannelConfigurationError(
                    f"{path} must include exactly one of webhook_env_var or webhook_url."
                )
            channels = []
            for row in reader:
                name = row["channel_name"].strip()
                source = row["webhook_env_var" if has_env_var else "webhook_url"].strip()
                if not name or not source:
                    continue
                channels.append(
                    Channel(
                        name,
                        webhook_env_var=source if has_env_var else None,
                        webhook_url=source if has_url else None,
                    )
                )
    except OSError as error:
        raise ChannelConfigurationError(f"Could not read channel configuration {path}: {error}") from error

    if not channels:
        raise ChannelConfigurationError(f"No channels were found in {path}.")
    if len({channel.name.casefold() for channel in channels}) != len(channels):
        raise ChannelConfigurationError(f"Channel names in {path} must be unique.")
    return channels

main()

Run the application or print a clear configuration error.

Source code in src/pynotify/app.py
202
203
204
205
206
207
208
209
def main() -> None:
    """Run the application or print a clear configuration error."""
    args = parse_args()
    try:
        channels = load_channels(args.channels)
    except ChannelConfigurationError as error:
        raise SystemExit(error) from error
    NotifyApp(channels).run()

post_message(webhook_url, heading, message)

Post one card, raising a descriptive error on delivery failure.

Source code in src/pynotify/app.py
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def post_message(webhook_url: str, heading: str, message: str) -> None:
    """Post one card, raising a descriptive error on delivery failure."""
    payload = json.dumps(build_adaptive_card(heading, message)).encode()
    request = Request(webhook_url, data=payload, headers={"Content-Type": "application/json"}, method="POST")
    try:
        with urlopen(request, timeout=30) as response:
            if response.status not in (200, 202):
                raise RuntimeError(f"Webhook returned HTTP {response.status}.")
    except HTTPError as error:
        detail = error.read().decode("utf-8", errors="replace").strip()
        raise RuntimeError(f"Webhook returned HTTP {error.code}: {detail}") from error
    except URLError as error:
        raise RuntimeError(f"Could not reach webhook: {error.reason}") from error