From 7cc831fca8a00e2b2a7cbe441057fc358ce33cc9 Mon Sep 17 00:00:00 2001 From: Jan Rummel Date: Sun, 8 Mar 2026 23:31:15 +0100 Subject: [PATCH] =?UTF-8?q?Add=20Scheduled=20Notifications=20example=20(tr?= =?UTF-8?q?igger=20=E2=86=92=20api=20=E2=86=92=20logic=20=E2=86=92=20deliv?= =?UTF-8?q?er)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fourth runnable example: notification scheduling with rule-based filtering, urgency determination, and digest delivery. 5 schedule presets, 60 tests. Co-Authored-By: Claude Opus 4.6 --- ...26-03-08-scheduled-notifications-design.md | 190 ++++++++++++++++++ examples/scheduled-notifications/.env.example | 2 + examples/scheduled-notifications/.gitignore | 4 + .../data/sample_events.json | 72 +++++++ .../scheduled-notifications/pyproject.toml | 19 ++ examples/scheduled-notifications/run.py | 124 ++++++++++++ .../src/notify_workflow/__init__.py | 0 .../src/notify_workflow/display.py | 40 ++++ .../src/notify_workflow/models.py | 49 +++++ .../src/notify_workflow/notify.py | 50 +++++ .../src/notify_workflow/rules.py | 106 ++++++++++ .../src/notify_workflow/schedules.py | 106 ++++++++++ .../src/notify_workflow/sources.py | 111 ++++++++++ .../scheduled-notifications/tests/__init__.py | 0 .../tests/test_display.py | 49 +++++ .../tests/test_models.py | 120 +++++++++++ .../tests/test_notify.py | 91 +++++++++ .../tests/test_rules.py | 146 ++++++++++++++ .../tests/test_schedules.py | 33 +++ .../tests/test_sources.py | 89 ++++++++ examples/scheduled-notifications/uv.lock | 79 ++++++++ 21 files changed, 1480 insertions(+) create mode 100644 docs/plans/2026-03-08-scheduled-notifications-design.md create mode 100644 examples/scheduled-notifications/.env.example create mode 100644 examples/scheduled-notifications/.gitignore create mode 100644 examples/scheduled-notifications/data/sample_events.json create mode 100644 examples/scheduled-notifications/pyproject.toml create mode 100644 examples/scheduled-notifications/run.py create mode 100644 examples/scheduled-notifications/src/notify_workflow/__init__.py create mode 100644 examples/scheduled-notifications/src/notify_workflow/display.py create mode 100644 examples/scheduled-notifications/src/notify_workflow/models.py create mode 100644 examples/scheduled-notifications/src/notify_workflow/notify.py create mode 100644 examples/scheduled-notifications/src/notify_workflow/rules.py create mode 100644 examples/scheduled-notifications/src/notify_workflow/schedules.py create mode 100644 examples/scheduled-notifications/src/notify_workflow/sources.py create mode 100644 examples/scheduled-notifications/tests/__init__.py create mode 100644 examples/scheduled-notifications/tests/test_display.py create mode 100644 examples/scheduled-notifications/tests/test_models.py create mode 100644 examples/scheduled-notifications/tests/test_notify.py create mode 100644 examples/scheduled-notifications/tests/test_rules.py create mode 100644 examples/scheduled-notifications/tests/test_schedules.py create mode 100644 examples/scheduled-notifications/tests/test_sources.py create mode 100644 examples/scheduled-notifications/uv.lock diff --git a/docs/plans/2026-03-08-scheduled-notifications-design.md b/docs/plans/2026-03-08-scheduled-notifications-design.md new file mode 100644 index 0000000..e65155a --- /dev/null +++ b/docs/plans/2026-03-08-scheduled-notifications-design.md @@ -0,0 +1,190 @@ +# Scheduled Notifications Example — Design + +**Date:** 2026-03-08 +**Pattern:** `trigger → api → logic → deliver` +**Status:** Approved + +## Goal + +Fourth runnable example for workflow-patterns. A notification scheduling system that loads events from schedule files, applies conditional rules to determine which notifications to fire, and delivers formatted notification digests. Demonstrates the `api → logic` pattern — data fetching followed by conditional branching. + +## Architecture + +``` +select_schedule() → load_events() → evaluate_rules() → deliver_notifications() +trigger api logic deliver +``` + +## Module Structure + +``` +examples/scheduled-notifications/ +├── run.py # CLI entry point +├── .env.example # Optional config (not needed for default mode) +├── .gitignore # .env, __pycache__, output/ +├── pyproject.toml # Zero runtime dependencies (stdlib only) +├── src/notify_workflow/ +│ ├── __init__.py +│ ├── models.py # Dataclasses: Event, Rule, Notification, Schedule +│ ├── schedules.py # 5 schedule presets with rules +│ ├── sources.py # Load events from JSON files (API layer) +│ ├── rules.py # Rule evaluation engine (logic layer) +│ ├── notify.py # Format and deliver notifications (file/terminal) +│ └── display.py # Terminal formatting +├── tests/ +│ ├── test_models.py +│ ├── test_schedules.py +│ ├── test_sources.py +│ ├── test_rules.py +│ ├── test_notify.py +│ ├── test_display.py +│ └── __init__.py +├── data/ # Sample event files (JSON) +│ └── sample_events.json +└── output/ # Generated notification digests +``` + +## Modules + +### models.py + +```python +@dataclass +class Event: + title: str + due_at: datetime + priority: str # "high", "medium", "low" + category: str # "meeting", "deadline", "reminder", "maintenance" + tags: list[str] + +@dataclass +class Rule: + name: str + description: str + match_categories: list[str] # event categories to match + time_window_hours: int # how far ahead to look + include_overdue: bool # also match past-due events + min_priority: str # minimum priority level + +@dataclass +class Notification: + event: Event + rule_name: str + urgency: str # "critical", "warning", "info" + message: str + +@dataclass +class Schedule: + name: str + description: str + rules: list[Rule] +``` + +### schedules.py + +5 curated schedule presets, each with specific rules: + +| Schedule | Focus | +|----------|-------| +| Morning Briefing | All events today, grouped by priority | +| Deadline Watch | Approaching and overdue deadlines only | +| Standup Prep | Open items and blockers for daily standup | +| Follow-up Tracker | Items awaiting response or action | +| Maintenance Alerts | System maintenance and downtime windows | + +Interactive selection menu. Also available via `--schedule` flag. + +### sources.py + +- `load_events(path)` → Parse JSON file into list of Event objects +- `load_events_from_dict(data)` → Create events from a dict (for testing/presets) +- `generate_sample_events(reference_time)` → Generate realistic sample events relative to a timestamp +- Default: generates sample events if no file provided + +### rules.py + +- `evaluate_rules(events, rules, now)` → Apply rules to events, return matching Notifications +- `determine_urgency(event, now)` → Calculate urgency based on time-to-due and priority +- `filter_by_time_window(events, window_hours, now, include_overdue)` → Time-based filtering +- `filter_by_priority(events, min_priority)` → Priority threshold filtering +- `filter_by_category(events, categories)` → Category matching +- `sort_notifications(notifications)` → Sort by urgency, then due time + +### notify.py + +- `format_notification(notification)` → Single notification as formatted string +- `format_digest(notifications, schedule)` → Full digest with header, grouped sections +- `save_digest(digest, output_dir)` → Write to timestamped markdown file +- `print_digest(digest)` → Print to terminal + +### display.py + +- `format_header(title)` → Boxed ASCII header +- `format_schedule_menu(schedules)` → Numbered schedule list +- `format_event_summary(event)` → One-line event summary with urgency indicator +- `format_urgency_badge(urgency)` → Visual urgency markers: [!!!] / [!!] / [i] + +## UX Flow + +``` +$ uv run python run.py + +╔══════════════════════════════════════════╗ +║ Scheduled Notifications — Setup ║ +╚══════════════════════════════════════════╝ + +Choose a schedule: + 1. Morning Briefing All events today, grouped by priority + 2. Deadline Watch Approaching and overdue deadlines + 3. Standup Prep Open items and blockers + 4. Follow-up Tracker Items awaiting response + 5. Maintenance Alerts System maintenance windows + +Schedule (1-5): 2 + +── Deadline Watch ── + +Loading events... + Found 8 events, applying rules... + +╔══════════════════════════════════════════╗ +║ Notification Digest ║ +╚══════════════════════════════════════════╝ + +[!!!] CRITICAL — API migration deadline (overdue by 2h) +[!!!] CRITICAL — Security patch deployment (due in 1h) +[!! ] WARNING — Q1 report submission (due in 6h) +[ i ] INFO — Team retrospective prep (due tomorrow) + +Summary: 2 critical, 1 warning, 1 info + +Digest saved to output/2026-03-08_deadline-watch.md +``` + +## Key Differences from Previous Examples + +| Aspect | Email Automation | Scheduled Notifications | +|--------|-----------------|------------------------| +| Pattern | `trigger → transform → deliver` | `trigger → api → logic → deliver` | +| New step | — | Conditional logic (rule engine) | +| Data source | User input | JSON events (simulated API) | +| Processing | Template rendering | Rule evaluation + filtering | +| Output | Single email | Multi-notification digest | +| Complexity | Template substitution | Time-based + priority logic | + +## Dependencies + +- Zero runtime dependencies (stdlib only) +- `pytest` (dev) +- Uses `datetime`, `json`, `pathlib` from stdlib + +## Testing Strategy + +- Rule evaluation with various time/priority scenarios +- Edge cases: no matching events, all overdue, empty schedule +- Event loading from JSON and dict sources +- Sample event generation with deterministic reference time +- Notification sorting and urgency determination +- Display formatting (badges, menus, headers) +- File I/O for digest output (tmp_path) +- Target: ~25 tests diff --git a/examples/scheduled-notifications/.env.example b/examples/scheduled-notifications/.env.example new file mode 100644 index 0000000..c0ecee6 --- /dev/null +++ b/examples/scheduled-notifications/.env.example @@ -0,0 +1,2 @@ +# Scheduled Notifications — no API keys required +# All functionality works with stdlib only diff --git a/examples/scheduled-notifications/.gitignore b/examples/scheduled-notifications/.gitignore new file mode 100644 index 0000000..f66c6ec --- /dev/null +++ b/examples/scheduled-notifications/.gitignore @@ -0,0 +1,4 @@ +.env +__pycache__/ +output/ +*.pyc diff --git a/examples/scheduled-notifications/data/sample_events.json b/examples/scheduled-notifications/data/sample_events.json new file mode 100644 index 0000000..9c314e6 --- /dev/null +++ b/examples/scheduled-notifications/data/sample_events.json @@ -0,0 +1,72 @@ +[ + { + "title": "API migration deadline", + "due_at": "2026-03-08T08:00:00", + "priority": "high", + "category": "deadline", + "tags": ["api", "migration"] + }, + { + "title": "Security patch deployment", + "due_at": "2026-03-08T15:00:00", + "priority": "high", + "category": "deadline", + "tags": ["security", "deploy"] + }, + { + "title": "Q1 report submission", + "due_at": "2026-03-08T18:00:00", + "priority": "medium", + "category": "deadline", + "tags": ["report", "q1"] + }, + { + "title": "Team retrospective", + "due_at": "2026-03-09T14:00:00", + "priority": "low", + "category": "meeting", + "tags": ["team", "retro"] + }, + { + "title": "Daily standup", + "due_at": "2026-03-08T09:30:00", + "priority": "medium", + "category": "meeting", + "tags": ["standup"] + }, + { + "title": "Review PR #142", + "due_at": "2026-03-08T13:00:00", + "priority": "medium", + "category": "reminder", + "tags": ["code-review"] + }, + { + "title": "Follow up with vendor", + "due_at": "2026-03-07T16:00:00", + "priority": "low", + "category": "reminder", + "tags": ["vendor", "followup"] + }, + { + "title": "Database maintenance window", + "due_at": "2026-03-10T02:00:00", + "priority": "high", + "category": "maintenance", + "tags": ["database", "downtime"] + }, + { + "title": "SSL certificate renewal", + "due_at": "2026-03-11T10:00:00", + "priority": "medium", + "category": "maintenance", + "tags": ["ssl", "security"] + }, + { + "title": "Client demo preparation", + "due_at": "2026-03-08T14:00:00", + "priority": "high", + "category": "reminder", + "tags": ["client", "demo"] + } +] diff --git a/examples/scheduled-notifications/pyproject.toml b/examples/scheduled-notifications/pyproject.toml new file mode 100644 index 0000000..b51415f --- /dev/null +++ b/examples/scheduled-notifications/pyproject.toml @@ -0,0 +1,19 @@ +[project] +name = "scheduled-notifications" +version = "0.1.0" +description = "Scheduled Notifications workflow: trigger -> api -> logic -> deliver" +requires-python = ">=3.12" +dependencies = [] + +[dependency-groups] +dev = ["pytest"] + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[tool.hatch.build.targets.wheel] +packages = ["src/notify_workflow"] + +[tool.pytest.ini_options] +testpaths = ["tests"] diff --git a/examples/scheduled-notifications/run.py b/examples/scheduled-notifications/run.py new file mode 100644 index 0000000..61b625e --- /dev/null +++ b/examples/scheduled-notifications/run.py @@ -0,0 +1,124 @@ +#!/usr/bin/env python3 +"""Scheduled Notifications workflow runner. + +Pattern: trigger -> api -> logic -> deliver + +Loads scheduled events, applies notification rules based on +time windows and priorities, and delivers formatted digests. + +Usage: + uv run python run.py # interactive schedule selection + uv run python run.py --schedule 2 # select schedule by number + uv run python run.py --events data/events.json # load events from file +""" + +import argparse +import os +import sys +from datetime import datetime +from pathlib import Path + +from notify_workflow.display import format_header, format_schedule_menu +from notify_workflow.notify import format_digest, format_notification, save_digest +from notify_workflow.rules import evaluate_rules +from notify_workflow.schedules import SCHEDULES, get_schedule +from notify_workflow.sources import generate_sample_events, load_events + +OUTPUT_DIR = Path(__file__).parent / "output" + + +def _load_dotenv(): + """Load .env file if it exists.""" + env_path = Path(__file__).parent / ".env" + if not env_path.exists(): + return + for line in env_path.read_text().splitlines(): + line = line.strip() + if not line or line.startswith("#"): + continue + key, _, value = line.partition("=") + if key and value: + os.environ.setdefault(key.strip(), value.strip()) + + +def _select_schedule() -> int: + """Interactive schedule selection. Returns 0-based index.""" + print(format_header("Scheduled Notifications — Setup")) + print("Choose a schedule:") + print(format_schedule_menu(SCHEDULES)) + print() + try: + choice = input(f"Schedule (1-{len(SCHEDULES)}): ").strip() + except (EOFError, KeyboardInterrupt): + print("\nAborted.") + sys.exit(0) + try: + return int(choice) - 1 + except ValueError: + return 0 + + +def main(): + _load_dotenv() + + parser = argparse.ArgumentParser( + description="Scheduled Notifications: trigger -> api -> logic -> deliver" + ) + parser.add_argument( + "--schedule", type=int, default=None, help="Schedule number (1-5)" + ) + parser.add_argument( + "--events", type=str, default=None, help="Path to events JSON file" + ) + args = parser.parse_args() + + # Step 1: Trigger — select schedule + if args.schedule is not None: + schedule = get_schedule(args.schedule - 1) + else: + schedule = get_schedule(_select_schedule()) + + print(f"\n── {schedule.name} ──\n") + + # Step 2: API — load events + now = datetime.now() + if args.events: + events_path = Path(args.events) + events = load_events(events_path) + if not events: + print(f" No events found in {events_path}") + sys.exit(1) + print(f" Loaded {len(events)} events from {events_path}") + else: + events = generate_sample_events(now) + print(f" Generated {len(events)} sample events") + + # Step 3: Logic — evaluate rules + print(f" Applying {len(schedule.rules)} rule(s)...\n") + notifications = evaluate_rules(events, schedule.rules, now) + + # Step 4: Deliver — format and save + print(format_header("Notification Digest")) + + if not notifications: + print(" No notifications match the current rules.\n") + else: + for n in notifications: + print(f" {format_notification(n)}") + print() + + counts = {} + for n in notifications: + counts[n.urgency] = counts.get(n.urgency, 0) + 1 + summary = ", ".join( + f"{c} {u}" for u, c in sorted(counts.items(), key=lambda x: -{"critical": 3, "warning": 2, "info": 1}.get(x[0], 0)) + ) + print(f" Summary: {summary}\n") + + digest = format_digest(notifications, schedule) + path = save_digest(digest, OUTPUT_DIR, schedule.name) + print(f" Digest saved to {path}\n") + + +if __name__ == "__main__": + main() diff --git a/examples/scheduled-notifications/src/notify_workflow/__init__.py b/examples/scheduled-notifications/src/notify_workflow/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/examples/scheduled-notifications/src/notify_workflow/display.py b/examples/scheduled-notifications/src/notify_workflow/display.py new file mode 100644 index 0000000..3a1a130 --- /dev/null +++ b/examples/scheduled-notifications/src/notify_workflow/display.py @@ -0,0 +1,40 @@ +"""Terminal display formatting for the Scheduled Notifications workflow.""" + +from notify_workflow.models import Schedule + + +def format_header(title: str) -> str: + """Create a boxed ASCII header.""" + width = max(len(title) + 8, 42) + top = "╔" + "═" * width + "╗" + mid = "║" + title.center(width) + "║" + bot = "╚" + "═" * width + "╝" + return f"\n{top}\n{mid}\n{bot}\n" + + +def format_urgency_badge(urgency: str) -> str: + """Format urgency as a visual badge.""" + badges = { + "critical": "[!!!]", + "warning": "[!! ]", + "info": "[ i ]", + } + return badges.get(urgency, "[ ]") + + +def format_urgency_label(urgency: str) -> str: + """Format urgency as an uppercase label.""" + labels = { + "critical": "CRITICAL", + "warning": "WARNING ", + "info": "INFO ", + } + return labels.get(urgency, "UNKNOWN ") + + +def format_schedule_menu(schedules: list[Schedule]) -> str: + """Format schedule selection menu.""" + lines = [] + for i, s in enumerate(schedules, 1): + lines.append(f" {i}. {s.name:<22s} {s.description}") + return "\n".join(lines) diff --git a/examples/scheduled-notifications/src/notify_workflow/models.py b/examples/scheduled-notifications/src/notify_workflow/models.py new file mode 100644 index 0000000..c4e9fc6 --- /dev/null +++ b/examples/scheduled-notifications/src/notify_workflow/models.py @@ -0,0 +1,49 @@ +"""Domain models for the Scheduled Notifications workflow.""" + +from dataclasses import dataclass, field +from datetime import datetime + + +PRIORITY_ORDER = {"high": 3, "medium": 2, "low": 1} +URGENCY_ORDER = {"critical": 3, "warning": 2, "info": 1} + + +@dataclass +class Event: + title: str + due_at: datetime + priority: str # "high", "medium", "low" + category: str # "meeting", "deadline", "reminder", "maintenance" + tags: list[str] = field(default_factory=list) + + def is_overdue(self, now: datetime) -> bool: + return self.due_at < now + + def hours_until(self, now: datetime) -> float: + delta = self.due_at - now + return delta.total_seconds() / 3600 + + +@dataclass +class Rule: + name: str + description: str + match_categories: list[str] + time_window_hours: int = 24 + include_overdue: bool = False + min_priority: str = "low" + + +@dataclass +class Notification: + event: Event + rule_name: str + urgency: str # "critical", "warning", "info" + message: str + + +@dataclass +class Schedule: + name: str + description: str + rules: list[Rule] diff --git a/examples/scheduled-notifications/src/notify_workflow/notify.py b/examples/scheduled-notifications/src/notify_workflow/notify.py new file mode 100644 index 0000000..1bdba26 --- /dev/null +++ b/examples/scheduled-notifications/src/notify_workflow/notify.py @@ -0,0 +1,50 @@ +"""Notification formatting and delivery for the Scheduled Notifications workflow (deliver layer).""" + +from datetime import datetime +from pathlib import Path + +from notify_workflow.display import format_urgency_badge, format_urgency_label +from notify_workflow.models import Notification, Schedule + + +def format_notification(notification: Notification) -> str: + """Format a single notification as a display line.""" + badge = format_urgency_badge(notification.urgency) + label = format_urgency_label(notification.urgency) + return f"{badge} {label} — {notification.message}" + + +def format_digest(notifications: list[Notification], schedule: Schedule) -> str: + """Format a full notification digest.""" + lines = [f"# {schedule.name}", f"_{schedule.description}_", ""] + + if not notifications: + lines.append("No notifications match the current rules.") + return "\n".join(lines) + + for n in notifications: + lines.append(format_notification(n)) + + lines.append("") + + counts = {} + for n in notifications: + counts[n.urgency] = counts.get(n.urgency, 0) + 1 + + summary_parts = [] + for urgency in ("critical", "warning", "info"): + if urgency in counts: + summary_parts.append(f"{counts[urgency]} {urgency}") + lines.append(f"Summary: {', '.join(summary_parts)}") + + return "\n".join(lines) + + +def save_digest(digest: str, output_dir: Path, schedule_name: str) -> Path: + """Save digest to a timestamped markdown file.""" + output_dir.mkdir(parents=True, exist_ok=True) + slug = schedule_name.lower().replace(" ", "-") + timestamp = datetime.now().strftime("%Y-%m-%d_%H%M%S") + path = output_dir / f"{timestamp}_{slug}.md" + path.write_text(digest) + return path diff --git a/examples/scheduled-notifications/src/notify_workflow/rules.py b/examples/scheduled-notifications/src/notify_workflow/rules.py new file mode 100644 index 0000000..61167db --- /dev/null +++ b/examples/scheduled-notifications/src/notify_workflow/rules.py @@ -0,0 +1,106 @@ +"""Rule evaluation engine for the Scheduled Notifications workflow (logic layer). + +Applies rules to events: filtering by time window, priority, and category, +then determines urgency and builds notifications. +""" + +from datetime import datetime + +from notify_workflow.models import ( + PRIORITY_ORDER, + URGENCY_ORDER, + Event, + Notification, + Rule, +) + + +def determine_urgency(event: Event, now: datetime) -> str: + """Determine notification urgency based on time-to-due and priority.""" + hours = event.hours_until(now) + is_high = event.priority == "high" + + if hours < 0: + return "critical" if is_high else "warning" + if hours <= 2: + return "critical" if is_high else "warning" + if hours <= 8: + return "warning" if event.priority in ("high", "medium") else "info" + return "info" + + +def filter_by_time_window( + events: list[Event], + window_hours: int, + now: datetime, + include_overdue: bool, +) -> list[Event]: + """Filter events by time window relative to now.""" + result = [] + for event in events: + hours = event.hours_until(now) + if hours < 0 and include_overdue: + result.append(event) + elif 0 <= hours <= window_hours: + result.append(event) + return result + + +def filter_by_priority(events: list[Event], min_priority: str) -> list[Event]: + """Filter events at or above minimum priority level.""" + threshold = PRIORITY_ORDER.get(min_priority, 1) + return [e for e in events if PRIORITY_ORDER.get(e.priority, 1) >= threshold] + + +def filter_by_category(events: list[Event], categories: list[str]) -> list[Event]: + """Filter events matching any of the given categories.""" + cat_set = set(categories) + return [e for e in events if e.category in cat_set] + + +def evaluate_rules( + events: list[Event], rules: list[Rule], now: datetime +) -> list[Notification]: + """Apply all rules to events and return deduplicated notifications.""" + seen_titles: set[str] = set() + notifications: list[Notification] = [] + + for rule in rules: + matched = filter_by_category(events, rule.match_categories) + matched = filter_by_priority(matched, rule.min_priority) + matched = filter_by_time_window( + matched, rule.time_window_hours, now, rule.include_overdue + ) + + for event in matched: + if event.title in seen_titles: + continue + seen_titles.add(event.title) + + urgency = determine_urgency(event, now) + hours = event.hours_until(now) + if hours < 0: + msg = f"{event.title} (overdue by {abs(hours):.0f}h)" + elif hours < 1: + msg = f"{event.title} (due in {hours * 60:.0f}min)" + else: + msg = f"{event.title} (due in {hours:.0f}h)" + + notifications.append( + Notification( + event=event, + rule_name=rule.name, + urgency=urgency, + message=msg, + ) + ) + + return sort_notifications(notifications) + + +def sort_notifications(notifications: list[Notification]) -> list[Notification]: + """Sort by urgency (critical first), then by due time (soonest first).""" + return sorted( + notifications, + key=lambda n: (-URGENCY_ORDER.get(n.urgency, 0), n.event.due_at), + ) diff --git a/examples/scheduled-notifications/src/notify_workflow/schedules.py b/examples/scheduled-notifications/src/notify_workflow/schedules.py new file mode 100644 index 0000000..cb6900b --- /dev/null +++ b/examples/scheduled-notifications/src/notify_workflow/schedules.py @@ -0,0 +1,106 @@ +"""Schedule presets for the Scheduled Notifications workflow.""" + +from notify_workflow.models import Rule, Schedule + +SCHEDULES: list[Schedule] = [ + Schedule( + name="Morning Briefing", + description="All events today, grouped by priority", + rules=[ + Rule( + name="today_all", + description="Everything due within 16 hours", + match_categories=["meeting", "deadline", "reminder", "maintenance"], + time_window_hours=16, + include_overdue=True, + min_priority="low", + ), + ], + ), + Schedule( + name="Deadline Watch", + description="Approaching and overdue deadlines", + rules=[ + Rule( + name="deadline_urgent", + description="Deadlines due within 4 hours or overdue", + match_categories=["deadline"], + time_window_hours=4, + include_overdue=True, + min_priority="medium", + ), + Rule( + name="deadline_upcoming", + description="Deadlines due within 24 hours", + match_categories=["deadline"], + time_window_hours=24, + include_overdue=False, + min_priority="low", + ), + ], + ), + Schedule( + name="Standup Prep", + description="Open items and blockers for daily standup", + rules=[ + Rule( + name="standup_blockers", + description="Overdue items and high-priority tasks", + match_categories=["deadline", "reminder"], + time_window_hours=8, + include_overdue=True, + min_priority="medium", + ), + Rule( + name="standup_meetings", + description="Today's meetings", + match_categories=["meeting"], + time_window_hours=10, + include_overdue=False, + min_priority="low", + ), + ], + ), + Schedule( + name="Follow-up Tracker", + description="Items awaiting response or action", + rules=[ + Rule( + name="followup_overdue", + description="Overdue reminders and deadlines", + match_categories=["reminder", "deadline"], + time_window_hours=0, + include_overdue=True, + min_priority="low", + ), + Rule( + name="followup_upcoming", + description="Reminders due within 48 hours", + match_categories=["reminder"], + time_window_hours=48, + include_overdue=False, + min_priority="low", + ), + ], + ), + Schedule( + name="Maintenance Alerts", + description="System maintenance and downtime windows", + rules=[ + Rule( + name="maintenance_soon", + description="Maintenance events within 72 hours", + match_categories=["maintenance"], + time_window_hours=72, + include_overdue=True, + min_priority="low", + ), + ], + ), +] + + +def get_schedule(index: int) -> Schedule: + """Get schedule by index, clamping to valid range.""" + clamped = max(0, min(index, len(SCHEDULES) - 1)) + return SCHEDULES[clamped] diff --git a/examples/scheduled-notifications/src/notify_workflow/sources.py b/examples/scheduled-notifications/src/notify_workflow/sources.py new file mode 100644 index 0000000..ce70b8f --- /dev/null +++ b/examples/scheduled-notifications/src/notify_workflow/sources.py @@ -0,0 +1,111 @@ +"""Event sources for the Scheduled Notifications workflow (API layer). + +Loads events from JSON files or generates sample data. +In a production workflow, this would call a calendar or task API. +""" + +import json +from datetime import datetime, timedelta +from pathlib import Path + +from notify_workflow.models import Event + + +def load_events_from_dicts(data: list[dict]) -> list[Event]: + """Create Event objects from a list of dicts.""" + events = [] + for item in data: + events.append( + Event( + title=item["title"], + due_at=datetime.fromisoformat(item["due_at"]), + priority=item["priority"], + category=item["category"], + tags=item.get("tags", []), + ) + ) + return events + + +def load_events(path: Path) -> list[Event]: + """Load events from a JSON file. Returns empty list if file not found.""" + if not path.exists(): + return [] + data = json.loads(path.read_text()) + return load_events_from_dicts(data) + + +def generate_sample_events(reference_time: datetime) -> list[Event]: + """Generate realistic sample events relative to a reference time.""" + return [ + Event( + title="API migration deadline", + due_at=reference_time + timedelta(hours=-2), + priority="high", + category="deadline", + tags=["api", "migration"], + ), + Event( + title="Security patch deployment", + due_at=reference_time + timedelta(hours=1), + priority="high", + category="deadline", + tags=["security", "deploy"], + ), + Event( + title="Q1 report submission", + due_at=reference_time + timedelta(hours=6), + priority="medium", + category="deadline", + tags=["report", "q1"], + ), + Event( + title="Team retrospective", + due_at=reference_time + timedelta(hours=28), + priority="low", + category="meeting", + tags=["team", "retro"], + ), + Event( + title="Daily standup", + due_at=reference_time + timedelta(hours=1, minutes=30), + priority="medium", + category="meeting", + tags=["standup"], + ), + Event( + title="Review PR #142", + due_at=reference_time + timedelta(hours=3), + priority="medium", + category="reminder", + tags=["code-review"], + ), + Event( + title="Follow up with vendor", + due_at=reference_time + timedelta(hours=-5), + priority="low", + category="reminder", + tags=["vendor", "followup"], + ), + Event( + title="Database maintenance window", + due_at=reference_time + timedelta(hours=48), + priority="high", + category="maintenance", + tags=["database", "downtime"], + ), + Event( + title="SSL certificate renewal", + due_at=reference_time + timedelta(hours=72), + priority="medium", + category="maintenance", + tags=["ssl", "security"], + ), + Event( + title="Client demo preparation", + due_at=reference_time + timedelta(hours=4), + priority="high", + category="reminder", + tags=["client", "demo"], + ), + ] diff --git a/examples/scheduled-notifications/tests/__init__.py b/examples/scheduled-notifications/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/examples/scheduled-notifications/tests/test_display.py b/examples/scheduled-notifications/tests/test_display.py new file mode 100644 index 0000000..c5e2cb4 --- /dev/null +++ b/examples/scheduled-notifications/tests/test_display.py @@ -0,0 +1,49 @@ +"""Tests for notify_workflow.display.""" + +from notify_workflow.display import ( + format_header, + format_schedule_menu, + format_urgency_badge, +) +from notify_workflow.models import Rule, Schedule + + +class TestFormatHeader: + def test_contains_title(self): + result = format_header("Test Title") + assert "Test Title" in result + + def test_has_box_chars(self): + result = format_header("Test") + assert "═" in result or "+" in result or "╔" in result + + +class TestFormatUrgencyBadge: + def test_critical(self): + badge = format_urgency_badge("critical") + assert "!!!" in badge + + def test_warning(self): + badge = format_urgency_badge("warning") + assert "!!" in badge + + def test_info(self): + badge = format_urgency_badge("info") + assert "i" in badge + + +class TestFormatScheduleMenu: + def test_lists_all_schedules(self): + schedules = [ + Schedule(name="A", description="Desc A", rules=[ + Rule(name="r", description="d", match_categories=["x"]) + ]), + Schedule(name="B", description="Desc B", rules=[ + Rule(name="r", description="d", match_categories=["x"]) + ]), + ] + result = format_schedule_menu(schedules) + assert "1." in result + assert "2." in result + assert "A" in result + assert "B" in result diff --git a/examples/scheduled-notifications/tests/test_models.py b/examples/scheduled-notifications/tests/test_models.py new file mode 100644 index 0000000..4a273a1 --- /dev/null +++ b/examples/scheduled-notifications/tests/test_models.py @@ -0,0 +1,120 @@ +"""Tests for notify_workflow.models.""" + +from datetime import datetime + +from notify_workflow.models import Event, Notification, Rule, Schedule + + +class TestEvent: + def test_create_event(self): + event = Event( + title="Deploy v2.0", + due_at=datetime(2026, 3, 8, 14, 0), + priority="high", + category="deadline", + tags=["deploy", "prod"], + ) + assert event.title == "Deploy v2.0" + assert event.priority == "high" + assert event.category == "deadline" + assert event.tags == ["deploy", "prod"] + + def test_event_default_tags(self): + event = Event( + title="Standup", + due_at=datetime(2026, 3, 8, 9, 0), + priority="low", + category="meeting", + ) + assert event.tags == [] + + def test_event_is_overdue(self): + past = datetime(2026, 3, 7, 10, 0) + event = Event(title="Past", due_at=past, priority="high", category="deadline") + now = datetime(2026, 3, 8, 10, 0) + assert event.is_overdue(now) is True + + def test_event_is_not_overdue(self): + future = datetime(2026, 3, 9, 10, 0) + event = Event(title="Future", due_at=future, priority="low", category="meeting") + now = datetime(2026, 3, 8, 10, 0) + assert event.is_overdue(now) is False + + def test_event_hours_until(self): + event = Event( + title="Meeting", + due_at=datetime(2026, 3, 8, 16, 0), + priority="medium", + category="meeting", + ) + now = datetime(2026, 3, 8, 10, 0) + assert event.hours_until(now) == 6.0 + + def test_event_hours_until_negative_when_overdue(self): + event = Event( + title="Past", + due_at=datetime(2026, 3, 8, 8, 0), + priority="high", + category="deadline", + ) + now = datetime(2026, 3, 8, 10, 0) + assert event.hours_until(now) == -2.0 + + +class TestRule: + def test_create_rule(self): + rule = Rule( + name="deadline_alert", + description="Alert for approaching deadlines", + match_categories=["deadline"], + time_window_hours=24, + include_overdue=True, + min_priority="low", + ) + assert rule.name == "deadline_alert" + assert rule.time_window_hours == 24 + assert rule.include_overdue is True + + def test_rule_defaults(self): + rule = Rule( + name="simple", + description="Simple rule", + match_categories=["meeting"], + ) + assert rule.time_window_hours == 24 + assert rule.include_overdue is False + assert rule.min_priority == "low" + + +class TestNotification: + def test_create_notification(self): + event = Event( + title="Deploy", + due_at=datetime(2026, 3, 8, 14, 0), + priority="high", + category="deadline", + ) + notif = Notification( + event=event, + rule_name="deadline_alert", + urgency="critical", + message="Deploy is due in 2 hours", + ) + assert notif.urgency == "critical" + assert notif.rule_name == "deadline_alert" + + +class TestSchedule: + def test_create_schedule(self): + rule = Rule( + name="test", + description="Test rule", + match_categories=["meeting"], + ) + schedule = Schedule( + name="Morning Briefing", + description="All events today", + rules=[rule], + ) + assert schedule.name == "Morning Briefing" + assert len(schedule.rules) == 1 diff --git a/examples/scheduled-notifications/tests/test_notify.py b/examples/scheduled-notifications/tests/test_notify.py new file mode 100644 index 0000000..bffa564 --- /dev/null +++ b/examples/scheduled-notifications/tests/test_notify.py @@ -0,0 +1,91 @@ +"""Tests for notify_workflow.notify.""" + +from datetime import datetime + +from notify_workflow.models import Event, Notification, Schedule, Rule +from notify_workflow.notify import format_digest, format_notification, save_digest + + +class TestFormatNotification: + def test_includes_urgency_and_message(self): + event = Event( + title="Deploy", + due_at=datetime(2026, 3, 8, 14, 0), + priority="high", + category="deadline", + ) + notif = Notification( + event=event, + rule_name="test", + urgency="critical", + message="Deploy (due in 2h)", + ) + result = format_notification(notif) + assert "Deploy" in result + assert "CRITICAL" in result or "!!!" in result + + +class TestFormatDigest: + def test_includes_schedule_name(self): + schedule = Schedule( + name="Test Schedule", + description="Testing", + rules=[Rule(name="r", description="d", match_categories=["x"])], + ) + result = format_digest([], schedule) + assert "Test Schedule" in result + + def test_includes_no_notifications_message(self): + schedule = Schedule( + name="Test", + description="Testing", + rules=[Rule(name="r", description="d", match_categories=["x"])], + ) + result = format_digest([], schedule) + assert "No notifications" in result or "no matching" in result.lower() + + def test_includes_notifications(self): + event = Event( + title="Meeting", + due_at=datetime(2026, 3, 8, 14, 0), + priority="medium", + category="meeting", + ) + notif = Notification(event=event, rule_name="r", urgency="warning", message="Meeting (due in 4h)") + schedule = Schedule( + name="Test", + description="Testing", + rules=[Rule(name="r", description="d", match_categories=["meeting"])], + ) + result = format_digest([notif], schedule) + assert "Meeting" in result + + def test_includes_summary_counts(self): + event = Event(title="A", due_at=datetime(2026, 3, 8, 14, 0), priority="high", category="deadline") + notifications = [ + Notification(event=event, rule_name="r", urgency="critical", message="A"), + Notification(event=event, rule_name="r", urgency="warning", message="B"), + ] + schedule = Schedule( + name="Test", + description="Testing", + rules=[Rule(name="r", description="d", match_categories=["deadline"])], + ) + result = format_digest(notifications, schedule) + assert "1 critical" in result.lower() + + +class TestSaveDigest: + def test_saves_to_file(self, tmp_path): + path = save_digest("Test digest content", tmp_path, schedule_name="test") + assert path.exists() + assert path.read_text() == "Test digest content" + + def test_filename_contains_schedule_name(self, tmp_path): + path = save_digest("Content", tmp_path, schedule_name="deadline-watch") + assert "deadline-watch" in path.name + + def test_creates_output_dir(self, tmp_path): + output = tmp_path / "subdir" + path = save_digest("Content", output, schedule_name="test") + assert path.exists() diff --git a/examples/scheduled-notifications/tests/test_rules.py b/examples/scheduled-notifications/tests/test_rules.py new file mode 100644 index 0000000..934342c --- /dev/null +++ b/examples/scheduled-notifications/tests/test_rules.py @@ -0,0 +1,146 @@ +"""Tests for notify_workflow.rules.""" + +from datetime import datetime, timedelta + +from notify_workflow.models import Event, Notification, Rule +from notify_workflow.rules import ( + determine_urgency, + evaluate_rules, + filter_by_category, + filter_by_priority, + filter_by_time_window, + sort_notifications, +) + +NOW = datetime(2026, 3, 8, 10, 0) + + +def _event(title="Test", hours_offset=2, priority="medium", category="deadline"): + return Event( + title=title, + due_at=NOW + timedelta(hours=hours_offset), + priority=priority, + category=category, + ) + + +class TestDetermineUrgency: + def test_overdue_high_is_critical(self): + event = _event(hours_offset=-1, priority="high") + assert determine_urgency(event, NOW) == "critical" + + def test_overdue_low_is_warning(self): + event = _event(hours_offset=-1, priority="low") + assert determine_urgency(event, NOW) == "warning" + + def test_due_soon_high_is_critical(self): + event = _event(hours_offset=1, priority="high") + assert determine_urgency(event, NOW) == "critical" + + def test_due_soon_medium_is_warning(self): + event = _event(hours_offset=2, priority="medium") + assert determine_urgency(event, NOW) == "warning" + + def test_far_future_low_is_info(self): + event = _event(hours_offset=20, priority="low") + assert determine_urgency(event, NOW) == "info" + + +class TestFilterByTimeWindow: + def test_includes_events_within_window(self): + events = [_event(hours_offset=2), _event(hours_offset=5)] + result = filter_by_time_window(events, window_hours=6, now=NOW, include_overdue=False) + assert len(result) == 2 + + def test_excludes_events_outside_window(self): + events = [_event(hours_offset=10)] + result = filter_by_time_window(events, window_hours=6, now=NOW, include_overdue=False) + assert len(result) == 0 + + def test_excludes_overdue_when_not_included(self): + events = [_event(hours_offset=-2)] + result = filter_by_time_window(events, window_hours=6, now=NOW, include_overdue=False) + assert len(result) == 0 + + def test_includes_overdue_when_included(self): + events = [_event(hours_offset=-2)] + result = filter_by_time_window(events, window_hours=6, now=NOW, include_overdue=True) + assert len(result) == 1 + + +class TestFilterByPriority: + def test_filter_minimum_high(self): + events = [_event(priority="high"), _event(priority="medium"), _event(priority="low")] + result = filter_by_priority(events, "high") + assert len(result) == 1 + + def test_filter_minimum_medium(self): + events = [_event(priority="high"), _event(priority="medium"), _event(priority="low")] + result = filter_by_priority(events, "medium") + assert len(result) == 2 + + def test_filter_minimum_low(self): + events = [_event(priority="high"), _event(priority="medium"), _event(priority="low")] + result = filter_by_priority(events, "low") + assert len(result) == 3 + + +class TestFilterByCategory: + def test_single_category(self): + events = [_event(category="deadline"), _event(category="meeting")] + result = filter_by_category(events, ["deadline"]) + assert len(result) == 1 + + def test_multiple_categories(self): + events = [_event(category="deadline"), _event(category="meeting"), _event(category="reminder")] + result = filter_by_category(events, ["deadline", "meeting"]) + assert len(result) == 2 + + def test_no_match(self): + events = [_event(category="meeting")] + result = filter_by_category(events, ["deadline"]) + assert len(result) == 0 + + +class TestEvaluateRules: + def test_basic_rule_evaluation(self): + events = [_event(hours_offset=2, category="deadline", priority="high")] + rules = [ + Rule( + name="test_rule", + description="Test", + match_categories=["deadline"], + time_window_hours=6, + include_overdue=False, + min_priority="low", + ) + ] + result = evaluate_rules(events, rules, NOW) + assert len(result) == 1 + assert result[0].rule_name == "test_rule" + + def test_no_matching_events(self): + events = [_event(hours_offset=2, category="meeting")] + rules = [ + Rule(name="deadline_only", description="Test", match_categories=["deadline"]) + ] + result = evaluate_rules(events, rules, NOW) + assert len(result) == 0 + + def test_deduplicates_across_rules(self): + event = _event(hours_offset=2, category="deadline", priority="high") + rules = [ + Rule(name="rule1", description="Test 1", match_categories=["deadline"], time_window_hours=6), + Rule(name="rule2", description="Test 2", match_categories=["deadline"], time_window_hours=24), + ] + result = evaluate_rules([event], rules, NOW) + assert len(result) == 1 # same event, first matching rule wins + + +class TestSortNotifications: + def test_sort_by_urgency_then_time(self): + n1 = Notification(event=_event(hours_offset=5), rule_name="r", urgency="info", message="") + n2 = Notification(event=_event(hours_offset=1), rule_name="r", urgency="critical", message="") + n3 = Notification(event=_event(hours_offset=3), rule_name="r", urgency="warning", message="") + result = sort_notifications([n1, n2, n3]) + assert [n.urgency for n in result] == ["critical", "warning", "info"] diff --git a/examples/scheduled-notifications/tests/test_schedules.py b/examples/scheduled-notifications/tests/test_schedules.py new file mode 100644 index 0000000..e8d09c6 --- /dev/null +++ b/examples/scheduled-notifications/tests/test_schedules.py @@ -0,0 +1,33 @@ +"""Tests for notify_workflow.schedules.""" + +from notify_workflow.models import Schedule +from notify_workflow.schedules import SCHEDULES, get_schedule + + +class TestSchedules: + def test_five_schedules_defined(self): + assert len(SCHEDULES) == 5 + + def test_all_are_schedule_instances(self): + for s in SCHEDULES: + assert isinstance(s, Schedule) + + def test_all_have_rules(self): + for s in SCHEDULES: + assert len(s.rules) >= 1, f"{s.name} has no rules" + + def test_schedule_names_unique(self): + names = [s.name for s in SCHEDULES] + assert len(names) == len(set(names)) + + def test_get_schedule_by_index(self): + schedule = get_schedule(0) + assert schedule == SCHEDULES[0] + + def test_get_schedule_clamps_negative(self): + schedule = get_schedule(-1) + assert schedule == SCHEDULES[0] + + def test_get_schedule_clamps_overflow(self): + schedule = get_schedule(99) + assert schedule == SCHEDULES[-1] diff --git a/examples/scheduled-notifications/tests/test_sources.py b/examples/scheduled-notifications/tests/test_sources.py new file mode 100644 index 0000000..091a756 --- /dev/null +++ b/examples/scheduled-notifications/tests/test_sources.py @@ -0,0 +1,89 @@ +"""Tests for notify_workflow.sources.""" + +import json +from datetime import datetime + +from notify_workflow.models import Event +from notify_workflow.sources import ( + generate_sample_events, + load_events, + load_events_from_dicts, +) + + +class TestLoadEventsFromDicts: + def test_load_single_event(self): + data = [ + { + "title": "Deploy", + "due_at": "2026-03-08T14:00:00", + "priority": "high", + "category": "deadline", + "tags": ["prod"], + } + ] + events = load_events_from_dicts(data) + assert len(events) == 1 + assert events[0].title == "Deploy" + assert events[0].priority == "high" + assert events[0].tags == ["prod"] + + def test_load_multiple_events(self): + data = [ + {"title": "A", "due_at": "2026-03-08T10:00:00", "priority": "low", "category": "meeting"}, + {"title": "B", "due_at": "2026-03-08T12:00:00", "priority": "high", "category": "deadline"}, + ] + events = load_events_from_dicts(data) + assert len(events) == 2 + + def test_load_event_default_tags(self): + data = [ + {"title": "No tags", "due_at": "2026-03-08T10:00:00", "priority": "low", "category": "meeting"} + ] + events = load_events_from_dicts(data) + assert events[0].tags == [] + + def test_load_empty_list(self): + assert load_events_from_dicts([]) == [] + + +class TestLoadEvents: + def test_load_from_json_file(self, tmp_path): + data = [ + {"title": "From file", "due_at": "2026-03-08T09:00:00", "priority": "medium", "category": "reminder"} + ] + path = tmp_path / "events.json" + path.write_text(json.dumps(data)) + events = load_events(path) + assert len(events) == 1 + assert events[0].title == "From file" + + def test_load_nonexistent_file(self, tmp_path): + path = tmp_path / "missing.json" + events = load_events(path) + assert events == [] + + +class TestGenerateSampleEvents: + def test_generates_events(self): + ref = datetime(2026, 3, 8, 9, 0) + events = generate_sample_events(ref) + assert len(events) >= 6 + + def test_all_are_event_instances(self): + ref = datetime(2026, 3, 8, 9, 0) + events = generate_sample_events(ref) + for e in events: + assert isinstance(e, Event) + + def test_includes_multiple_categories(self): + ref = datetime(2026, 3, 8, 9, 0) + events = generate_sample_events(ref) + categories = {e.category for e in events} + assert len(categories) >= 3 + + def test_includes_multiple_priorities(self): + ref = datetime(2026, 3, 8, 9, 0) + events = generate_sample_events(ref) + priorities = {e.priority for e in events} + assert len(priorities) >= 2 diff --git a/examples/scheduled-notifications/uv.lock b/examples/scheduled-notifications/uv.lock new file mode 100644 index 0000000..4188ddc --- /dev/null +++ b/examples/scheduled-notifications/uv.lock @@ -0,0 +1,79 @@ +version = 1 +revision = 3 +requires-python = ">=3.12" + +[[package]] +name = "colorama" +version = "0.4.6" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/d8/53/6f443c9a4a8358a93a6792e2acffb9d9d5cb0a5cfd8802644b7b1c9a02e4/colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44", size = 27697, upload-time = "2022-10-25T02:36:22.414Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/d1/d6/3965ed04c63042e047cb6a3e6ed1a63a35087b6a609aa3a15ed8ac56c221/colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6", size = 25335, upload-time = "2022-10-25T02:36:20.889Z" }, +] + +[[package]] +name = "iniconfig" +version = "2.3.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/72/34/14ca021ce8e5dfedc35312d08ba8bf51fdd999c576889fc2c24cb97f4f10/iniconfig-2.3.0.tar.gz", hash = "sha256:c76315c77db068650d49c5b56314774a7804df16fee4402c1f19d6d15d8c4730", size = 20503, upload-time = "2025-10-18T21:55:43.219Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/cb/b1/3846dd7f199d53cb17f49cba7e651e9ce294d8497c8c150530ed11865bb8/iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12", size = 7484, upload-time = "2025-10-18T21:55:41.639Z" }, +] + +[[package]] +name = "packaging" +version = "26.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/65/ee/299d360cdc32edc7d2cf530f3accf79c4fca01e96ffc950d8a52213bd8e4/packaging-26.0.tar.gz", hash = "sha256:00243ae351a257117b6a241061796684b084ed1c516a08c48a3f7e147a9d80b4", size = 143416, upload-time = "2026-01-21T20:50:39.064Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b7/b9/c538f279a4e237a006a2c98387d081e9eb060d203d8ed34467cc0f0b9b53/packaging-26.0-py3-none-any.whl", hash = "sha256:b36f1fef9334a5588b4166f8bcd26a14e521f2b55e6b9de3aaa80d3ff7a37529", size = 74366, upload-time = "2026-01-21T20:50:37.788Z" }, +] + +[[package]] +name = "pluggy" +version = "1.6.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/f9/e2/3e91f31a7d2b083fe6ef3fa267035b518369d9511ffab804f839851d2779/pluggy-1.6.0.tar.gz", hash = "sha256:7dcc130b76258d33b90f61b658791dede3486c3e6bfb003ee5c9bfb396dd22f3", size = 69412, upload-time = "2025-05-15T12:30:07.975Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/54/20/4d324d65cc6d9205fabedc306948156824eb9f0ee1633355a8f7ec5c66bf/pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746", size = 20538, upload-time = "2025-05-15T12:30:06.134Z" }, +] + +[[package]] +name = "pygments" +version = "2.19.2" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/b0/77/a5b8c569bf593b0140bde72ea885a803b82086995367bf2037de0159d924/pygments-2.19.2.tar.gz", hash = "sha256:636cb2477cec7f8952536970bc533bc43743542f70392ae026374600add5b887", size = 4968631, upload-time = "2025-06-21T13:39:12.283Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/c7/21/705964c7812476f378728bdf590ca4b771ec72385c533964653c68e86bdc/pygments-2.19.2-py3-none-any.whl", hash = "sha256:86540386c03d588bb81d44bc3928634ff26449851e99741617ecb9037ee5ec0b", size = 1225217, upload-time = "2025-06-21T13:39:07.939Z" }, +] + +[[package]] +name = "pytest" +version = "9.0.2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "colorama", marker = "sys_platform == 'win32'" }, + { name = "iniconfig" }, + { name = "packaging" }, + { name = "pluggy" }, + { name = "pygments" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/d1/db/7ef3487e0fb0049ddb5ce41d3a49c235bf9ad299b6a25d5780a89f19230f/pytest-9.0.2.tar.gz", hash = "sha256:75186651a92bd89611d1d9fc20f0b4345fd827c41ccd5c299a868a05d70edf11", size = 1568901, upload-time = "2025-12-06T21:30:51.014Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/3b/ab/b3226f0bd7cdcf710fbede2b3548584366da3b19b5021e74f5bde2a8fa3f/pytest-9.0.2-py3-none-any.whl", hash = "sha256:711ffd45bf766d5264d487b917733b453d917afd2b0ad65223959f59089f875b", size = 374801, upload-time = "2025-12-06T21:30:49.154Z" }, +] + +[[package]] +name = "scheduled-notifications" +version = "0.1.0" +source = { editable = "." } + +[package.dev-dependencies] +dev = [ + { name = "pytest" }, +] + +[package.metadata] + +[package.metadata.requires-dev] +dev = [{ name = "pytest" }]