Add support for community invites

discord/abc.py

@@ -1283,6 +1283,8 @@ class GuildChannel:
         target_user: Optional[User] = None,
         target_application_id: Optional[int] = None,
         guest: bool = False,
+        roles: Optional[List[Snowflake]] = None,
+        users: Optional[List[Snowflake]] = None,
     ) -> Invite:
         """|coro|
 
@@ -1325,6 +1327,18 @@ class GuildChannel:
             Whether the invite is a guest invite.
 
             .. versionadded:: 2.6
+        roles: Optional[List[:class:`~discord.abc.Snowflake`]]
+            A list of roles that should be granted to the users joining via this invite.
+
+            Requires :attr:`~discord.Permissions.manage_guild` permission.
+
+            .. versionadded:: 2.7
+        users: Optional[List[:class:`~discord.abc.Snowflake`]]
+            A list of user IDs that should be able to access this invite.
+
+            Requires :attr:`~discord.Permissions.manage_guild` permission.
+
+            .. versionadded:: 2.7
 
         Raises
         -------
@@ -1358,6 +1372,8 @@ class GuildChannel:
             target_user_id=target_user.id if target_user else None,
             target_application_id=target_application_id,
             flags=flags.value if flags else None,
+            role_ids=[role.id for role in roles or []],
+            user_ids=[user.id for user in users or []],
         )
         return Invite.from_incomplete(data=data, state=self._state)
 

discord/enums.py

@@ -87,6 +87,7 @@ __all__ = (
     'MediaItemLoadingState',
     'CollectibleType',
     'NameplatePalette',
+    'InviteTargetUsersJobErrorStatus'
 )
 
 
@@ -1002,6 +1003,13 @@ class NameplatePalette(Enum):
     white = 'white'
 
 
+class InviteTargetUsersJobErrorStatus(Enum):
+    unspecified = 0
+    pending = 1
+    completed = 2
+    failed = 3
+
+
 def create_unknown_value(cls: Type[E], val: Any) -> E:
     value_cls = cls._enum_value_cls_  # type: ignore # This is narrowed below
     name = f'unknown_{val}'

discord/http.py

@@ -1837,6 +1837,24 @@ class HTTPClient:
 
     # Invite management
 
+    def _generate_invite_multipart(
+        self,
+        *,
+        payload: dict[str, Any],
+        user_ids: List[Snowflake],
+    ) -> MultipartParameters:
+        users ="Users\n" + "\n".join(str(user_id) for user_id in user_ids)
+        form = [
+            {'name': 'payload_json', 'value': utils._to_json(payload)},
+            {'name': 'target_users_file', 'value': users, 'filename': 'users.csv', 'content_type': 'text/csv'}
+        ]
+
+        return MultipartParameters(
+            multipart=form,
+            payload={},
+            files=None,
+        )
+
     def create_invite(
         self,
         channel_id: Snowflake,
@@ -1850,6 +1868,8 @@ class HTTPClient:
         target_user_id: Optional[Snowflake] = None,
         target_application_id: Optional[Snowflake] = None,
         flags: Optional[int] = None,
+        role_ids: Optional[List[Snowflake]] = None,
+        user_ids: Optional[List[Snowflake]] = None,
     ) -> Response[invite.Invite]:
         r = Route('POST', '/channels/{channel_id}/invites', channel_id=channel_id)
         payload = {
@@ -1871,6 +1891,17 @@ class HTTPClient:
         if flags:
             payload['flags'] = flags
 
+        if role_ids:
+            payload["role_ids"] = list(map(str, role_ids))
+
+        if user_ids:
+            multipart_params = self._generate_invite_multipart(payload=payload, user_ids=user_ids)
+            return self.request(
+                r,
+                form=multipart_params.multipart,
+                reason=reason,
+            )
+
         return self.request(r, reason=reason, json=payload)
 
     def get_invite(
@@ -1898,6 +1929,40 @@ class HTTPClient:
     def delete_invite(self, invite_id: str, *, reason: Optional[str] = None) -> Response[invite.Invite]:
         return self.request(Route('DELETE', '/invites/{invite_id}', invite_id=invite_id), reason=reason)
 
+    def get_invite_target_users(
+        self,
+        invite_id: str,
+    ) -> Response[str]:
+        return self.request(
+            Route('GET', '/invites/{invite_id}/target-users', invite_id=invite_id)
+        )
+
+    def edit_invite_target_users(
+        self,
+        invite_id: str,
+        user_ids: List[Snowflake],
+    ) -> Response[None]:
+        multipart_params = self._generate_invite_multipart(
+            payload={},
+            user_ids=user_ids,
+        )
+        return self.request(
+            Route('PUT', '/invites/{invite_id}/target-users', invite_id=invite_id),
+            form=multipart_params.multipart,
+        )
+
+    def get_invite_target_users_job_status(
+        self,
+        invite_id: str,
+    ) -> Response[invite.InviteTargetUsersJobStatus]:
+        return self.request(
+            Route(
+                "GET",
+                '/invites/{invite_id}/target-users/job-status',
+                invite_id=invite_id,
+            )
+        )
+
     # Role management
 
     def get_roles(self, guild_id: Snowflake) -> Response[List[role.Role]]:

discord/invite.py

@@ -26,13 +26,22 @@ from __future__ import annotations
 
 from typing import List, Optional, Union, TYPE_CHECKING
 from .asset import Asset
-from .utils import parse_time, snowflake_time, _get_as_snowflake
+from .utils import parse_time, snowflake_time, _get_as_snowflake, MISSING
 from .object import Object
 from .mixins import Hashable
-from .enums import ChannelType, NSFWLevel, VerificationLevel, InviteTarget, InviteType, try_enum
+from .enums import (
+    ChannelType,
+    NSFWLevel,
+    VerificationLevel,
+    InviteTarget,
+    InviteType,
+    InviteTargetUsersJobErrorStatus,
+    try_enum,
+)
 from .appinfo import PartialAppInfo
 from .scheduled_event import ScheduledEvent
 from .flags import InviteFlags
+from .role import Role
 
 __all__ = (
     'PartialInviteChannel',
@@ -47,6 +56,7 @@ if TYPE_CHECKING:
         Invite as InvitePayload,
         InviteGuild as InviteGuildPayload,
         GatewayInvite as GatewayInvitePayload,
+        InviteTargetUsersJobStatus as InviteTargetUsersJobStatusPayload,
     )
     from .types.guild import GuildFeature
     from .types.channel import (
@@ -64,6 +74,47 @@ if TYPE_CHECKING:
     import datetime
 
 
+class InviteTargetUsersJobStatus:
+    """Represents the status of an invite's target users job.
+
+    .. versionadded:: 2.7
+
+    Attributes
+    -----------
+    invite: :class:`Invite`
+        The invite this job status is for.
+    status: :class:`InviteTargetUsersJobStatus`
+        The status of the job.
+    total_users: :class:`int`
+        The total number of users in the job.
+    processed_users: :class:`int`
+        The number of users that have been processed so far.
+    created_at: :class:`datetime.datetime`
+        The time the job was created.
+    error_message: :class:`str`
+        The error message.
+    completed_at: Optional[:class:`datetime.datetime`]
+        The time the job was completed, if applicable.
+    """
+
+    def __init__(self, *, invite: Invite, data: InviteTargetUsersJobStatusPayload) -> None:
+        self.invite: Invite = invite
+        self.status: InviteTargetUsersJobErrorStatus = try_enum(InviteTargetUsersJobErrorStatus, data['status'])
+        self.total_users: int = data['total_users']
+        self.processed_users: int = data['processed_users']
+        self.created_at: datetime.datetime = parse_time(data['created_at'])
+        self.error_message: str = data['error_message']
+        self.completed_at: Optional[datetime.datetime] = (
+            parse_time(data['completed_at']) if data.get('completed_at') else None
+        )
+
+    def __repr__(self) -> str:
+        return (
+            f'<InviteTargetUsersJobStatus invite={self.invite.code!r} status={self.status} '
+            f'total_users={self.total_users} processed_users={self.processed_users}>'
+        )
+
+
 class PartialInviteChannel:
     """Represents a "partial" invite channel.
 
@@ -358,6 +409,9 @@ class Invite(Hashable):
         The ID of the scheduled event associated with this invite, if any.
 
         .. versionadded:: 2.0
+    roles: List[:class:`Role`]
+        A list of roles the invite grants access to. Only available if the
+        invite's guild is cached.
     """
 
     __slots__ = (
@@ -382,6 +436,7 @@ class Invite(Hashable):
         'scheduled_event_id',
         'type',
         '_flags',
+        'roles',
     )
 
     BASE = 'https://discord.gg'
@@ -437,6 +492,13 @@ class Invite(Hashable):
         self.scheduled_event_id: Optional[int] = self.scheduled_event.id if self.scheduled_event else None
         self._flags: int = data.get('flags', 0)
 
+        if self.guild is not None and not isinstance(self.guild, (PartialInviteGuild, Object)):
+            self.roles: List[Role] = [
+                Role(state=self._state, guild=self.guild, data=role_data) for role_data in data.get('roles', [])
+            ]
+        else:
+            self.roles: List[Role] = []
+
     @classmethod
     def from_incomplete(cls, *, state: ConnectionState, data: InvitePayload) -> Self:
         guild: Optional[Union[Guild, PartialInviteGuild]]
@@ -582,3 +644,83 @@ class Invite(Hashable):
 
         data = await self._state.http.delete_invite(self.code, reason=reason)
         return self.from_incomplete(state=self._state, data=data)
+
+    async def fetch_target_users(self) -> list[int]:
+        """|coro|
+
+        Fetches the target users CSV file for this invite.
+
+        Requires the :attr:`~Permissions.manage_guild` permission.
+
+        Returns
+        --------
+        List[:class:`int`]
+            A list of user IDs representing the target users.
+
+        Raises
+        -------
+        Forbidden
+            You do not have permissions to fetch target users.
+        NotFound
+            The invite is invalid or expired or the invite does not have target users.
+        HTTPException
+            Fetching the target users failed.
+        """
+
+        string = await self._state.http.get_invite_target_users(self.code)
+        users = string.lstrip('Users\n').split('\n')
+        return [int(user_id) for user_id in users if user_id]
+
+    async def fetch_target_users_job_status(self) -> InviteTargetUsersJobStatus:
+        """|coro|
+
+        Fetches the status of the target users job for this invite.
+
+        Requires the :attr:`~Permissions.manage_guild` permission.
+
+        Returns
+        --------
+        :class:`InviteTargetUsersJobStatus`
+            A dictionary containing the status of the target users job.
+
+        Raises
+        -------
+        Forbidden
+            You do not have permissions to fetch target users job status.
+        NotFound
+            The invite is invalid or expired or there is no ongoing target users job.
+        HTTPException
+            Fetching the target users job status failed.
+        """
+
+        data = await self._state.http.get_invite_target_users_job_status(self.code)
+        return InviteTargetUsersJobStatus(invite=self, data=data)
+
+    async def edit(
+        self,
+        *,
+        users: List[Snowflake] = MISSING,
+    ) -> None:
+        """|coro|
+
+        Edits the invite.
+
+        Requires the :attr:`~Permissions.manage_guild` permission.
+
+        Parameters
+        -----------
+        users: List[:class:`~discord.abc.Snowflake`]
+            A list of users that should be able to use this invite.
+
+        Raises
+        -------
+        Forbidden
+            You do not have permissions to edit invites.
+        NotFound
+            The invite is invalid or expired.
+        HTTPException
+            Editing the invite failed.
+        """
+
+        if users is not MISSING:
+            await self._state.http.edit_invite_target_users(self.code, user_ids=[user.id for user in users])

discord/types/invite.py

@@ -33,6 +33,7 @@ from .guild import InviteGuild, _GuildPreviewUnique
 from .channel import PartialChannel
 from .user import PartialUser
 from .appinfo import PartialAppInfo
+from .role import Role
 
 InviteTargetType = Literal[1, 2]
 InviteType = Literal[0, 1, 2]
@@ -66,6 +67,7 @@ class Invite(IncompleteInvite, total=False):
     type: InviteType
     flags: NotRequired[int]
     expires_at: Optional[str]
+    roles: NotRequired[list[Role]]
 
 
 class InviteWithCounts(Invite, _GuildPreviewUnique): ...
@@ -95,3 +97,12 @@ class GatewayInviteDelete(TypedDict):
 
 
 GatewayInvite = Union[GatewayInviteCreate, GatewayInviteDelete]
+
+
+class InviteTargetUsersJobStatus(TypedDict):
+    status: int
+    total_users: int
+    processed_users: int
+    created_at: str
+    error_message: str
+    completed_at: Optional[str]

docs/api.rst

@@ -4154,6 +4154,28 @@ of :class:`enum.Enum`.
 
         The collectible nameplate palette is white.
 
+.. class:: InviteTargetUsersJobErrorStatus
+
+    Represents the error status of an invite target users job.
+
+    .. versionadded:: 2.7
+
+    .. attribute:: unspecified
+
+        The default value.
+
+    .. attribute:: processing
+
+        The job is still being processed.
+
+    .. attribute:: completed
+
+        The job has been completed successfully.
+
+    .. attribute:: failed
+
+        The job has failed, see `error_message` field for more details.
+
 .. _discord-api-audit-logs:
 
 Audit Log Data