Coverage for polar/integrations/discord/client.py: 31%

1from typing import Any, Literal 1 ctx1a

2 

3import httpx 1 ctx1a

4import structlog 1 ctx1a

5 

6from polar.config import settings 1 ctx1a

7 

8log = structlog.get_logger() 1 ctx1a

9 

10BASE_URL = "https://discord.com/api/v10" 1 ctx1a

11 

12 

13class DiscordClient: 1 ctx1a

14 def __init__(self, scheme: Literal["Bot", "Bearer"], token: str) -> None: 1 ctx1a

15 self.client = httpx.AsyncClient( 1 ctx1a

16 base_url=BASE_URL, 

17 headers={"Authorization": f"{scheme} {token}"}, 

18 ) 

19 

20 async def get_me(self) -> dict[str, Any]: 1 ctx1a

21 response = await self.client.get("/users/@me") 

22 self._handle_response(response) 

23 return response.json() 

24 

25 async def get_guild(self, id: str) -> dict[str, Any]: 1 ctx1a

26 response = await self.client.get(f"/guilds/{id}") 

27 self._handle_response(response) 

28 

29 return response.json() 

30 

31 async def add_member( 1 ctx1a

32 self, 

33 guild_id: str, 

34 discord_user_id: str, 

35 discord_user_access_token: str, 

36 role_id: str, 

37 nick: str | None = None, 

38 ) -> None: 

39 endpoint = f"/guilds/{guild_id}/members/{discord_user_id}" 

40 

41 data: dict[str, Any] = {} 

42 data["access_token"] = discord_user_access_token 

43 data["roles"] = [role_id] 

44 if nick: 

45 data["nick"] = nick 

46 

47 response = await self.client.put(endpoint, json=data) 

48 self._handle_response(response) 

49 

50 if response.status_code == 201: 

51 log.info( 

52 "discord.add_member.success", 

53 guild_id=guild_id, 

54 discord_user_id=discord_user_id, 

55 ) 

56 return 

57 

58 log.debug( 

59 "discord.add_member.already_present", 

60 guild_id=guild_id, 

61 discord_user_id=discord_user_id, 

62 ) 

63 await self.add_member_role( 

64 guild_id=guild_id, 

65 discord_user_id=discord_user_id, 

66 role_id=role_id, 

67 ) 

68 

69 async def add_member_role( 1 ctx1a

70 self, 

71 guild_id: str, 

72 discord_user_id: str, 

73 role_id: str, 

74 ) -> None: 

75 endpoint = f"/guilds/{guild_id}/members/{discord_user_id}/roles/{role_id}" 

76 

77 response = await self.client.put(endpoint) 

78 self._handle_response(response) 

79 

80 log.info( 

81 "discord.add_member_role.success", 

82 guild_id=guild_id, 

83 discord_user_id=discord_user_id, 

84 role_id=role_id, 

85 ) 

86 return None 

87 

88 async def remove_member_role( 1 ctx1a

89 self, 

90 guild_id: str, 

91 discord_user_id: str, 

92 role_id: str, 

93 ) -> None: 

94 endpoint = f"/guilds/{guild_id}/members/{discord_user_id}/roles/{role_id}" 

95 

96 response = await self.client.delete(endpoint) 

97 self._handle_response(response) 

98 

99 log.info( 

100 "discord.remove_member_role.success", 

101 guild_id=guild_id, 

102 discord_user_id=discord_user_id, 

103 role_id=role_id, 

104 ) 

105 return None 

106 

107 async def remove_member(self, guild_id: str, discord_user_id: str) -> None: 1 ctx1a

108 endpoint = f"/guilds/{guild_id}/members/{discord_user_id}" 

109 

110 response = await self.client.delete(endpoint) 

111 self._handle_response(response) 

112 

113 log.info( 

114 "discord.remove_member.success", 

115 guild_id=guild_id, 

116 discord_user_id=discord_user_id, 

117 ) 

118 return None 

119 

120 def _handle_response(self, response: httpx.Response) -> httpx.Response: 1 ctx1a

121 response.raise_for_status() 

122 return response 

123 

124 

125bot_client = DiscordClient("Bot", settings.DISCORD_BOT_TOKEN) 1 ctx1a

126 

127__all__ = ["DiscordClient", "bot_client"] 1 ctx1a