Coverage for polar/integrations/open_collective/service.py: 50%

1import contextlib 

2import dataclasses 

3from collections.abc import AsyncGenerator 

4from typing import Any 

5 

6import httpx 

7 

8from polar.config import settings 

9from polar.exceptions import PolarError 

10 

11 

12@dataclasses.dataclass 

13class OpenCollectiveCollective: 

14 slug: str 

15 isActive: bool 

16 isApproved: bool 

17 isArchived: bool 

18 isFrozen: bool 

19 host_slug: str | None = None 

20 

21 @property 

22 def is_eligible(self) -> bool: 

23 return ( 

24 self.isActive 

25 and self.isApproved 

26 and not self.isArchived 

27 and not self.isFrozen 

28 and self.host_slug == "opensource" 

29 ) 

30 

31 

32class OpenCollectiveServiceError(PolarError): 

33 pass 

34 

35 

36class OpenCollectiveAPIError(OpenCollectiveServiceError): 

37 pass 

38 

39 

40class CollectiveNotFoundError(OpenCollectiveServiceError): 

41 def __init__(self, slug: str): 

42 super().__init__(f"No collective found with slug {slug}.", 400) 

43 

44 

45class OpenCollectiveService: 

46 def __init__(self, personal_token: str | None = None) -> None: 

47 self.personal_token = personal_token 

48 

49 def create_dashboard_link(self, slug: str) -> str: 

50 return f"https://opencollective.com/{slug}" 

51 

52 async def get_collective(self, slug: str) -> OpenCollectiveCollective: 

53 async with self._get_graphql_client() as client: 

54 query = """ 

55 query GetCollective($slug: String!) { 

56 collective(slug: $slug) { 

57 slug 

58 host { 

59 slug 

60 } 

61 isActive 

62 isApproved 

63 isArchived 

64 isFrozen 

65 } 

66 } 

67 """ 

68 

69 response = await client.post( 

70 "/", 

71 json={ 

72 "query": query, 

73 "operationName": "GetCollective", 

74 "variables": {"slug": slug}, 

75 }, 

76 ) 

77 

78 try: 

79 response.raise_for_status() 

80 except httpx.HTTPStatusError as e: 

81 raise OpenCollectiveAPIError(str(e)) from e 

82 

83 json = response.json() 

84 

85 if "errors" in json: 

86 raise CollectiveNotFoundError(slug) 

87 

88 collective: dict[str, Any] = json["data"]["collective"] 

89 host = collective.pop("host") 

90 host_slug = host["slug"] if host is not None else None 

91 return OpenCollectiveCollective(**collective, host_slug=host_slug) 

92 

93 @contextlib.asynccontextmanager 

94 async def _get_graphql_client(self) -> AsyncGenerator[httpx.AsyncClient, None]: 

95 headers = {} 

96 if self.personal_token is not None: 

97 headers["Personal-Token"] = self.personal_token 

98 async with httpx.AsyncClient( 

99 base_url="https://api.opencollective.com/graphql/v2", headers=headers 

100 ) as client: 

101 yield client 

102 

103 

104open_collective = OpenCollectiveService(settings.OPEN_COLLECTIVE_PERSONAL_TOKEN)