Coverage for polar/integrations/discord/client.py: 31%
54 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 15:52 +0000
1from typing import Any, Literal 1a
3import httpx 1a
4import structlog 1a
6from polar.config import settings 1a
8log = structlog.get_logger() 1a
10BASE_URL = "https://discord.com/api/v10" 1a
13class DiscordClient: 1a
14 def __init__(self, scheme: Literal["Bot", "Bearer"], token: str) -> None: 1a
15 self.client = httpx.AsyncClient( 1a
16 base_url=BASE_URL,
17 headers={"Authorization": f"{scheme} {token}"},
18 )
20 async def get_me(self) -> dict[str, Any]: 1a
21 response = await self.client.get("/users/@me")
22 self._handle_response(response)
23 return response.json()
25 async def get_guild(self, id: str) -> dict[str, Any]: 1a
26 response = await self.client.get(f"/guilds/{id}")
27 self._handle_response(response)
29 return response.json()
31 async def add_member( 1a
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}"
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
47 response = await self.client.put(endpoint, json=data)
48 self._handle_response(response)
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
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 )
69 async def add_member_role( 1a
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}"
77 response = await self.client.put(endpoint)
78 self._handle_response(response)
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
88 async def remove_member_role( 1a
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}"
96 response = await self.client.delete(endpoint)
97 self._handle_response(response)
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
107 async def remove_member(self, guild_id: str, discord_user_id: str) -> None: 1a
108 endpoint = f"/guilds/{guild_id}/members/{discord_user_id}"
110 response = await self.client.delete(endpoint)
111 self._handle_response(response)
113 log.info(
114 "discord.remove_member.success",
115 guild_id=guild_id,
116 discord_user_id=discord_user_id,
117 )
118 return None
120 def _handle_response(self, response: httpx.Response) -> httpx.Response: 1a
121 response.raise_for_status()
122 return response
125bot_client = DiscordClient("Bot", settings.DISCORD_BOT_TOKEN) 1a
127__all__ = ["DiscordClient", "bot_client"] 1a