NOTICE: All information contained herein is, and remains
the property of TechnoCore.
The intellectual and technical concepts contained
herein are proprietary to TechnoCore and dissemination of this information or reproduction of this material
is strictly forbidden unless prior written permission is obtained
from TechnoCore.
The ObjNotifyPagerduty class provides an interface for sending incident alerts to PagerDuty using the Events API v2. It is designed to be used as part of the centralized notification system for critical alerts and incident management.
The class is configured through the main config.yaml file. The following section must be present for the class to initialize correctly:
pagerduty:
integration_key: "your-integration-key-here"
api_url: "https://events.pagerduty.com/v2/enqueue" # Optional: defaults to US region
__init__(self, db_connection: Any = 0)The constructor initializes the PagerDuty client. It reads the integration_key and optional api_url from the global configuration (config.yaml). If the integration key is missing, the client will not be initialized, and errors will be logged.
transmit_message(self, message_text: str, severity: str = "error", event_action: str = "trigger", dedup_key: str = None, custom_details: Dict[str, Any] = None)Sends an event to PagerDuty using the Events API v2.
'critical', 'error', 'warning', or 'info'. Defaults to 'error'.'trigger', 'acknowledge', or 'resolve'. Defaults to 'trigger'.Returns: The deduplication key (str) if successful, None otherwise.
trigger_incident(self, summary: str, severity: str = "error", custom_details: Dict[str, Any] = None, dedup_key: str = None)Convenience method to trigger a new PagerDuty incident.
'critical', 'error', 'warning', 'info'). Defaults to 'error'.Returns: The deduplication key for the incident.
acknowledge_incident(self, dedup_key: str)Acknowledge an existing PagerDuty incident.
Returns: The deduplication key if successful, None otherwise.
resolve_incident(self, dedup_key: str)Resolve an existing PagerDuty incident.
Returns: The deduplication key if successful, None otherwise.
If you receive authentication errors, verify that:
config.yaml is correct and properly formatted.If events are accepted but not creating incidents:
from factory.sms.package.core.ObjNotifyPagerduty import ObjNotifyPagerduty
# Instantiate the notifier
pagerduty = ObjNotifyPagerduty(DB)
# Trigger a critical incident
dedup_key = pagerduty.trigger_incident(
summary="Database connection pool exhausted on production server",
severity="critical",
custom_details={
"server": "db-prod-01",
"pool_size": 100,
"active_connections": 100,
"queue_length": 47
}
)
# Acknowledge the incident (e.g., when engineer starts working on it)
if dedup_key:
pagerduty.acknowledge_incident(dedup_key)
# Resolve the incident when fixed
if dedup_key:
pagerduty.resolve_incident(dedup_key)
# Trigger a warning-level alert
pagerduty.transmit_message(
message_text="Disk usage on server 'web-prod-02' has exceeded 80%",
severity="warning",
custom_details={
"server": "web-prod-02",
"disk_usage_percent": 85,
"mount_point": "/var/log"
}
)