The ObjNotify.py module provides a centralized system for sending notifications through various channels, including Slack, Discord, SMS, and more. It is designed to be highly configurable and can be used for both immediate, direct notifications and asynchronous, queued delivery.
The integration is built on a clear separation of concerns:
def_notify table): Defines if, how, and where a notification should be sent for each channel.ObjNotifySlack.py): Dedicated modules that handle the low-level communication with each specific service's API (like Slack or Discord).ObjNotify.py): This module, which contains the Notify class, provides the high-level logic to trigger notifications based on the defined configurations.def_notify TableThe entire notification system is driven by data in the def_notify table. To enable a channel for a specific notification, you need a row in this table with a unique notifycode. The key columns for each channel follow a consistent pattern.
NotifySlackSend (char(1)): This acts as a feature flag. If this is set to 'Y', Slack notifications are enabled for this notifycode.NotifySlackConnection (char(255)): This field stores the destination for the message, which for Slack is the channel name (e.g., #general, #alerts).NotifySlackText (mediumtext): This contains the default message template to be sent. It can include placeholders like $message$ or $user$ that will be replaced with dynamic content at runtime.This pattern is repeated for each supported channel (e.g., NotifyDiscordSend, NotifySmsConnection, etc.).
ObjNotifySlack.py)Each notification channel has its own dedicated wrapper module responsible for the specifics of communicating with its API.
Objects.global_config).transmit_message Method: Each wrapper has a standardized transmit_message method. This method takes the message content and destination (like a channel name or phone number) as arguments and handles the API call to send the message.ObjNotify.pyThe Notify class provides two distinct methods for sending notifications, each suited for different use cases.
Run Method)This is the most common way to send a single, immediate notification directly from your code. It's ideal for real-time alerts and confirmations.
Notify().Run("YOUR_NOTIFY_CODE", "Your dynamic message").Run method first calls self.Read("YOUR_NOTIFY_CODE"), which loads the full configuration for that notification from the def_notify table into the object's attributes (e.g., self._Notifyslackconnection, self._Notifydiscordsend).if hasattr(self, "_Notifyslackconnection") and self._Notifyslackconnection:).ObjNotifySlack) and calls its transmit_message method, passing the message and the destination from the configuration.# This will immediately send a message to all channels configured
# for the 'NEW_USER_ALERT' notification code.
notifier = ObjNotify()
notifier.Run("NEW_USER_ALERT", "A new user has just registered: John Doe")
deliver Method)This flow is designed for asynchronous, batched delivery. It's perfect for high-volume notifications or for situations where you want to ensure delivery even if there's a temporary network issue.
stage_notify table. This record includes the NotifyCode, the target User, and the Message.deliver() method periodically (e.g., by running python ObjNotify.py deliver).deliver method queries the stage_notify table for any messages that have not yet been sent for a specific channel (e.g., where EventSlackSent is NULL).transmit_message to send it.stage_notify, setting the appropriate flag (e.g., EventSlackSent = now()) to ensure the message is not sent again on the next run.This queued approach decouples the message generation from the delivery, making the system more resilient and scalable.
ObjNotify ClassHandles the sending and management of notifications across multiple channels.
__init__(self, db_connection: Any = 0) -> NoneInitializes the ObjNotify object. Sets up default notification modes and ensures necessary database columns and a default SYSADMIN notification exist.
db_connection: An optional database connection object.Example:
notifier = ObjNotify(db_connection)
Modify(self) -> NonePlaceholder for future modifications to the notification object.
Example:
notifier = ObjNotify(db_connection)
notifier.Modify()
Read(self, notify_code: str = "") -> NoneReads notification configuration from the def_notify table based on the provided notify_code. If the code is not found, a default entry is created.
notify_code: The unique code identifying the notification configuration.Example:
notifier = ObjNotify(db_connection)
notifier.Read("SYSTEM_ALERT")
mark_read(self, note_id: str = "", user_code: str = "") -> NoneMarks a notification as read in the database.
note_id: The ID of the notification to mark as read.user_code: The user for whom the notification should be marked as read.Example:
notifier = ObjNotify(db_connection)
notifier.mark_read("some-note-id", "john.doe")
Update(self) -> NoneUpdates the notification object's configuration in the database.
Example:
notifier = ObjNotify(db_connection)
notifier.Read("SYSTEM_ALERT")
notifier._Notifydiscordconnection = "#new-alerts"
notifier.Update()
deliver(self) -> NoneProcesses and delivers pending notifications from the stage_notify table to their configured channels. Handles different notification modes (LOCAL, REMOTE, PROXY, GUI) and prevents message loops in PROXY mode.
Example:
notifier = ObjNotify(db_connection)
notifier.deliver()
Run(self, notify_code: str, message_text: str = "") -> NoneTriggers a notification event based on a notification code and message. This method handles the logic for LOCAL, REMOTE, and PROXY modes, staging messages for users or sending directly to channels.
notify_code: The unique code identifying the notification configuration.message_text: The message content to be sent.Example:
notifier = ObjNotify(db_connection)
notifier.Run("USER_LOGIN", "User 'admin' logged in successfully.")
Select(self, selection_id: str, message_text: str = "") -> NoneAllows for interactive selection of a notification to run from a list of available options.
selection_id: An identifier for the selection process.message_text: The message content to be used with the selected notification.Example:
notifier = ObjNotify(db_connection)
notifier.Select("interactive_notification", "This is an interactive message.")
read_slack(self, channel_name: str = "", message_limit: int = 10) -> NoneReads recent messages from a specified Slack channel.
channel_name: The name of the Slack channel to read from.message_limit: The maximum number of messages to retrieve.Example:
notifier = ObjNotify(db_connection)
notifier.read_slack("#general", message_limit=5)
test_discord(self, message_text: str) -> NoneSends a direct Discord message for testing purposes.
message_text: The message content to send to Discord.Example:
notifier = ObjNotify(db_connection)
notifier.test_discord("This is a test Discord message.")
test_slack(self, message_text: str, channel_name: str = "") -> NoneSends a direct Slack message for testing purposes.
message_text: The message content to send to Slack.channel_name: The name of the Slack channel to send the message to.Example:
notifier = ObjNotify(db_connection)
notifier.test_slack("This is a test Slack message.", "#dev-alerts")
The module also provides a command-line interface using typer for various testing and service operations:
test_mqtt(message_text: str, topic: str = typer.Option("", "--topic", help="The MQTT topic to publish to.")) -> NoneSends a test MQTT message.
message_text: The message content to publish.topic: The MQTT topic to publish to.Example:
python ObjNotify.py test-mqtt "Hello MQTT!" --topic "test/notifications"
test_sms(msisdn: str, message_text: str) -> NoneSends a test SMS message.
msisdn: The recipient's mobile number.message_text: The message content to send via SMS.Example:
python ObjNotify.py test-sms "27821234567" "Test SMS from Axion."
test_email(email: str, message_text: str) -> NoneSends a test Email message.
email: The recipient's email address.message_text: The message content to send via email.Example:
python ObjNotify.py test-email "test@example.com" "Test Email from Axion."
test_module(message_text: str) -> NoneTests the full notification module by running a SYSADMIN notification and then delivering pending messages.
message_text: The message content for the test notification.Example:
python ObjNotify.py test-module "System test notification."
test_slack_image(message_text: str, image_path: str = typer.Option("resource.images/package.reference/squid.jpg", "--image", help="Path to the image file."), channel_name: str = typer.Option("", "--channel", help="The Slack channel to send the message to.")) -> NoneSends a test Slack message with an image attachment.
message_text: The message content.image_path: Path to the image file.channel_name: The Slack channel to send the message to.Example:
python ObjNotify.py test-slack-image "Check out this image!" --image "path/to/my_image.png" --channel "#general"
test_discord_image(message_text: str, image_path: str = typer.Option("resource.images/package.reference/squid.jpg", "--image", help="Path to the image file.")) -> NoneSends a test Discord message with an image attachment.
message_text: The message content.image_path: Path to the image file.Example:
python ObjNotify.py test-discord-image "Here's a cool picture!" --image "path/to/another_image.jpg"
test_mqtt_image(topic: str, message_text: str, image_path: str = typer.Option("resource.images/package.reference/squid.jpg", "--image", help="Path to the image file.")) -> NoneSends a test MQTT message with an image attachment.
topic: The MQTT topic to publish to.message_text: The message content.image_path: Path to the image file.Example:
python ObjNotify.py test-mqtt-image "image/alerts" "New image alert!" --image "path/to/alert_image.gif"
test_status_image(channel_name: str = typer.Option("", "--channel", help="The Slack channel to send the image to.")) -> NoneGenerates a system status dashboard and sends it to Slack.
channel_name: The Slack channel to send the image to.Example:
python ObjNotify.py test-status-image --channel "#status-updates"
ObjNotify integrates with the ObjAlertIncident module to provide
automatic communication tracking for all incident notifications.
When notifications are sent through ObjAlertIncident.notify_incident(),
the system sets incident context via private attributes:
_Incidentguid: The incident GUID being notified about_Incidentsentby: The user sending the notificationThis context enables automatic features:
Every incident notification receives a unique, human-readable reference:
Format: REF-YYYYMMDD-XXX
REF: PrefixYYYYMMDD: Date (e.g., 20251229)XXX: 3-character random code (A-Z, 0-9)Example: REF-20251229-A7K
The reference is:
[Ref: REF-20251229-A7K]data_incident_communication tableWhen _Incidentguid is set, all notifications are automatically logged:
Logged Information:
Database Table: data_incident_communication
from factory.core import ObjAlertIncident
# High-level incident notification
incident = ObjAlertIncident.ObjAlertIncident()
stats = incident.notify_incident(
incident_guid="incident-123",
notify_code="INCIDENT_CRITICAL",
sent_by="SYSTEM"
)
# Behind the scenes:
# 1. Creates ObjNotify instance
# 2. Sets _Incidentguid = "incident-123"
# 3. Sets _Incidentsentby = "SYSTEM"
# 4. Calls deliver() for each channel
# 5. For each delivery:
# - Generates unique short reference
# - Embeds reference in message
# - Sends notification
# - Logs communication to database
# 6. Returns: {"sent": 3, "failed": 0}
When incident context is present, notifications automatically include
the short reference:
CRITICAL INCIDENT ALERT
=======================
Database Connection Pool Exhausted
Status: INVESTIGATING
Package: DATABASE
Assigned: dba-team
[Ref: REF-20251229-A7K]
This allows recipients to:
Use the short reference to retrieve communication and incident details:
from factory.core import ObjAlertIncident
incident = ObjAlertIncident.ObjAlertIncident()
# Look up by reference
comm = incident.get_communication_by_ref("REF-20251229-A7K")
if comm:
print(f"Channel: {comm['Channel']}")
print(f"Status: {comm['Status']}")
print(f"Incident: {comm['IncidentTitle']}")
print(f"Severity: {comm['IncidentSeverity']}")
You can also manually log communications for incidents:
# Example: Log a phone call
comm_guid, short_ref = incident.log_communication(
incident_guid="incident-123",
channel="PHONE",
direction="OUTBOUND",
message="Called on-call engineer to escalate",
recipient="+1-555-0123",
status="SENT",
metadata={"duration_seconds": 180, "answered": True}
)
print(f"Logged communication: {short_ref}")
All channels support incident integration:
| Channel | External ID Tracked | Notes |
|---|---|---|
| PAGERDUTY | dedup_key | Incident correlation |
| SLACK | message_ts | Thread tracking |
| DISCORD | message_id | Message reference |
| SMS | provider_id | SMS gateway ID |
| message_id | WhatsApp API ID | |
| DBUS | N/A | Local notifications |
| MQTT | N/A | Publish-only |
from factory.core import ObjAlertIncident
incident = ObjAlertIncident.ObjAlertIncident()
# Get all communications for an incident
comms = incident.get_incident_communications("incident-123")
# Filter by channel
slack_comms = incident.get_incident_communications(
"incident-123",
channel="SLACK"
)
# Paginate results
recent = incident.get_incident_communications(
"incident-123",
limit=10
)
# Iterate through communications
for comm in comms:
print(f"{comm['ShortRef']}: {comm['Channel']} - {comm['Status']}")
print(f" To: {comm['Recipient']}")
print(f" At: {comm['Timestamp']}")
The notify_incident() method returns delivery statistics:
stats = incident.notify_incident(
incident_guid="incident-123",
notify_code="INCIDENT_CRITICAL"
)
print(f"Channels attempted: {stats['sent'] + stats['failed']}")
print(f"Successfully sent: {stats['sent']}")
print(f"Failed: {stats['failed']}")
Always use notify_incident() for incident-related notifications
Include short references when communicating with teams
Log manual communications (phone calls, in-person discussions)
Use metadata for additional context
The ObjNotify module has comprehensive test coverage for notification delivery across multiple channels and modes.
| Category | Tests | Purpose |
|---|---|---|
| Configuration & Reading | 4 | Notification code reading, admin normalization, default creation |
| Run Method - Modes | 4 | REMOTE, PROXY, GUI, LOCAL mode behaviors |
| Run Method - Users | 6 | Direct users, groups, ALL group, inactive filtering, placeholders |
| Run Method - Channels | 5 | Slack, MQTT, Discord, DBus delivery when no users configured |
| Run Method - Edge Cases | 3 | Empty codes, None messages, long messages, special characters, deduplication |
| Deliver Method | 3 | Statistics, all notify types, GUI/proxy modes, caching |
| Status Formatting | 3 | Simple JSON, nested JSON, invalid JSON handling |
| Test Commands | 8 | MQTT, SMS, Discord, Slack testing with various parameters |
| Security | 3 | SQL injection protection in Read, Run, and deliver methods |
| Constants | 4 | Constant definitions and usage validation |
| Full Integration | 2 | Complete notification flow, database connection |
| 3 | WhatsApp delivery, missing MSISDN, error handling | |
| Incident Integration | 3 | Communication logging, exception handling |
| Status Commands | 2 | Status message formatting and delivery |
| Mark Read | 3 | Note ID marking, existing notecode, user code skipping |
| PagerDuty | 2 | Cached notify codes, default severity |
| Slack Reading | 3 | Message fetching, default channel, empty messages |
| Channel Testing | 3 | Discord, Slack channel tests |
| SMS | 2 | Incident GUID handling |
| Update/Select | 3 | SQL execution, notification selection |
| Total | 77 (76 passed, 1 skipped) |
# Run all ObjNotify tests
pytest resource.test/pytests/factory.core/test_ObjNotify.py -v
# Run specific test category
pytest resource.test/pytests/factory.core/test_ObjNotify.py -v -k "test_run"
pytest resource.test/pytests/factory.core/test_ObjNotify.py -v -k "test_deliver"
pytest resource.test/pytests/factory.core/test_ObjNotify.py -v -k "test_sql_injection"