Coverage for src/integry/resources/apps/api.py: 100%

1from integry.resources.base import BaseResource, AsyncPaginator 

2from .types import App, AppsPage 

3 

4 

5class Apps(BaseResource): 

6 

7 def list(self, user_id: str, cursor: str = "") -> AsyncPaginator[App, AppsPage]: 

8 """ 

9 Lists all apps. 

10 

11 Args: 

12 user_id: The ID of the user. 

13 cursor: Provide the cursor from last page to fetch the next page of apps. 

14 

15 Returns: 

16 List of apps. 

17 """ 

18 return AsyncPaginator( 

19 self, 

20 user_id, 

21 "", 

22 explicit_cursor=cursor, 

23 model=App, 

24 paginated_response_model=AppsPage, 

25 ) 

26 

27 async def get(self, app_name: str, user_id: str): 

28 """ 

29 Gets an app by name. 

30 

31 Args: 

32 app_name: The name of the app. 

33 user_id: The ID of the user. 

34 

35 Returns: 

36 The app. 

37 """ 

38 

39 response = await self.http_client.post( 

40 f"{self.name}/{app_name}/get/", 

41 headers=self._get_signed_request_headers(user_id), 

42 ) 

43 data = self._get_response_data_or_raise(response) 

44 

45 return App(**data) 

46 

47 async def is_connected(self, app_name: str, user_id: str) -> bool: 

48 """ 

49 Returns whether user has connected an account of the app. 

50 

51 Args: 

52 app_name: The name of the app. 

53 user_id: The ID of the user. 

54 

55 Returns: 

56 Whether user has connected an account of the app. 

57 """ 

58 

59 response = await self.http_client.post( 

60 f"{self.name}/{app_name}/get/", 

61 headers=self._get_signed_request_headers(user_id), 

62 ) 

63 

64 data = self._get_response_data_or_raise(response) 

65 

66 app = App(**data) 

67 return len(app.connected_accounts) > 0