Add support for premium app integrations

Co-authored-by: Danny <1695103+Rapptz@users.noreply.github.com>
Co-authored-by: Lucas Hardt <lucas.hardt@fu-berlin.de>
Co-authored-by: Andrin S. <65789180+Puncher1@users.noreply.github.com>

discord/__init__.py

@@ -41,6 +41,7 @@ from .integrations import *
 from .invite import *
 from .template import *
 from .welcome_screen import *
+from .sku import *
 from .widget import *
 from .object import *
 from .reaction import *

discord/client.py

@@ -48,6 +48,7 @@ from typing import (
 
 import aiohttp
 
+from .sku import SKU, Entitlement
 from .user import User, ClientUser
 from .invite import Invite
 from .template import Template
@@ -55,7 +56,7 @@ from .widget import Widget
 from .guild import Guild
 from .emoji import Emoji
 from .channel import _threaded_channel_factory, PartialMessageable
-from .enums import ChannelType
+from .enums import ChannelType, EntitlementOwnerType
 from .mentions import AllowedMentions
 from .errors import *
 from .enums import Status
@@ -83,7 +84,7 @@ if TYPE_CHECKING:
     from typing_extensions import Self
 
     from .abc import Messageable, PrivateChannel, Snowflake, SnowflakeTime
-    from .app_commands import Command, ContextMenu
+    from .app_commands import Command, ContextMenu, MissingApplicationID
     from .automod import AutoModAction, AutoModRule
     from .channel import DMChannel, GroupChannel
     from .ext.commands import AutoShardedBot, Bot, Context, CommandError
@@ -674,7 +675,6 @@ class Client:
                 aiohttp.ClientError,
                 asyncio.TimeoutError,
             ) as exc:
-
                 self.dispatch('disconnect')
                 if not reconnect:
                     await self.close()
@@ -2632,6 +2632,242 @@ class Client:
         # The type checker is not smart enough to figure out the constructor is correct
         return cls(state=self._connection, data=data)  # type: ignore
 
+    async def fetch_skus(self) -> List[SKU]:
+        """|coro|
+
+        Retrieves the bot's available SKUs.
+
+        .. versionadded:: 2.4
+
+        Raises
+        -------
+        MissingApplicationID
+            The application ID could not be found.
+        HTTPException
+            Retrieving the SKUs failed.
+
+        Returns
+        --------
+        List[:class:`.SKU`]
+            The bot's available SKUs.
+        """
+
+        if self.application_id is None:
+            raise MissingApplicationID
+
+        data = await self.http.get_skus(self.application_id)
+        return [SKU(state=self._connection, data=sku) for sku in data]
+
+    async def fetch_entitlement(self, entitlement_id: int, /) -> Entitlement:
+        """|coro|
+
+        Retrieves a :class:`.Entitlement` with the specified ID.
+
+        .. versionadded:: 2.4
+
+        Parameters
+        -----------
+        entitlement_id: :class:`int`
+            The entitlement's ID to fetch from.
+
+        Raises
+        -------
+        NotFound
+            An entitlement with this ID does not exist.
+        MissingApplicationID
+            The application ID could not be found.
+        HTTPException
+            Fetching the entitlement failed.
+
+        Returns
+        --------
+        :class:`.Entitlement`
+            The entitlement you requested.
+        """
+
+        if self.application_id is None:
+            raise MissingApplicationID
+
+        data = await self.http.get_entitlement(self.application_id, entitlement_id)
+        return Entitlement(state=self._connection, data=data)
+
+    async def entitlements(
+        self,
+        *,
+        limit: Optional[int] = 100,
+        before: Optional[SnowflakeTime] = None,
+        after: Optional[SnowflakeTime] = None,
+        skus: Optional[Sequence[Snowflake]] = None,
+        user: Optional[Snowflake] = None,
+        guild: Optional[Snowflake] = None,
+        exclude_ended: bool = False,
+    ) -> AsyncIterator[Entitlement]:
+        """Retrieves an :term:`asynchronous iterator` of the :class:`.Entitlement` that applications has.
+
+        .. versionadded:: 2.4
+
+        Examples
+        ---------
+
+        Usage ::
+
+            async for entitlement in client.entitlements(limit=100):
+                print(entitlement.user_id, entitlement.ends_at)
+
+        Flattening into a list ::
+
+            entitlements = [entitlement async for entitlement in client.entitlements(limit=100)]
+            # entitlements is now a list of Entitlement...
+
+        All parameters are optional.
+
+        Parameters
+        -----------
+        limit: Optional[:class:`int`]
+            The number of entitlements to retrieve. If ``None``, it retrieves every entitlement for this application.
+            Note, however, that this would make it a slow operation. Defaults to ``100``.
+        before: Optional[Union[:class:`~discord.abc.Snowflake`, :class:`datetime.datetime`]]
+            Retrieve entitlements before this date or entitlement.
+            If a datetime is provided, it is recommended to use a UTC aware datetime.
+            If the datetime is naive, it is assumed to be local time.
+        after: Optional[Union[:class:`~discord.abc.Snowflake`, :class:`datetime.datetime`]]
+            Retrieve entitlements after this date or entitlement.
+            If a datetime is provided, it is recommended to use a UTC aware datetime.
+            If the datetime is naive, it is assumed to be local time.
+        skus: Optional[Sequence[:class:`~discord.abc.Snowflake`]]
+            A list of SKUs to filter by.
+        user: Optional[:class:`~discord.abc.Snowflake`]
+            The user to filter by.
+        guild: Optional[:class:`~discord.abc.Snowflake`]
+            The guild to filter by.
+        exclude_ended: :class:`bool`
+            Whether to exclude ended entitlements. Defaults to ``False``.
+
+        Raises
+        -------
+        MissingApplicationID
+            The application ID could not be found.
+        HTTPException
+            Fetching the entitlements failed.
+        TypeError
+            Both ``after`` and ``before`` were provided, as Discord does not
+            support this type of pagination.
+
+        Yields
+        --------
+        :class:`.Entitlement`
+            The entitlement with the application.
+        """
+
+        if self.application_id is None:
+            raise MissingApplicationID
+
+        if before is not None and after is not None:
+            raise TypeError('entitlements pagination does not support both before and after')
+
+        # This endpoint paginates in ascending order.
+        endpoint = self.http.get_entitlements
+
+        async def _before_strategy(retrieve: int, before: Optional[Snowflake], limit: Optional[int]):
+            before_id = before.id if before else None
+            data = await endpoint(
+                self.application_id,  # type: ignore  # We already check for None above
+                limit=retrieve,
+                before=before_id,
+                sku_ids=[sku.id for sku in skus] if skus else None,
+                user_id=user.id if user else None,
+                guild_id=guild.id if guild else None,
+                exclude_ended=exclude_ended,
+            )
+
+            if data:
+                if limit is not None:
+                    limit -= len(data)
+
+                before = Object(id=int(data[0]['id']))
+
+            return data, before, limit
+
+        async def _after_strategy(retrieve: int, after: Optional[Snowflake], limit: Optional[int]):
+            after_id = after.id if after else None
+            data = await endpoint(
+                self.application_id,  # type: ignore  # We already check for None above
+                limit=retrieve,
+                after=after_id,
+                sku_ids=[sku.id for sku in skus] if skus else None,
+                user_id=user.id if user else None,
+                guild_id=guild.id if guild else None,
+                exclude_ended=exclude_ended,
+            )
+
+            if data:
+                if limit is not None:
+                    limit -= len(data)
+
+                after = Object(id=int(data[-1]['id']))
+
+            return data, after, limit
+
+        if isinstance(before, datetime.datetime):
+            before = Object(id=utils.time_snowflake(before, high=False))
+        if isinstance(after, datetime.datetime):
+            after = Object(id=utils.time_snowflake(after, high=True))
+
+        if before:
+            strategy, state = _before_strategy, before
+        else:
+            strategy, state = _after_strategy, after
+
+        while True:
+            retrieve = 100 if limit is None else min(limit, 100)
+            if retrieve < 1:
+                return
+
+            data, state, limit = await strategy(retrieve, state, limit)
+
+            # Terminate loop on next iteration; there's no data left after this
+            if len(data) < 1000:
+                limit = 0
+
+            for e in data:
+                yield Entitlement(self._connection, e)
+
+    async def create_entitlement(
+        self,
+        sku: Snowflake,
+        owner: Snowflake,
+        owner_type: EntitlementOwnerType,
+    ) -> None:
+        """|coro|
+
+        Creates a test :class:`.Entitlement` for the application.
+
+        .. versionadded:: 2.4
+
+        Parameters
+        -----------
+        sku: :class:`~discord.abc.Snowflake`
+            The SKU to create the entitlement for.
+        owner: :class:`~discord.abc.Snowflake`
+            The ID of the owner.
+        owner_type: :class:`.EntitlementOwnerType`
+            The type of the owner.
+
+        Raises
+        -------
+        MissingApplicationID
+            The application ID could not be found.
+        NotFound
+            The SKU or owner could not be found.
+        HTTPException
+            Creating the entitlement failed.
+        """
+
+        if self.application_id is None:
+            raise MissingApplicationID
+
+        await self.http.create_entitlement(self.application_id, sku.id, owner.id, owner_type.value)
+
     async def fetch_premium_sticker_packs(self) -> List[StickerPack]:
         """|coro|
 

discord/enums.py

@@ -70,6 +70,9 @@ __all__ = (
     'ForumLayoutType',
     'ForumOrderType',
     'SelectDefaultValueType',
+    'SKUType',
+    'EntitlementType',
+    'EntitlementOwnerType',
 )
 
 if TYPE_CHECKING:
@@ -591,6 +594,7 @@ class InteractionResponseType(Enum):
     message_update = 7  # for components
     autocomplete_result = 8
     modal = 9  # for modals
+    premium_required = 10
 
 
 class VideoQualityMode(Enum):
@@ -782,6 +786,20 @@ class SelectDefaultValueType(Enum):
     channel = 'channel'
 
 
+class SKUType(Enum):
+    subscription = 5
+    subscription_group = 6
+
+
+class EntitlementType(Enum):
+    application_subscription = 8
+
+
+class EntitlementOwnerType(Enum):
+    guild = 1
+    user = 2
+
+
 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/flags.py

@@ -60,6 +60,7 @@ __all__ = (
     'MemberFlags',
     'AttachmentFlags',
     'RoleFlags',
+    'SKUFlags',
 )
 
 BF = TypeVar('BF', bound='BaseFlags')
@@ -1971,3 +1972,76 @@ class RoleFlags(BaseFlags):
     def in_prompt(self):
         """:class:`bool`: Returns ``True`` if the role can be selected by members in an onboarding prompt."""
         return 1 << 0
+
+
+@fill_with_flags()
+class SKUFlags(BaseFlags):
+    r"""Wraps up the Discord SKU flags
+
+    .. versionadded:: 2.4
+
+    .. container:: operations
+
+        .. describe:: x == y
+
+            Checks if two SKUFlags are equal.
+
+        .. describe:: x != y
+
+            Checks if two SKUFlags are not equal.
+
+        .. describe:: x | y, x |= y
+
+            Returns a SKUFlags instance with all enabled flags from
+            both x and y.
+
+        .. describe:: x & y, x &= y
+
+            Returns a SKUFlags instance with only flags enabled on
+            both x and y.
+
+        .. describe:: x ^ y, x ^= y
+
+            Returns a SKUFlags instance with only flags enabled on
+            only one of x or y, not on both.
+
+        .. describe:: ~x
+
+            Returns a SKUFlags instance with all flags inverted from x.
+
+        .. describe:: hash(x)
+
+            Return the flag's hash.
+
+        .. describe:: iter(x)
+
+            Returns an iterator of ``(name, value)`` pairs. This allows it
+            to be, for example, constructed as a dict or a list of pairs.
+            Note that aliases are not shown.
+
+        .. describe:: bool(b)
+
+            Returns whether any flag is set to ``True``.
+
+
+    Attributes
+    -----------
+    value: :class:`int`
+        The raw value. You should query flags via the properties
+        rather than using this raw value.
+    """
+
+    @flag_value
+    def available(self):
+        """:class:`bool`: Returns ``True`` if the SKU is available for purchase."""
+        return 1 << 2
+
+    @flag_value
+    def guild_subscription(self):
+        """:class:`bool`: Returns ``True`` if the SKU is a guild subscription."""
+        return 1 << 7
+
+    @flag_value
+    def user_subscription(self):
+        """:class:`bool`: Returns ``True`` if the SKU is a user subscription."""
+        return 1 << 8

discord/http.py

@@ -90,6 +90,7 @@ if TYPE_CHECKING:
         scheduled_event,
         sticker,
         welcome_screen,
+        sku,
     )
     from .types.snowflake import Snowflake, SnowflakeList
 
@@ -2375,6 +2376,81 @@ class HTTPClient:
             reason=reason,
         )
 
+    # SKU
+
+    def get_skus(self, application_id: Snowflake) -> Response[List[sku.SKU]]:
+        return self.request(Route('GET', '/applications/{application_id}/skus', application_id=application_id))
+
+    def get_entitlements(
+        self,
+        application_id: Snowflake,
+        user_id: Optional[Snowflake] = None,
+        sku_ids: Optional[SnowflakeList] = None,
+        before: Optional[Snowflake] = None,
+        after: Optional[Snowflake] = None,
+        limit: Optional[int] = None,
+        guild_id: Optional[Snowflake] = None,
+        exclude_ended: Optional[bool] = None,
+    ) -> Response[List[sku.Entitlement]]:
+        params: Dict[str, Any] = {}
+
+        if user_id is not None:
+            params['user_id'] = user_id
+        if sku_ids is not None:
+            params['sku_ids'] = ','.join(map(str, sku_ids))
+        if before is not None:
+            params['before'] = before
+        if after is not None:
+            params['after'] = after
+        if limit is not None:
+            params['limit'] = limit
+        if guild_id is not None:
+            params['guild_id'] = guild_id
+        if exclude_ended is not None:
+            params['exclude_ended'] = int(exclude_ended)
+
+        return self.request(
+            Route('GET', '/applications/{application_id}/entitlements', application_id=application_id), params=params
+        )
+
+    def get_entitlement(self, application_id: Snowflake, entitlement_id: Snowflake) -> Response[sku.Entitlement]:
+        return self.request(
+            Route(
+                'GET',
+                '/applications/{application_id}/entitlements/{entitlement_id}',
+                application_id=application_id,
+                entitlement_id=entitlement_id,
+            ),
+        )
+
+    def create_entitlement(
+        self, application_id: Snowflake, sku_id: Snowflake, owner_id: Snowflake, owner_type: sku.EntitlementOwnerType
+    ) -> Response[sku.Entitlement]:
+        payload = {
+            'sku_id': sku_id,
+            'owner_id': owner_id,
+            'owner_type': owner_type,
+        }
+
+        return self.request(
+            Route(
+                'POST',
+                '/applications/{application.id}/entitlements',
+                application_id=application_id,
+            ),
+            json=payload,
+        )
+
+    def delete_entitlement(self, application_id: Snowflake, entitlement_id: Snowflake) -> Response[sku.Entitlement]:
+        return self.request(
+            Route(
+                'DELETE',
+                '/applications/{application_id}/entitlements/{entitlement_id}',
+                application_id=application_id,
+                entitlement_id=entitlement_id,
+            ),
+        )
+
     # Misc
 
     def application_info(self) -> Response[appinfo.AppInfo]:

discord/interactions.py

@@ -27,7 +27,7 @@ DEALINGS IN THE SOFTWARE.
 from __future__ import annotations
 
 import logging
-from typing import Any, Dict, Optional, Generic, TYPE_CHECKING, Sequence, Tuple, Union
+from typing import Any, Dict, Optional, Generic, TYPE_CHECKING, Sequence, Tuple, Union, List
 import asyncio
 import datetime
 
@@ -37,6 +37,7 @@ from .errors import InteractionResponded, HTTPException, ClientException, Discor
 from .flags import MessageFlags
 from .channel import ChannelType
 from ._types import ClientT
+from .sku import Entitlement
 
 from .user import User
 from .member import Member
@@ -110,6 +111,10 @@ class Interaction(Generic[ClientT]):
         The channel the interaction was sent from.
 
         Note that due to a Discord limitation, if sent from a DM channel :attr:`~DMChannel.recipient` is ``None``.
+    entitlement_sku_ids: List[:class:`int`]
+        The entitlement SKU IDs that the user has.
+    entitlements: List[:class:`Entitlement`]
+        The entitlements that the guild or user has.
     application_id: :class:`int`
         The application ID that the interaction was for.
     user: Union[:class:`User`, :class:`Member`]
@@ -150,6 +155,8 @@ class Interaction(Generic[ClientT]):
         'guild_locale',
         'extras',
         'command_failed',
+        'entitlement_sku_ids',
+        'entitlements',
         '_permissions',
         '_app_permissions',
         '_state',
@@ -185,6 +192,8 @@ class Interaction(Generic[ClientT]):
         self.guild_id: Optional[int] = utils._get_as_snowflake(data, 'guild_id')
         self.channel: Optional[InteractionChannel] = None
         self.application_id: int = int(data['application_id'])
+        self.entitlement_sku_ids: List[int] = [int(x) for x in data.get('entitlement_skus', []) or []]
+        self.entitlements: List[Entitlement] = [Entitlement(self._state, x) for x in data.get('entitlements', [])]
 
         self.locale: Locale = try_enum(Locale, data.get('locale', 'en-US'))
         self.guild_locale: Optional[Locale]
@@ -984,6 +993,38 @@ class InteractionResponse(Generic[ClientT]):
             self._parent._state.store_view(modal)
         self._response_type = InteractionResponseType.modal
 
+    async def require_premium(self) -> None:
+        """|coro|
+
+        Sends a message to the user prompting them that a premium purchase is required for this interaction.
+
+        This type of response is only available for applications that have a premium SKU set up.
+
+        Raises
+        -------
+        HTTPException
+            Sending the response failed.
+        InteractionResponded
+            This interaction has already been responded to before.
+        """
+        if self._response_type:
+            raise InteractionResponded(self._parent)
+
+        parent = self._parent
+        adapter = async_context.get()
+        http = parent._state.http
+
+        params = interaction_response_params(InteractionResponseType.premium_required.value)
+        await adapter.create_interaction_response(
+            parent.id,
+            parent.token,
+            session=parent._session,
+            proxy=http.proxy,
+            proxy_auth=http.proxy_auth,
+            params=params,
+        )
+        self._response_type = InteractionResponseType.premium_required
+
     async def autocomplete(self, choices: Sequence[Choice[ChoiceT]]) -> None:
         """|coro|
 

discord/sku.py

@@ -0,0 +1,200 @@
+"""
+The MIT License (MIT)
+
+Copyright (c) 2015-present Rapptz
+
+Permission is hereby granted, free of charge, to any person obtaining a
+copy of this software and associated documentation files (the "Software"),
+to deal in the Software without restriction, including without limitation
+the rights to use, copy, modify, merge, publish, distribute, sublicense,
+and/or sell copies of the Software, and to permit persons to whom the
+Software is furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+DEALINGS IN THE SOFTWARE.
+"""
+
+
+from __future__ import annotations
+
+from typing import Optional, TYPE_CHECKING
+
+from . import utils
+from .app_commands import MissingApplicationID
+from .enums import try_enum, SKUType, EntitlementType
+from .flags import SKUFlags
+
+if TYPE_CHECKING:
+    from datetime import datetime
+
+    from .guild import Guild
+    from .state import ConnectionState
+    from .types.sku import (
+        SKU as SKUPayload,
+        Entitlement as EntitlementPayload,
+    )
+    from .user import User
+
+__all__ = (
+    'SKU',
+    'Entitlement',
+)
+
+
+class SKU:
+    """Represents a premium offering as a stock-keeping unit (SKU).
+
+    .. versionadded:: 2.4
+
+    Attributes
+    -----------
+    id: :class:`int`
+        The SKU's ID.
+    type: :class:`SKUType`
+        The type of the SKU.
+    application_id: :class:`int`
+        The ID of the application that the SKU belongs to.
+    name: :class:`str`
+        The consumer-facing name of the premium offering.
+    slug: :class:`str`
+        A system-generated URL slug based on the SKU name.
+    """
+
+    __slots__ = (
+        '_state',
+        'id',
+        'type',
+        'application_id',
+        'name',
+        'slug',
+        '_flags',
+    )
+
+    def __init__(self, *, state: ConnectionState, data: SKUPayload):
+        self._state: ConnectionState = state
+        self.id: int = int(data['id'])
+        self.type: SKUType = try_enum(SKUType, data['type'])
+        self.application_id: int = int(data['application_id'])
+        self.name: str = data['name']
+        self.slug: str = data['slug']
+        self._flags: int = data['flags']
+
+    def __repr__(self) -> str:
+        return f'<SKU id={self.id} name={self.name!r} slug={self.slug!r}>'
+
+    @property
+    def flags(self) -> SKUFlags:
+        """Returns the flags of the SKU."""
+        return SKUFlags._from_value(self._flags)
+
+    @property
+    def created_at(self) -> datetime:
+        """:class:`datetime.datetime`: Returns the sku's creation time in UTC."""
+        return utils.snowflake_time(self.id)
+
+
+class Entitlement:
+    """Represents an entitlement from user or guild which has been granted access to a premium offering.
+
+    .. versionadded:: 2.4
+
+    Attributes
+    -----------
+    id: :class:`int`
+        The entitlement's ID.
+    sku_id: :class:`int`
+        The ID of the SKU that the entitlement belongs to.
+    application_id: :class:`int`
+        The ID of the application that the entitlement belongs to.
+    user_id: Optional[:class:`int`]
+        The ID of the user that is granted access to the entitlement.
+    type: :class:`EntitlementType`
+        The type of the entitlement.
+    deleted: :class:`bool`
+        Whether the entitlement has been deleted.
+    starts_at: Optional[:class:`datetime.datetime`]
+        A UTC start date which the entitlement is valid. Not present when using test entitlements.
+    ends_at: Optional[:class:`datetime.datetime`]
+        A UTC date which entitlement is no longer valid. Not present when using test entitlements.
+    guild_id: Optional[:class:`int`]
+        The ID of the guild that is granted access to the entitlement
+    """
+
+    __slots__ = (
+        '_state',
+        'id',
+        'sku_id',
+        'application_id',
+        'user_id',
+        'type',
+        'deleted',
+        'starts_at',
+        'ends_at',
+        'guild_id',
+    )
+
+    def __init__(self, state: ConnectionState, data: EntitlementPayload):
+        self._state: ConnectionState = state
+        self.id: int = int(data['id'])
+        self.sku_id: int = int(data['sku_id'])
+        self.application_id: int = int(data['application_id'])
+        self.user_id: Optional[int] = utils._get_as_snowflake(data, 'user_id')
+        self.type: EntitlementType = try_enum(EntitlementType, data['type'])
+        self.deleted: bool = data['deleted']
+        self.starts_at: Optional[datetime] = utils.parse_time(data.get('starts_at', None))
+        self.ends_at: Optional[datetime] = utils.parse_time(data.get('ends_at', None))
+        self.guild_id: Optional[int] = utils._get_as_snowflake(data, 'guild_id')
+
+    def __repr__(self) -> str:
+        return f'<Entitlement id={self.id} type={self.type!r} user_id={self.user_id}>'
+
+    @property
+    def user(self) -> Optional[User]:
+        """The user that is granted access to the entitlement"""
+        if self.user_id is None:
+            return None
+        return self._state.get_user(self.user_id)
+
+    @property
+    def guild(self) -> Optional[Guild]:
+        """The guild that is granted access to the entitlement"""
+        return self._state._get_guild(self.guild_id)
+
+    @property
+    def created_at(self) -> datetime:
+        """:class:`datetime.datetime`: Returns the entitlement's creation time in UTC."""
+        return utils.snowflake_time(self.id)
+
+    def is_expired(self) -> bool:
+        """:class:`bool`: Returns ``True`` if the entitlement is expired. Will be always False for test entitlements."""
+        if self.ends_at is None:
+            return False
+        return utils.utcnow() >= self.ends_at
+
+    async def delete(self) -> None:
+        """|coro|
+
+        Deletes the entitlement.
+
+        Raises
+        -------
+        MissingApplicationID
+            The application ID could not be found.
+        NotFound
+            The entitlement could not be found.
+        HTTPException
+            Deleting the entitlement failed.
+        """
+
+        if self.application_id is None:
+            raise MissingApplicationID
+
+        await self._state.http.delete_entitlement(self.application_id, self.id)

discord/state.py

@@ -53,6 +53,7 @@ import os
 
 from .guild import Guild
 from .activity import BaseActivity
+from .sku import Entitlement
 from .user import User, ClientUser
 from .emoji import Emoji
 from .mentions import AllowedMentions
@@ -1584,6 +1585,18 @@ class ConnectionState(Generic[ClientT]):
 
         self.dispatch('raw_typing', raw)
 
+    def parse_entitlement_create(self, data: gw.EntitlementCreateEvent) -> None:
+        entitlement = Entitlement(data=data, state=self)
+        self.dispatch('entitlement_create', entitlement)
+
+    def parse_entitlement_update(self, data: gw.EntitlementUpdateEvent) -> None:
+        entitlement = Entitlement(data=data, state=self)
+        self.dispatch('entitlement_update', entitlement)
+
+    def parse_entitlement_delete(self, data: gw.EntitlementDeleteEvent) -> None:
+        entitlement = Entitlement(data=data, state=self)
+        self.dispatch('entitlement_update', entitlement)
+
     def _get_reaction_user(self, channel: MessageableChannel, user_id: int) -> Optional[Union[User, Member]]:
         if isinstance(channel, (TextChannel, Thread, VoiceChannel)):
             return channel.guild.get_member(user_id)

discord/types/gateway.py

@@ -27,6 +27,7 @@ from typing_extensions import NotRequired, Required
 
 from .automod import AutoModerationAction, AutoModerationRuleTriggerType
 from .activity import PartialPresenceUpdate
+from .sku import Entitlement
 from .voice import GuildVoiceState
 from .integration import BaseIntegration, IntegrationApplication
 from .role import Role
@@ -347,3 +348,6 @@ class AutoModerationActionExecution(TypedDict):
 
 class GuildAuditLogEntryCreate(AuditLogEntry):
     guild_id: Snowflake
+
+
+EntitlementCreateEvent = EntitlementUpdateEvent = EntitlementDeleteEvent = Entitlement

discord/types/interactions.py

@@ -28,6 +28,7 @@ from typing import TYPE_CHECKING, Dict, List, Literal, TypedDict, Union
 from typing_extensions import NotRequired
 
 from .channel import ChannelTypeWithoutThread, ThreadMetadata, GuildChannel, InteractionDMChannel, GroupDMChannel
+from .sku import Entitlement
 from .threads import ThreadType
 from .member import Member
 from .message import Attachment
@@ -208,6 +209,8 @@ class _BaseInteraction(TypedDict):
     app_permissions: NotRequired[str]
     locale: NotRequired[str]
     guild_locale: NotRequired[str]
+    entitlement_sku_ids: NotRequired[List[Snowflake]]
+    entitlements: NotRequired[List[Entitlement]]
 
 
 class PingInteraction(_BaseInteraction):

discord/types/sku.py

@@ -0,0 +1,52 @@
+"""
+The MIT License (MIT)
+
+Copyright (c) 2015-present Rapptz
+
+Permission is hereby granted, free of charge, to any person obtaining a
+copy of this software and associated documentation files (the "Software"),
+to deal in the Software without restriction, including without limitation
+the rights to use, copy, modify, merge, publish, distribute, sublicense,
+and/or sell copies of the Software, and to permit persons to whom the
+Software is furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+DEALINGS IN THE SOFTWARE.
+"""
+
+from __future__ import annotations
+
+from typing import TypedDict, Optional, Literal
+from typing_extensions import NotRequired
+
+
+class SKU(TypedDict):
+    id: str
+    type: int
+    application_id: str
+    name: str
+    slug: str
+    flags: int
+
+
+class Entitlement(TypedDict):
+    id: str
+    sku_id: str
+    application_id: str
+    user_id: Optional[str]
+    type: int
+    deleted: bool
+    starts_at: NotRequired[str]
+    ends_at: NotRequired[str]
+    guild_id: Optional[str]
+
+
+EntitlementOwnerType = Literal[1, 2]

docs/api.rst

@@ -496,6 +496,47 @@ Debug
     :type payload: Union[:class:`bytes`, :class:`str`]
 
 
+Entitlements
+~~~~~~~~~~~~
+
+.. function:: on_entitlement_create(entitlement)
+
+    Called when a user subscribes to a SKU.
+
+    .. versionadded:: 2.4
+
+    :param entitlement: The entitlement that was created.
+    :type entitlement: :class:`Entitlement`
+
+.. function:: on_entitlement_update(entitlement)
+
+    Called when a user updates their subscription to a SKU. This is usually called when
+    the user renews or cancels their subscription.
+
+    .. versionadded:: 2.4
+
+    :param entitlement: The entitlement that was updated.
+    :type entitlement: :class:`Entitlement`
+
+.. function:: on_entitlement_delete(entitlement)
+
+    Called when a users subscription to a SKU is cancelled. This is typically only called when:
+
+    - Discord issues a refund for the subscription.
+    - Discord removes an entitlement from a user.
+
+    .. warning::
+
+        This event won't be called if the user cancels their subscription manually, instead
+        :func:`on_entitlement_update` will be called with :attr:`Entitlement.ends_at` set to the end of the
+        current billing period.
+
+    .. versionadded:: 2.4
+
+    :param entitlement: The entitlement that was deleted.
+    :type entitlement: :class:`Entitlement`
+
+
 Gateway
 ~~~~~~~~
 
@@ -3429,6 +3470,47 @@ of :class:`enum.Enum`.
         The underlying type of the ID is a channel or thread.
 
 
+.. class:: SKUType
+
+    Represents the type of a SKU.
+
+    .. versionadded:: 2.4
+
+    .. attribute:: subscription
+
+        The SKU is a recurring subscription.
+
+    .. attribute:: subscription_group
+
+        The SKU is a system-generated group which is created for each :attr:`SKUType.subscription`.
+
+
+.. class:: EntitlementType
+
+    Represents the type of an entitlement.
+
+    .. versionadded:: 2.4
+
+    .. attribute:: application_subscription
+
+        The entitlement was purchased as an app subscription.
+
+
+.. class:: EntitlementOwnerType
+
+    Represents the type of an entitlement owner.
+
+    .. versionadded:: 2.4
+
+    .. attribute:: guild
+
+        The entitlement owner is a guild.
+
+    .. attribute:: user
+
+            The entitlement owner is a user.
+
+
 .. _discord-api-audit-logs:
 
 Audit Log Data
@@ -4714,6 +4796,30 @@ ShardInfo
 .. autoclass:: ShardInfo()
     :members:
 
+SKU
+~~~~~~~~~~~
+
+.. attributetable:: SKU
+
+.. autoclass:: SKU()
+    :members:
+
+SKUFlags
+~~~~~~~~~~~
+
+.. attributetable:: SKUFlags
+
+.. autoclass:: SKUFlags()
+    :members:
+
+Entitlement
+~~~~~~~~~~~
+
+.. attributetable:: Entitlement
+
+.. autoclass:: Entitlement()
+    :members:
+
 RawMessageDeleteEvent
 ~~~~~~~~~~~~~~~~~~~~~~~