- NotifyClient with simple notify() and event builder APIs - Full async/await support with httpx - Offline queue with JSON file persistence - Event builder with fluent API - Notification management (get, markAsRead, unreadCount) - Type hints with dataclasses - Python 3.8+ compatible |
||
|---|---|---|
| src/ironnotify | ||
| .gitignore | ||
| LICENSE | ||
| README.md | ||
| pyproject.toml | ||
README.md
IronNotify SDK for Python
Event notifications and alerts SDK for Python applications. Send notifications, receive real-time updates, and manage notification state.
Installation
pip install ironnotify
Quick Start
Send a Simple Notification
import ironnotify
# Initialize
ironnotify.init("ak_live_xxxxx")
# Send a simple notification
ironnotify.notify(
"order.created",
"New Order Received",
message="Order #1234 has been placed",
severity="success",
metadata={"order_id": "1234", "amount": 99.99},
)
Fluent Event Builder
import ironnotify
ironnotify.init("ak_live_xxxxx")
# Build complex notifications with the fluent API
result = ironnotify.event("payment.failed") \
.with_title("Payment Failed") \
.with_message("Payment could not be processed") \
.with_severity("error") \
.with_metadata({"order_id": "1234", "reason": "Card declined"}) \
.with_action("Retry Payment", url="/orders/1234/retry", style="primary") \
.with_action("Contact Support", action="open_support") \
.for_user("user-123") \
.with_deduplication_key("payment-failed-1234") \
.expires_in(86400) # 24 hours \
.send()
Async Support
import asyncio
import ironnotify
async def main():
ironnotify.init("ak_live_xxxxx")
# Async notification
result = await ironnotify.notify_async(
"user.signup",
"New User Registered",
severity="success",
)
# Async event builder
result = await ironnotify.event("order.shipped") \
.with_title("Order Shipped") \
.for_user("user-123") \
.send_async()
# Get notifications async
notifications = await ironnotify.get_notifications_async(limit=10)
asyncio.run(main())
Using the Client Directly
from ironnotify import NotifyClient, NotifyOptions
client = NotifyClient(NotifyOptions(
api_key="ak_live_xxxxx",
debug=True,
))
# Send notification
result = client.notify("event.type", "Title")
# Use event builder
result = client.event("event.type") \
.with_title("Title") \
.send()
# Clean up
client.close()
Configuration
from ironnotify import NotifyOptions, NotifyClient
options = NotifyOptions(
api_key="ak_live_xxxxx",
api_base_url="https://api.ironnotify.com",
ws_url="wss://ws.ironnotify.com",
debug=False,
enable_offline_queue=True,
max_offline_queue_size=100,
auto_reconnect=True,
max_reconnect_attempts=5,
reconnect_delay=1.0,
)
client = NotifyClient(options)
Configuration Options
| Option | Type | Default | Description |
|---|---|---|---|
api_key |
str | required | Your API key (ak_live_xxx or ak_test_xxx) |
api_base_url |
str | 'https://api.ironnotify.com' | API base URL |
ws_url |
str | 'wss://ws.ironnotify.com' | WebSocket URL |
debug |
bool | False | Enable debug logging |
enable_offline_queue |
bool | True | Queue notifications when offline |
max_offline_queue_size |
int | 100 | Max offline queue size |
auto_reconnect |
bool | True | Auto-reconnect WebSocket |
max_reconnect_attempts |
int | 5 | Max reconnection attempts |
reconnect_delay |
float | 1.0 | Base reconnection delay (seconds) |
Features
- Simple Notifications: Send notifications with one line of code
- Fluent Builder: Build complex notifications with an intuitive API
- Async Support: Full async/await support for all operations
- Offline Queue: Notifications are queued when offline
- Type Hints: Full type annotations for IDE support
- Dataclasses: Clean data models using Python dataclasses
Notification Options
Severity Levels
from ironnotify import SeverityLevel
# Available levels: "info", "success", "warning", "error", "critical"
ironnotify.notify("alert", "System Alert", severity="critical")
Actions
ironnotify.event("order.shipped") \
.with_title("Order Shipped") \
.with_action("Track Package", url="https://tracking.example.com/123", style="primary") \
.with_action("View Order", action="view_order") \
.send()
Deduplication
Prevent duplicate notifications:
ironnotify.event("reminder") \
.with_title("Daily Reminder") \
.with_deduplication_key("daily-reminder-2024-01-15") \
.send()
Grouping
Group related notifications:
ironnotify.event("comment.new") \
.with_title("New Comment") \
.with_group_key("post-123-comments") \
.send()
Expiration
from datetime import datetime, timedelta
# Expires in 1 hour
ironnotify.event("flash_sale") \
.with_title("Flash Sale!") \
.expires_in(3600) \
.send()
# Expires at specific time
ironnotify.event("event_reminder") \
.with_title("Event Tomorrow") \
.expires_on(datetime.utcnow() + timedelta(days=1)) \
.send()
Managing Notifications
Get Notifications
# Get all notifications
notifications = ironnotify.get_notifications()
# With options
unread = ironnotify.get_notifications(limit=10, offset=0, unread_only=True)
# Async
notifications = await ironnotify.get_notifications_async()
Mark as Read
# Mark single notification
client.mark_as_read("notification-id")
# Mark all as read
client.mark_all_as_read()
Get Unread Count
count = client.get_unread_count()
print(f"You have {count} unread notifications")
Real-Time Notifications
import ironnotify
ironnotify.init("ak_live_xxxxx")
def handle_notification(notification):
print(f"New notification: {notification.title}")
ironnotify.on_notification(handle_notification)
ironnotify.connect()
ironnotify.subscribe_to_user("user-123")
ironnotify.subscribe_to_app()
Offline Support
Notifications are automatically queued when offline:
import ironnotify
ironnotify.init("ak_live_xxxxx")
# This will be queued if offline
ironnotify.notify("event", "Title")
# Manually flush the queue
ironnotify.flush()
# Or async
await ironnotify.flush_async()
Requirements
- Python 3.8+
- httpx
- websockets (for real-time features)
Links
License
MIT License - see LICENSE for details.