feat: Implement Community Invites

CHANGELOG.md

@@ -12,6 +12,8 @@ These changes are available on the `master` branch, but have not yet been releas
 
 ### Added
 
+- Added support for community invites.
+  ([#3044](https://github.com/Pycord-Development/pycord/pull/3044))
 - Added `Member.colours` and `Member.colors` properties.
   ([#3063](https://github.com/Pycord-Development/pycord/pull/3063))
 - Added `RadioGroup`, `CheckboxGroup`, and `Checkbox` for modals.

discord/abc.py

@@ -50,6 +50,7 @@
 from .invite import Invite
 from .iterators import HistoryIterator, MessagePinIterator
 from .mentions import AllowedMentions
+from .object import Object
 from .partial_emoji import PartialEmoji, _EmojiTag
 from .permissions import PermissionOverwrite, Permissions
 from .role import Role
@@ -1208,6 +1209,8 @@ async def create_invite(
         target_type: InviteTarget | None = None,
         target_user: User | None = None,
         target_application_id: int | None = None,
+        roles: list[Role | Object] | None = None,
+        target_users_file: File | None = None,
     ) -> Invite:
         """|coro|
 
@@ -1259,6 +1262,20 @@ async def create_invite(
 
             .. versionadded:: 2.0
 
+        roles: Optional[List[Union[:class:`.Role`, :class:`.Object`]]]
+            The roles to give a user when joining through this invite.
+
+            You must have the :attr:`~Permissions.manage_roles` permission to do this and roles cannot be higher than your own.
+
+            .. versionadded:: 2.8
+
+        target_users_file: Optional[:class:`File`]
+            A CSV file with a single column of user IDs for all the users able to accept this invite.
+
+            You can use :func:`utils.users_to_csv` to generate a virtual CSV file from a sequence of user IDs.
+
+            .. versionadded:: 2.8
+
         Returns
         -------
         :class:`~discord.Invite`
@@ -1283,8 +1300,11 @@ async def create_invite(
             target_type=target_type.value if target_type else None,
             target_user_id=target_user.id if target_user else None,
             target_application_id=target_application_id,
+            roles=[str(r.id) for r in roles] if roles else None,
+            target_users_file=target_users_file,
         )
         invite = Invite.from_incomplete(data=data, state=self._state)
+
         if target_event:
             invite.set_scheduled_event(target_event)
         return invite

discord/enums.py

@@ -84,6 +84,7 @@
     "SubscriptionStatus",
     "SeparatorSpacingSize",
     "SelectDefaultValueType",
+    "InviteTargetUsersJobStatusCode",
 )
 
 
@@ -1133,6 +1134,13 @@ class SelectDefaultValueType(Enum):
     user = "user"
 
 
+class InviteTargetUsersJobStatusCode(Enum):
+    unspecified = 0
+    processing = 1
+    completed = 2
+    failed = 3
+
+
 T = TypeVar("T")
 
 

discord/http.py

@@ -92,6 +92,9 @@
         welcome_screen,
         widget,
     )
+    from .types.invite import (
+        InviteTargetUsersJobStatus as InviteTargetUsersJobStatusPayload,
+    )
     from .types.snowflake import Snowflake, SnowflakeList
     from .types.soundboard import SoundboardSound as SoundboardSoundPayload
 
@@ -103,16 +106,19 @@
 API_VERSION: int = 10
 
 
-async def json_or_text(response: aiohttp.ClientResponse) -> dict[str, Any] | str:
-    text = await response.text(encoding="utf-8")
+async def parse_response(
+    response: aiohttp.ClientResponse,
+) -> dict[str, Any] | str | bytes:
     try:
         if response.headers["content-type"] == "application/json":
-            return utils._from_json(text)
+            return utils._from_json(await response.text(encoding="utf-8"))
+        elif response.headers["content-type"] == "text/csv":
+            return await response.read()
     except KeyError:
         # Thanks Cloudflare
         pass
 
-    return text
+    return await response.text(encoding="utf-8")
 
 
 class Route:
@@ -281,7 +287,7 @@ async def request(
             await self._global_over.wait()
 
         response: aiohttp.ClientResponse | None = None
-        data: dict[str, Any] | str | None = None
+        data: dict[str, Any] | str | bytes | None = None
         await lock.acquire()
         with MaybeUnlock(lock) as maybe_lock:
             for tries in range(5):
@@ -308,7 +314,7 @@ async def request(
                         )
 
                         # even errors have text involved in them so this is safe to call
-                        data = await json_or_text(response)
+                        data = await parse_response(response)
 
                         # check if we have rate limit header information
                         remaining = response.headers.get("X-Ratelimit-Remaining")
@@ -2046,9 +2052,10 @@ def create_invite(
         target_type: invite.InviteTargetType | None = None,
         target_user_id: Snowflake | None = None,
         target_application_id: Snowflake | None = None,
+        roles: list[Snowflake] | None = None,
+        target_users_file: File | None = None,
     ) -> Response[invite.Invite]:
-        r = Route("POST", "/channels/{channel_id}/invites", channel_id=channel_id)
-        payload = {
+        payload: dict[str, Any] = {
             "max_age": max_age,
             "max_uses": max_uses,
             "temporary": temporary,
@@ -2064,7 +2071,27 @@ def create_invite(
         if target_application_id:
             payload["target_application_id"] = str(target_application_id)
 
-        return self.request(r, reason=reason, json=payload)
+        if roles:
+            payload["role_ids"] = roles
+
+        route = Route("POST", "/channels/{channel_id}/invites", channel_id=channel_id)
+
+        if target_users_file is not None:
+            form = [
+                {
+                    "name": "target_users_file",
+                    "value": target_users_file.fp,
+                    "filename": target_users_file.filename,
+                    "content_type": "text/csv",
+                },
+                {
+                    "name": "payload_json",
+                    "value": utils._to_json(payload),
+                },
+            ]
+            return self.request(route, reason=reason, form=form)
+
+        return self.request(route, reason=reason, json=payload)
 
     def get_invite(
         self,
@@ -2086,6 +2113,46 @@ def get_invite(
             Route("GET", "/invites/{invite_id}", invite_id=invite_id), params=params
         )
 
+    def get_invite_target_users(
+        self,
+        invite_id: str,
+    ) -> Response[bytes]:
+        return self.request(
+            Route("GET", "/invites/{invite_id}/target-users", invite_id=invite_id)
+        )
+
+    def update_invite_target_users(
+        self,
+        invite_id: str,
+        *,
+        target_users_file: File,
+    ) -> Response[invite.Invite]:
+        form = [
+            {
+                "name": "target_users_file",
+                "value": target_users_file.fp,
+                "filename": target_users_file.filename,
+                "content_type": "text/csv",
+            },
+        ]
+
+        return self.request(
+            Route("PUT", "/invites/{invite_id}/target-users", invite_id=invite_id),
+            form=form,
+        )
+
+    def get_invite_target_users_job_status(
+        self,
+        invite_id: str,
+    ) -> Response[InviteTargetUsersJobStatusPayload]:
+        return self.request(
+            Route(
+                "GET",
+                "/invites/{invite_id}/target-users/job-status",
+                invite_id=invite_id,
+            ),
+        )
+
     def invites_from(self, guild_id: Snowflake) -> Response[list[invite.Invite]]:
         return self.request(
             Route("GET", "/guilds/{guild_id}/invites", guild_id=guild_id)