Greeting my lovely Community,
This time I need your help :). Is there any possibility to test connectivity between notification channel and PagerDuty by triggering "TEST CONNECTION" via Python ? I'm writing Cloud Function to check deployment and one of the task is verify connectivity between project and PagerDuty service. I'm able to create notification channels via code, but each time I'm creating new PD channel, I'm forced to manually test connection which is not sufficient for me. Any advice on that ?
cheers,
DamianS
Solved! Go to Solution.
Absolutely! Here's how you can test the connection between your notification channel and PagerDuty using Python in your Cloud Function:
1. Using pdpyras library:
The recommended approach is to leverage the pdpyras library, a popular Python client for interacting with the PagerDuty API. Here's how:
from pdpyras import PagerDuty # Replace with your PagerDuty API key api_key = "YOUR_PAGERDUTY_API_KEY" # Initialize PagerDuty client client = PagerDuty(api_key=api_key) # Get the notification channel object notification_channel = client.get_notification_channel(id="CHANNEL_ID") try: # Attempt to list events for the channel (empty list on success) events = notification_channel.list_events(limit=1) print("Connection to PagerDuty successful!") except pdpyras.exceptions.APIError as e: print(f"Error testing connection: {e}")
Explanation:
2. Using Events API v2:
While not officially designed for testing connections, you can use the Events API v2 to send a dummy event and check for successful reception:
import requests # Replace with your PagerDuty API key api_key = "YOUR_PAGERDUTY_API_KEY" integration_key = "YOUR_INTEGRATION_KEY" # Key from your notification channel settings # Event data event_data = { "service_key": integration_key, "event_type": "trigger", "description": "Test connection from Cloud Function", } # Set headers headers = { "Authorization": f"Token token={api_key}", "Content-Type": "application/json", } # Send event response = requests.post("https://events.pagerduty.com/v2/enqueue", json=event_data, headers=headers) if response.status_code == 202: print("Connection to PagerDuty successful!") else: print(f"Error testing connection: {response.text}")
Explanation:
Important Note:
While these methods test basic connectivity, they don't simulate the actual notification delivery process. For a more thorough test, consider setting up a test incident in PagerDuty and triggering a notification through your Cloud Function.
Choose the method that best suits your needs. The pdpyras approach is generally recommended for a more robust and official way to interact with the API.
Hello @pgstorm148 ,
I had to adjust your API sample because you're using v2 of the Events API endpoint but are sending data structured for v1.
import requests
# Replace with your PagerDuty API key
api_key = "API_KEY"
integration_key = "PD_SERVICE_KEY" # Key from your notification channel settings
# Define the routing key using the integration key
routing_key = integration_key
# Event data for v2
event_data = {
"routing_key": routing_key,
"event_action": "trigger",
"payload": {
"summary": "Test connection from Cloud Function",
"source": "Cloud Function Test",
"severity": "info",
}
}
# Set headers
headers = {
"Content-Type": "application/json",
}
# Send event
response = requests.post("https://events.pagerduty.com/v2/enqueue", json=event_data, headers=headers)
if response.status_code == 202:
print("Connection to PagerDuty successful!")
else:
print(f"Error testing connection: {response.text}")
This code works like a charm.
cheers,
DamianS
Absolutely! Here's how you can test the connection between your notification channel and PagerDuty using Python in your Cloud Function:
1. Using pdpyras library:
The recommended approach is to leverage the pdpyras library, a popular Python client for interacting with the PagerDuty API. Here's how:
from pdpyras import PagerDuty # Replace with your PagerDuty API key api_key = "YOUR_PAGERDUTY_API_KEY" # Initialize PagerDuty client client = PagerDuty(api_key=api_key) # Get the notification channel object notification_channel = client.get_notification_channel(id="CHANNEL_ID") try: # Attempt to list events for the channel (empty list on success) events = notification_channel.list_events(limit=1) print("Connection to PagerDuty successful!") except pdpyras.exceptions.APIError as e: print(f"Error testing connection: {e}")
Explanation:
2. Using Events API v2:
While not officially designed for testing connections, you can use the Events API v2 to send a dummy event and check for successful reception:
import requests # Replace with your PagerDuty API key api_key = "YOUR_PAGERDUTY_API_KEY" integration_key = "YOUR_INTEGRATION_KEY" # Key from your notification channel settings # Event data event_data = { "service_key": integration_key, "event_type": "trigger", "description": "Test connection from Cloud Function", } # Set headers headers = { "Authorization": f"Token token={api_key}", "Content-Type": "application/json", } # Send event response = requests.post("https://events.pagerduty.com/v2/enqueue", json=event_data, headers=headers) if response.status_code == 202: print("Connection to PagerDuty successful!") else: print(f"Error testing connection: {response.text}")
Explanation:
Important Note:
While these methods test basic connectivity, they don't simulate the actual notification delivery process. For a more thorough test, consider setting up a test incident in PagerDuty and triggering a notification through your Cloud Function.
Choose the method that best suits your needs. The pdpyras approach is generally recommended for a more robust and official way to interact with the API.
Hello @pgstorm148 ,
Let me check this and back to you. Thanks for reply 🙂
cheers,
DamianS
Hello @pgstorm148 ,
I had to adjust your API sample because you're using v2 of the Events API endpoint but are sending data structured for v1.
import requests
# Replace with your PagerDuty API key
api_key = "API_KEY"
integration_key = "PD_SERVICE_KEY" # Key from your notification channel settings
# Define the routing key using the integration key
routing_key = integration_key
# Event data for v2
event_data = {
"routing_key": routing_key,
"event_action": "trigger",
"payload": {
"summary": "Test connection from Cloud Function",
"source": "Cloud Function Test",
"severity": "info",
}
}
# Set headers
headers = {
"Content-Type": "application/json",
}
# Send event
response = requests.post("https://events.pagerduty.com/v2/enqueue", json=event_data, headers=headers)
if response.status_code == 202:
print("Connection to PagerDuty successful!")
else:
print(f"Error testing connection: {response.text}")
This code works like a charm.
cheers,
DamianS