feat: application subscriptions

changelog/1113.feature.rst

@@ -0,0 +1,5 @@
+Support application subscriptions (see the :ddocs:`official docs <monetization/overview>` for more info).
+- New types: :class:`SKU`, :class:`Entitlement`.
+- New :attr:`Interaction.entitlements` attribute, and :meth:`InteractionResponse.require_premium` response type.
+- New events: :func:`on_entitlement_create`, :func:`on_entitlement_update`, :func:`on_entitlement_delete`.
+- New :class:`Client` methods: :meth:`~Client.skus`, :meth:`~Client.entitlements`, :meth:`~Client.create_entitlement`.

disnake/__init__.py

@@ -37,6 +37,7 @@
 from .custom_warnings import *
 from .embeds import *
 from .emoji import *
+from .entitlement import *
 from .enums import *
 from .errors import *
 from .file import *
@@ -60,6 +61,7 @@
 from .reaction import *
 from .role import *
 from .shard import *
+from .sku import *
 from .stage_instance import *
 from .sticker import *
 from .team import *

disnake/client.py

@@ -33,7 +33,7 @@
 
 import aiohttp
 
-from . import utils
+from . import abc, utils
 from .activity import ActivityTypes, BaseActivity, create_activity
 from .app_commands import (
     APIMessageCommand,
@@ -47,6 +47,7 @@
 from .backoff import ExponentialBackoff
 from .channel import PartialMessageable, _threaded_channel_factory
 from .emoji import Emoji
+from .entitlement import Entitlement
 from .enums import ApplicationCommandType, ChannelType, Event, Status
 from .errors import (
     ConnectionClosed,
@@ -63,9 +64,10 @@
 from .http import HTTPClient
 from .i18n import LocalizationProtocol, LocalizationStore
 from .invite import Invite
-from .iterators import GuildIterator
+from .iterators import EntitlementIterator, GuildIterator
 from .mentions import AllowedMentions
 from .object import Object
+from .sku import SKU
 from .stage_instance import StageInstance
 from .state import ConnectionState
 from .sticker import GuildSticker, StandardSticker, StickerPack, _sticker_factory
@@ -2989,3 +2991,133 @@ async def edit_role_connection_metadata(
             self.application_id, payload
         )
         return [ApplicationRoleConnectionMetadata._from_data(record) for record in data]
+
+    async def skus(self) -> List[SKU]:
+        """|coro|
+
+        Retrieves the :class:`.SKU`\\s for the application.
+
+        To manage application subscription entitlements, you should use the SKU
+        with :attr:`.SKUType.subscription`.
+
+        .. versionadded:: 2.10
+
+        Raises
+        ------
+        HTTPException
+            Retrieving the SKUs failed.
+
+        Returns
+        -------
+        List[:class:`.SKU`]
+            The list of SKUs.
+        """
+        data = await self.http.get_skus(self.application_id)
+        return [SKU(data=d) for d in data]
+
+    def entitlements(
+        self,
+        *,
+        limit: Optional[int] = 100,
+        before: Optional[SnowflakeTime] = None,
+        after: Optional[SnowflakeTime] = None,
+        user: Optional[Snowflake] = None,
+        guild: Optional[Snowflake] = None,
+        skus: Optional[Sequence[Snowflake]] = None,
+        exclude_ended: bool = False,
+        oldest_first: bool = False,
+    ) -> EntitlementIterator:
+        """Retrieves an :class:`.AsyncIterator` that enables receiving entitlements for the application.
+
+        .. note::
+
+            This method is an API call. To get the entitlements of the invoking user/guild
+            in interactions, consider using :attr:`.Interaction.entitlements`.
+
+        Entries are returned in order from newest to oldest by default;
+        pass ``oldest_first=True`` to reverse the iteration order.
+
+        All parameters are optional.
+
+        .. versionadded:: 2.10
+
+        Parameters
+        ----------
+        limit: Optional[:class:`int`]
+            The number of entitlements to retrieve.
+            If ``None``, retrieves every entitlement.
+            Note, however, that this would make it a slow operation.
+            Defaults to ``100``.
+        before: Union[:class:`.abc.Snowflake`, :class:`datetime.datetime`]
+            Retrieves entitlements created before this date or object.
+            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: Union[:class:`.abc.Snowflake`, :class:`datetime.datetime`]
+            Retrieve entitlements created after this date or object.
+            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.
+        user: Optional[:class:`.abc.Snowflake`]
+            The user to retrieve entitlements for.
+        guild: Optional[:class:`.abc.Snowflake`]
+            The guild to retrieve entitlements for.
+        skus: Optional[Sequence[:class:`.abc.Snowflake`]]
+            The SKUs for which entitlements are retrieved.
+        exclude_ended: :class:`bool`
+            Whether to exclude ended/expired entitlements. Defaults to ``False``.
+        oldest_first: :class:`bool`
+            If set to ``True``, return entries in oldest->newest order. Defaults to ``False``.
+
+        Raises
+        ------
+        HTTPException
+            Retrieving the entitlements failed.
+
+        Yields
+        ------
+        :class:`.Entitlement`
+            The entitlements for the given parameters.
+        """
+        return EntitlementIterator(
+            self.application_id,
+            state=self._connection,
+            limit=limit,
+            before=before,
+            after=after,
+            user_id=user.id if user is not None else None,
+            guild_id=guild.id if guild is not None else None,
+            sku_ids=[sku.id for sku in skus] if skus else None,
+            exclude_ended=exclude_ended,
+            oldest_first=oldest_first,
+        )
+
+    async def create_entitlement(
+        self, sku: Snowflake, owner: Union[abc.User, Guild]
+    ) -> Entitlement:
+        """|coro|
+
+        Creates a new test :class:`.Entitlement` for the given user or guild, with no expiry.
+
+        Parameters
+        ----------
+        sku: :class:`.abc.Snowflake`
+            The :class:`.SKU` to grant the entitlement for.
+        owner: Union[:class:`.abc.User`, :class:`.Guild`]
+            The user or guild to grant the entitlement to.
+
+        Raises
+        ------
+        HTTPException
+            Creating the entitlement failed.
+
+        Returns
+        -------
+        :class:`.Entitlement`
+            The newly created entitlement.
+        """
+        data = await self.http.create_test_entitlement(
+            self.application_id,
+            sku_id=sku.id,
+            owner_id=owner.id,
+            owner_type=2 if isinstance(owner, abc.User) else 1,
+        )
+        return Entitlement(data=data, state=self._connection)

disnake/entitlement.py

@@ -0,0 +1,173 @@
+# SPDX-License-Identifier: MIT
+
+from __future__ import annotations
+
+import datetime
+from typing import TYPE_CHECKING, Optional
+
+from .enums import EntitlementType, try_enum
+from .mixins import Hashable
+from .utils import _get_as_snowflake, parse_time, snowflake_time, utcnow
+
+if TYPE_CHECKING:
+    from .guild import Guild
+    from .state import ConnectionState
+    from .types.entitlement import Entitlement as EntitlementPayload
+    from .user import User
+
+
+__all__ = ("Entitlement",)
+
+
+class Entitlement(Hashable):
+    """Represents an entitlement.
+
+    This is usually retrieved using :meth:`Client.entitlements`, from
+    :attr:`Interaction.entitlements` when using interactions, or provided by
+    events (e.g. :func:`on_entitlement_create`).
+
+    Note that some entitlements may have ended already; consider using
+    :meth:`is_active` to check whether a given entitlement is considered active at the current time,
+    or use ``exclude_ended=True`` when fetching entitlements using :meth:`Client.entitlements`.
+
+    You may create new entitlements for testing purposes using :meth:`Client.create_entitlement`.
+
+    .. collapse:: operations
+
+        .. describe:: x == y
+
+            Checks if two :class:`Entitlement`\\s are equal.
+
+        .. describe:: x != y
+
+            Checks if two :class:`Entitlement`\\s are not equal.
+
+        .. describe:: hash(x)
+
+            Returns the entitlement's hash.
+
+    .. versionadded:: 2.10
+
+    Attributes
+    ----------
+    id: :class:`int`
+        The entitlement's ID.
+    type: :class:`EntitlementType`
+        The entitlement's type.
+    sku_id: :class:`int`
+        The ID of the associated SKU.
+    user_id: Optional[:class:`int`]
+        The ID of the user that is granted access to the entitlement's SKU.
+
+        See also :attr:`user`.
+    guild_id: Optional[:class:`int`]
+        The ID of the guild that is granted access to the entitlement's SKU.
+
+        See also :attr:`guild`.
+    application_id: :class:`int`
+        The parent application's ID.
+    deleted: :class:`bool`
+        Whether the entitlement was deleted.
+    starts_at: Optional[:class:`datetime.datetime`]
+        The time at which the entitlement starts being active.
+        Set to ``None`` when this is a test entitlement.
+    ends_at: Optional[:class:`datetime.datetime`]
+        The time at which the entitlement stops being active.
+        Set to ``None`` when this is a test entitlement.
+
+        You can use :meth:`is_active` to check whether this entitlement is still active.
+    """
+
+    __slots__ = (
+        "_state",
+        "id",
+        "sku_id",
+        "user_id",
+        "guild_id",
+        "application_id",
+        "type",
+        "deleted",
+        "starts_at",
+        "ends_at",
+    )
+
+    def __init__(self, *, data: EntitlementPayload, state: ConnectionState) -> None:
+        self._state: ConnectionState = state
+
+        self.id: int = int(data["id"])
+        self.sku_id: int = int(data["sku_id"])
+        self.user_id: Optional[int] = _get_as_snowflake(data, "user_id")
+        self.guild_id: Optional[int] = _get_as_snowflake(data, "guild_id")
+        self.application_id: int = int(data["application_id"])
+        self.type: EntitlementType = try_enum(EntitlementType, data["type"])
+        self.deleted: bool = data.get("deleted", False)
+        self.starts_at: Optional[datetime.datetime] = parse_time(data.get("starts_at"))
+        self.ends_at: Optional[datetime.datetime] = parse_time(data.get("ends_at"))
+
+    def __repr__(self) -> str:
+        # presumably one of these is set
+        if self.user_id:
+            grant_repr = f"user_id={self.user_id!r}"
+        else:
+            grant_repr = f"guild_id={self.guild_id!r}"
+        return (
+            f"<Entitlement id={self.id!r} sku_id={self.sku_id!r} type={self.type!r} {grant_repr}>"
+        )
+
+    @property
+    def created_at(self) -> datetime.datetime:
+        """:class:`datetime.datetime`: Returns the entitlement's creation time in UTC."""
+        return snowflake_time(self.id)
+
+    @property
+    def guild(self) -> Optional[Guild]:
+        """Optional[:class:`Guild`]: The guild that is granted access to
+        this entitlement's SKU, if applicable.
+        """
+        return self._state._get_guild(self.guild_id)
+
+    @property
+    def user(self) -> Optional[User]:
+        """Optional[:class:`User`]: The user that is granted access to
+        this entitlement's SKU, if applicable.
+
+        Requires the user to be cached.
+        See also :attr:`user_id`.
+        """
+        return self._state.get_user(self.user_id)
+
+    def is_active(self) -> bool:
+        """Whether the entitlement is currently active,
+        based on :attr:`starts_at` and :attr:`ends_at`.
+
+        Always returns ``True`` for test entitlements.
+
+        :return type: :class:`bool`
+        """
+        if self.deleted:
+            return False
+
+        now = utcnow()
+        if self.starts_at is not None and now < self.starts_at:
+            return False
+        if self.ends_at is not None and now >= self.ends_at:
+            return False
+
+        return True
+
+    async def delete(self) -> None:
+        """|coro|
+
+        Deletes the entitlement.
+
+        This is only valid for test entitlements; you cannot use this to
+        delete entitlements that users purchased.
+
+        Raises
+        ------
+        NotFound
+            The entitlement does not exist.
+        HTTPException
+            Deleting the entitlement failed.
+        """
+        await self._state.http.delete_test_entitlement(self.application_id, self.id)