52 lines
1.3 KiB
Python
52 lines
1.3 KiB
Python
import os
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
GITEA_URL = os.getenv("GITEA_URL", "http://127.0.0.1:3000").rstrip("/")
|
|
GITEA_TOKEN = os.getenv("GITEA_TOKEN", "")
|
|
|
|
|
|
def _auth() -> dict[str, str]:
|
|
headers: dict[str, str] = {"Accept": "application/json"}
|
|
if GITEA_TOKEN:
|
|
headers["Authorization"] = f"token {GITEA_TOKEN}"
|
|
return headers
|
|
|
|
|
|
async def fetch(path: str) -> Any:
|
|
async with httpx.AsyncClient(base_url=GITEA_URL, timeout=10) as client:
|
|
r = await client.get(f"/api/v1/{path}", headers=_auth())
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
async def current_user() -> dict:
|
|
return await fetch("user")
|
|
|
|
|
|
async def repos() -> list[dict]:
|
|
return await fetch("user/repos?limit=50")
|
|
|
|
|
|
async def issues() -> list[dict]:
|
|
return await fetch("user/issues?limit=50&type=all")
|
|
|
|
|
|
async def pull_requests() -> list[dict]:
|
|
return await fetch("user/pulls?limit=50")
|
|
|
|
|
|
async def activity_events() -> list[dict]:
|
|
user = await current_user()
|
|
events = await fetch(f"users/{user['login']}/activities/feeds?limit=20")
|
|
return [
|
|
{
|
|
"type": event.get("op_type", "activity"),
|
|
"actor": event.get("act_user") or {},
|
|
"repo": event.get("repo") or {},
|
|
"created_at": event.get("created", ""),
|
|
}
|
|
for event in events
|
|
]
|