Back

Step by Step Guide to Building a PagerDuty API Integration in Python

Aug 15, 2024 • 5 minute read

Introduction

Hey there, fellow developer! Ready to supercharge your incident management with PagerDuty's API? You're in the right place. We're going to walk through building a Python integration that'll have you managing incidents like a pro in no time.

Prerequisites

Before we dive in, make sure you've got:

  • A Python environment (3.6+ recommended)
  • A PagerDuty account with an API key (if you don't have one, hop over to PagerDuty and grab it)

Setting up the project

Let's get our hands dirty:

mkdir pagerduty_integration cd pagerduty_integration pip install requests

Authentication

First things first, let's set up our authentication:

import requests API_KEY = 'your_api_key_here' BASE_URL = 'https://api.pagerduty.com' headers = { 'Authorization': f'Token token={API_KEY}', 'Accept': 'application/vnd.pagerduty+json;version=2' }

Basic API Requests

Now, let's fetch some incidents:

response = requests.get(f'{BASE_URL}/incidents', headers=headers) incidents = response.json()['incidents']

Creating an incident? Easy peasy:

payload = { "incident": { "type": "incident", "title": "The server is on fire!", "service": { "id": "YOUR_SERVICE_ID", "type": "service_reference" } } } response = requests.post(f'{BASE_URL}/incidents', json=payload, headers=headers)

Handling Responses

Always check your responses:

if response.status_code == 201: print("Incident created successfully!") else: print(f"Oops! Something went wrong: {response.text}")

Advanced Features

Need to handle pagination? We've got you covered:

def get_all_incidents(): incidents = [] offset = 0 while True: response = requests.get(f'{BASE_URL}/incidents?offset={offset}', headers=headers) data = response.json() incidents.extend(data['incidents']) if not data['more']: break offset += data['limit'] return incidents

Webhooks

Setting up a webhook listener? Here's a quick Flask example:

from flask import Flask, request app = Flask(__name__) @app.route('/webhook', methods=['POST']) def webhook(): data = request.json # Process your webhook data here return '', 204 if __name__ == '__main__': app.run(port=5000)

Best Practices

Remember to respect rate limits and log everything. Your future self will thank you!

import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Use this instead of print statements logger.info("Incident created successfully!")

Testing

Don't forget to test! Here's a quick unittest example:

import unittest from unittest.mock import patch from your_module import create_incident class TestPagerDutyIntegration(unittest.TestCase): @patch('requests.post') def test_create_incident(self, mock_post): mock_post.return_value.status_code = 201 result = create_incident("Test incident") self.assertTrue(result) if __name__ == '__main__': unittest.main()

Deployment

When you're ready to deploy, consider using environment variables for your API key and other sensitive info. And maybe wrap your integration in a nice Docker container for easy deployment?

Conclusion

And there you have it! You're now armed with the knowledge to build a robust PagerDuty integration. Remember, the API docs are your friend, so don't be shy about diving deeper. Happy coding, and may your incidents be few and far between!