-
Notifications
You must be signed in to change notification settings - Fork 35
/
sentry_api_client.py
91 lines (75 loc) · 3.1 KB
/
sentry_api_client.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
from __future__ import annotations
import os
import requests
from dotenv import load_dotenv
from datetime import datetime
from src import app
from src.database import db_session
from src.models import Organization, SentryInstallation
load_dotenv()
class SentryAPIClient:
def __init__(self, token):
self.token = token
@staticmethod
def get_sentry_api_token(organization: Organization) -> str:
"""
Fetches an organization's Sentry API token, refreshing it if necessary.
"""
sentry_installation = SentryInstallation.query.filter(
SentryInstallation.organization_id == organization.id
).first()
# If the token is not expired, no need to refresh it
if sentry_installation.expires_at.timestamp() > datetime.now().timestamp():
return sentry_installation.token
# If the token is expired, we'll need to refresh it...
app.logger.info(
f"Token for {sentry_installation.org_slug} has expired. Refreshing..."
)
# Construct a payload to ask Sentry for a new token
payload = {
"grant_type": "refresh_token",
"refresh_token": sentry_installation.refresh_token,
"client_id": os.getenv("SENTRY_CLIENT_ID"),
"client_secret": os.getenv("SENTRY_CLIENT_SECRET"),
}
# Send that payload to Sentry and parse the response
token_response = requests.post(
url=(
f"{os.getenv('SENTRY_URL')}/api/0/sentry-app-installations/"
f"{sentry_installation.uuid}/authorizations/"
),
json=payload,
).json()
# Store the token information for future requests
sentry_installation.token = token_response["token"]
sentry_installation.refresh_token = token_response["refreshToken"]
sentry_installation.expires_at = token_response["expiresAt"]
db_session.commit()
app.logger.info(
f"Token for '{sentry_installation.org_slug}' has been refreshed."
)
# Return the newly refreshed token
return sentry_installation.token
# We create a static wrapper on the constructor to ensure our token is always refreshed
@staticmethod
def create(organization: Organization) -> "SentryAPIClient":
token = SentryAPIClient.get_sentry_api_token(organization)
return SentryAPIClient(token)
def request(
self, method: str, path: str, data: dict | None = None
) -> requests.Response:
response = requests.request(
method=method,
url=f"{os.getenv('SENTRY_URL')}/api/0{path}",
headers={"Authorization": f"Bearer {self.token}"},
data=data,
)
try:
response.raise_for_status()
except requests.exceptions.HTTPError as e:
# TODO(you): Catch these sorta errors in Sentry!
app.logger.error(f"Error while making a request to Sentry: {e}")
return response
def get(self, path: str) -> requests.Response:
return self.request("GET", path)
# TODO(you): Extend as you see fit!