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

53 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-12-05 17:15 +0000

1import contextlib 1a

2import dataclasses 1a

3from collections.abc import AsyncGenerator 1a

4from typing import Any 1a

5 

6import httpx 1a

7 

8from polar.config import settings 1a

9from polar.exceptions import PolarError 1a

10 

11 

12@dataclasses.dataclass 1a

13class OpenCollectiveCollective: 1a

14 slug: str 1a

15 isActive: bool 1a

16 isApproved: bool 1a

17 isArchived: bool 1a

18 isFrozen: bool 1a

19 host_slug: str | None = None 1a

20 

21 @property 1a

22 def is_eligible(self) -> bool: 1a

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): 1a

33 pass 1a

34 

35 

36class OpenCollectiveAPIError(OpenCollectiveServiceError): 1a

37 pass 1a

38 

39 

40class CollectiveNotFoundError(OpenCollectiveServiceError): 1a

41 def __init__(self, slug: str): 1a

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

43 

44 

45class OpenCollectiveService: 1a

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

47 self.personal_token = personal_token 1a

48 

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

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

51 

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

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 1a

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

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) 1a