c46dff09d9
# What this PR does Refactors the PagerDuty migration script to be a bit more generic + adds a migration script to migrate from Splunk OnCall (VictorOps) tldr; ```bash ❯ docker build -t oncall-migrator . [+] Building 0.4s (10/10) FINISHED ❯ docker run --rm \ -e MIGRATING_FROM="pagerduty" \ -e MODE="plan" \ -e ONCALL_API_URL="http://localhost:8080" \ -e ONCALL_API_TOKEN="<ONCALL_API_TOKEN>" \ -e PAGERDUTY_API_TOKEN="<PAGERDUTY_API_TOKEN>" \ oncall-migrator running pagerduty migration script... ❯ docker run --rm \ -e MIGRATING_FROM="splunk" \ -e MODE="plan" \ -e ONCALL_API_URL="http://localhost:8080" \ -e ONCALL_API_TOKEN="<ONCALL_API_TOKEN>" \ -e SPLUNK_API_ID="<SPLUNK_API_ID>" \ -e SPLUNK_API_KEY="<SPLUNK_API_KEY>" \ oncall-migrator migrating from splunk oncall... ``` https://www.loom.com/share/a855062d436a4ef79f030e22528d8c71 ## Checklist - [x] Unit, integration, and e2e (if applicable) tests updated - [x] Documentation added (or `pr:no public docs` PR label added if not required) - [x] Added the relevant release notes label (see labels prefixed w/ `release:`). These labels dictate how your PR will show up in the autogenerated release notes.
66 lines
1.9 KiB
Python
66 lines
1.9 KiB
Python
import requests
|
|
|
|
from lib.base_config import ONCALL_API_TOKEN, ONCALL_API_URL
|
|
from lib.network import api_call as _api_call
|
|
|
|
|
|
class OnCallAPIClient:
|
|
@classmethod
|
|
def api_call(cls, method: str, path: str, **kwargs) -> requests.Response:
|
|
return _api_call(
|
|
method,
|
|
ONCALL_API_URL,
|
|
path,
|
|
headers={"Authorization": ONCALL_API_TOKEN},
|
|
**kwargs
|
|
)
|
|
|
|
@classmethod
|
|
def list_all(cls, path: str) -> list[dict]:
|
|
response = cls.api_call("get", path)
|
|
|
|
data = response.json()
|
|
results = data["results"]
|
|
|
|
while data["next"]:
|
|
response = cls.api_call("get", data["next"])
|
|
|
|
data = response.json()
|
|
results += data["results"]
|
|
|
|
return results
|
|
|
|
@classmethod
|
|
def create(cls, path: str, payload: dict) -> dict:
|
|
response = cls.api_call("post", path, json=payload)
|
|
return response.json()
|
|
|
|
@classmethod
|
|
def delete(cls, path: str) -> None:
|
|
try:
|
|
cls.api_call("delete", path)
|
|
except requests.exceptions.HTTPError as e:
|
|
# ignore 404s on delete so deleting resources manually while running the script doesn't break it
|
|
if e.response.status_code != 404:
|
|
raise
|
|
|
|
@classmethod
|
|
def update(cls, path: str, payload: dict) -> dict:
|
|
response = cls.api_call("put", path, json=payload)
|
|
return response.json()
|
|
|
|
@classmethod
|
|
def list_users_with_notification_rules(cls):
|
|
oncall_users = cls.list_all("users")
|
|
oncall_notification_rules = cls.list_all(
|
|
"personal_notification_rules/?important=false"
|
|
)
|
|
|
|
for user in oncall_users:
|
|
user["notification_rules"] = [
|
|
rule
|
|
for rule in oncall_notification_rules
|
|
if rule["user_id"] == user["id"]
|
|
]
|
|
|
|
return oncall_users
|