107 lines
3.2 KiB
Python
107 lines
3.2 KiB
Python
import logging
|
|
|
|
import httpx
|
|
|
|
from glitchup_bot.config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class GlitchTipClient:
|
|
def __init__(self) -> None:
|
|
self.base_url = settings.glitchtip_url.rstrip("/")
|
|
self.headers = {"Authorization": f"Bearer {settings.glitchtip_api_token}"}
|
|
self._client: httpx.AsyncClient | None = None
|
|
|
|
async def _get_client(self) -> httpx.AsyncClient:
|
|
if self._client is None:
|
|
self._client = httpx.AsyncClient(
|
|
base_url=self.base_url,
|
|
headers=self.headers,
|
|
timeout=30,
|
|
)
|
|
return self._client
|
|
|
|
async def _get(self, path: str, params: dict | None = None) -> list | dict:
|
|
client = await self._get_client()
|
|
response = await client.get(path, params=params)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def _get_paginated(self, path: str, params: dict | None = None) -> list:
|
|
results: list = []
|
|
base_params = params or {}
|
|
base_params.setdefault("limit", 100)
|
|
cursor: str | None = None
|
|
client = await self._get_client()
|
|
|
|
while True:
|
|
request_params = dict(base_params)
|
|
if cursor:
|
|
request_params["cursor"] = cursor
|
|
|
|
response = await client.get(path, params=request_params)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
results.extend(data)
|
|
|
|
next_cursor = self._parse_next_cursor(response.headers.get("link", ""))
|
|
if not next_cursor or not data:
|
|
break
|
|
cursor = next_cursor
|
|
|
|
return results
|
|
|
|
@staticmethod
|
|
def _parse_next_cursor(link_header: str) -> str | None:
|
|
for part in link_header.split(","):
|
|
if 'rel="next"' not in part or 'results="true"' not in part or "cursor=" not in part:
|
|
continue
|
|
|
|
start = part.index("cursor=") + len("cursor=")
|
|
end = part.find(">", start)
|
|
return part[start:end] if end != -1 else part[start:]
|
|
|
|
return None
|
|
|
|
async def list_projects(self) -> list[dict]:
|
|
return await self._get_paginated(
|
|
f"/api/0/organizations/{settings.glitchtip_org_slug}/projects/"
|
|
)
|
|
|
|
async def list_issues(
|
|
self, project_slug: str, query: str = "is:unresolved", sort: str = "date"
|
|
) -> list[dict]:
|
|
return await self._get_paginated(
|
|
f"/api/0/projects/{settings.glitchtip_org_slug}/{project_slug}/issues/",
|
|
params={"query": query, "sort": sort},
|
|
)
|
|
|
|
async def get_issue(self, issue_id: int) -> dict:
|
|
return await self._get(f"/api/0/issues/{issue_id}/")
|
|
|
|
async def close(self) -> None:
|
|
if self._client is not None:
|
|
await self._client.aclose()
|
|
self._client = None
|
|
|
|
|
|
glitchtip_client: GlitchTipClient | None = None
|
|
|
|
|
|
def get_glitchtip_client() -> GlitchTipClient:
|
|
global glitchtip_client
|
|
|
|
if glitchtip_client is None:
|
|
glitchtip_client = GlitchTipClient()
|
|
|
|
return glitchtip_client
|
|
|
|
|
|
async def close_glitchtip_client() -> None:
|
|
global glitchtip_client
|
|
|
|
if glitchtip_client is not None:
|
|
await glitchtip_client.close()
|
|
glitchtip_client = None
|