Coverage for backend / app / utils / github_client.py: 100%

1from contextlib import AsyncExitStack 

2 

3import httpx 

4import stamina 

5from pydantic import SecretStr 

6 

7from app.schemas import GhCommit, GhCommitStatus 

8 

9BASE_URL = "https://api.github.com" 

10 

11 

12class GithubClientError(Exception): 

13 pass 

14 

15 

16class GithubClient: 

17 def __init__(self, token: SecretStr): 

18 self._token = token 

19 self._httpx_client: httpx.AsyncClient | None = None 

20 self._exit_stack = AsyncExitStack() 

21 

22 def ensure_initialized(self) -> None: 

23 if self._httpx_client is None: 

24 raise GithubClientError("Client not initialized") 

25 

26 def _headers(self) -> dict[str, str]: 

27 return {"Authorization": f"token {self._token.get_secret_value()}"} 

28 

29 async def __aenter__(self): 

30 if self._httpx_client is not None: 

31 raise GithubClientError("Client already initialized") 

32 client = httpx.AsyncClient(base_url=BASE_URL, headers=self._headers()) 

33 self._httpx_client = await self._exit_stack.enter_async_context(client) 

34 return self 

35 

36 async def __aexit__(self, exc_type, exc_val, exc_tb): 

37 await self._exit_stack.aclose() 

38 self._httpx_client = None 

39 

40 async def _get(self, url: str, params: dict | None = None) -> httpx.Response: 

41 self.ensure_initialized() 

42 assert self._httpx_client is not None 

43 

44 async for attempt in stamina.retry_context( 

45 on=(httpx.TransportError, httpx.HTTPStatusError), 

46 attempts=3, 

47 wait_jitter=2.0, 

48 ): 

49 with attempt: 

50 response = await self._httpx_client.get(url, params=params) 

51 if response.status_code >= 500: 

52 response.raise_for_status() 

53 response.raise_for_status() 

54 return response 

55 

56 async def get_latest_commits( 

57 self, owner: str, repo: str, limit: int = 5 

58 ) -> list[GhCommit]: 

59 response = await self._get( 

60 f"/repos/{owner}/{repo}/commits", params={"per_page": limit} 

61 ) 

62 resp_json = response.json() 

63 assert isinstance(resp_json, list) 

64 return [GhCommit.model_validate(item) for item in resp_json] 

65 

66 async def get_commit_statuses( 

67 self, owner: str, repo: str, sha: str 

68 ) -> list[GhCommitStatus]: 

69 response = await self._get(f"/repos/{owner}/{repo}/statuses/{sha}") 

70 resp_json = response.json() 

71 assert isinstance(resp_json, list) 

72 return [GhCommitStatus.model_validate(item) for item in resp_json]