✨ Add guild sheduled event endpoints

pincer/objects/guild/__init__.py

@@ -23,7 +23,7 @@
 from .overwrite import Overwrite
 from .permissions import Permissions
 from .role import RoleTags, Role
-from .scheduled_events import GuildScheduledEventEntityType, EventStatus, ScheduledEvent
+from .scheduled_events import GuildScheduledEventEntityType, GuildScheduledEventUser, EventStatus, ScheduledEvent
 from .stage import PrivacyLevel, StageInstance
 from .template import GuildTemplate
 from .thread import ThreadMetadata, ThreadMember
@@ -44,5 +44,6 @@
     "PrivacyLevel", "Role", "RoleTags", "ScheduledEvent", "StageInstance",
     "SystemChannelFlags", "TextChannel", "ThreadMember", "ThreadMetadata",
     "UnavailableGuild", "VerificationLevel", "VoiceChannel", "Webhook",
-    "WebhookType", "WelcomeScreen", "WelcomeScreenChannel"
+    "WebhookType", "WelcomeScreen", "WelcomeScreenChannel",
+    "WelcomeScreenChannelGuildScheduledEventEntityType"
 )

pincer/objects/guild/guild.py

@@ -4,14 +4,17 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
+from datetime import datetime
 from enum import IntEnum
 from typing import AsyncGenerator, overload, TYPE_CHECKING
 
 from aiohttp import FormData
 
 from .channel import Channel, Thread
+from .scheduled_events import ScheduledEvent, GuildScheduledEventUser
 from ..message.file import File
 from ...exceptions import UnavailableGuildError
+from ...utils import remove_none
 from ...utils.api_object import APIObject
 from ...utils.types import MISSING
 
@@ -27,7 +30,7 @@
     from .invite import Invite
     from .overwrite import Overwrite
     from .role import Role
-    from .scheduled_events import ScheduledEvent
+    from .scheduled_events import ScheduledEvent, GuildScheduledEventUser
     from .stage import StageInstance
     from .template import GuildTemplate
     from .welcome_screen import WelcomeScreen, WelcomeScreenChannel
@@ -1916,6 +1919,141 @@ async def get_webhooks(self) -> AsyncGenerator[Webhook, None]:
         for webhook_data in data:
             yield Webhook.from_dict(webhook_data)
 
+    async def get_scheduled_events(
+        self, with_user_count: bool = False
+    ) -> AsyncGenerator[ScheduledEvent, None]:
+        data = await self._http.get(
+            f"guilds/{self.id}/scheduled-events?{with_user_count=!s}"
+        )
+        for event_data in data:
+            yield ScheduledEvent.from_dict(event_data)
+
+    async def create_scheduled_event(
+        self,
+        name: str,
+        privacy_level: int,
+        entity_type: int,
+        scheduled_start_time: datetime,
+        scheduled_end_time: Optional[datetime] = None,
+        entity_metadata: Optional[str] = None,
+        channel_id: Optional[Snowflake] = None,
+        description: Optional[str] = None,
+        reason: Optional[str] = None,
+    ) -> ScheduledEvent:
+        if scheduled_start_time.timestamp() < datetime.now().timestamp():
+            raise ValueError("An event cannot be created in the past")
+
+        if (
+            scheduled_end_time is not None
+            and scheduled_end_time.timestamp() < datetime.now().timestamp()
+        ):
+            raise ValueError("An event cannot start before it ends")
+
+        data = await self._http.post(
+            f"guilds/{self.id}/scheduled-events",
+            data={
+                "name": name,
+                "privacy_level": privacy_level,
+                "entity_type": entity_type,
+                "scheduled_start_time": scheduled_start_time.isoformat(),
+                "scheduled_end_time": scheduled_end_time.isoformat()
+                if scheduled_end_time is not None
+                else None,
+                "entity_metadata": entity_metadata,
+                "channel_id": channel_id,
+                "description": description,
+            },
+            headers={"X-Audit-Log-Reason": reason},
+        )
+        return ScheduledEvent.from_dict(data)
+
+    async def get_scheduled_event(
+        self, _id: Snowflake, with_user_count: bool = False
+    ) -> ScheduledEvent:
+        data = await self._http.get(
+            f"guilds/{self.id}/scheduled-events/{_id}",
+            params={"with_user_count": with_user_count},
+        )
+        return ScheduledEvent.from_dict(data)
+
+    async def modify_scheduled_event(
+        self,
+        _id: Snowflake,
+        name: Optional[str] = None,
+        entity_type: Optional[int] = None,
+        privacy_level: Optional[int] = None,
+        scheduled_start_time: Optional[datetime] = None,
+        scheduled_end_time: Optional[datetime] = None,
+        entity_metadata: Optional[str] = None,
+        channel_id: Optional[Snowflake] = None,
+        description: Optional[str] = None,
+        status: Optional[int] = None,
+        reason: Optional[str] = None,
+    ) -> ScheduledEvent:
+        if scheduled_start_time:
+            timestamp = scheduled_start_time.timestamp()
+            if timestamp < datetime.now().timestamp():
+                raise ValueError("An event cannot be created in the past")
+
+            if (
+                scheduled_end_time
+                and timestamp > scheduled_end_time.timestamp()
+            ):
+                raise ValueError("An event cannot start before it ends")
+
+        kwargs: Dict[str, str] = remove_none(
+            {
+                "name": name,
+                "privacy_level": privacy_level,
+                "entity_type": entity_type,
+                "scheduled_start_time": scheduled_start_time.isoformat()
+                if scheduled_start_time is not None
+                else None,
+                "scheduled_end_time": scheduled_end_time.isoformat()
+                if scheduled_end_time is not None
+                else None,
+                "entity_metadata": entity_metadata,
+                "channel_id": channel_id,
+                "description": description,
+                "status": status,
+            }
+        )
+
+        data = await self._http.patch(
+            f"guilds/{self.id}/scheduled-events/{_id}",
+            data=kwargs,
+            headers={"X-Audit-Log-Reason": reason},
+        )
+        return ScheduledEvent.from_dict(data)
+
+    async def delete_scheduled_event(self, _id: Union[int, Snowflake]):
+        await self._http.delete(f"guilds/{self.id}/scheduled-events/{_id}")
+
+    async def get_guild_scheduled_event_users(
+        self,
+        _id: Snowflake,
+        limit: int = 100,
+        with_member: bool = False,
+        before: Optional[Snowflake] = None,
+        after: Optional[Snowflake] = None,
+    ) -> AsyncGenerator[GuildScheduledEventUser, None]:
+        params = {
+            "limit": limit,
+            "with_member": with_member,
+        }
+        if before is not None:
+            params.update({"before": before})
+        if after is not None:
+            params.update({"after": after})
+
+        data = await self._http.get(
+            f"guilds/{self.id}/scheduled-events/{_id}/users?",
+            params=params,
+        )
+
+        for user_data in data:
+            yield GuildScheduledEventUser.from_dict(user_data)
+
     @classmethod
     def from_dict(cls, data) -> Guild:
         """

pincer/objects/guild/scheduled_events.py

@@ -11,6 +11,7 @@
 
 if TYPE_CHECKING:
     from ..guild.stage import PrivacyLevel
+    from ..guild.member import GuildMember
     from ..user.user import User
     from ...utils.snowflake import Snowflake
     from ...utils.timestamp import Timestamp
@@ -77,22 +78,22 @@ class ScheduledEvent(APIObject):
         The status of the scheduled event.
     entity_type: :class:`~pincer.guild.schedule_events.GuildScheduledEventEntityType`
         The type of the scheduled event
-    channel_id: :class:`int`
+    channel_id: APINullable[:class:`int`]
         The channel id in which the scheduled event will be hosted,
         or null if scheduled entity type is EXTERNAL
-    creator_id: :class:`int`
+    creator_id: APINullable[:class:`int`]
         The user id of the creator of the scheduled event
     scheduled_end_time: str
         The time the scheduled event will end, required if entity_type is EXTERNAL
-    description: :class:`str`
+    description: APINullable[:class:`str`]
         The description of the scheduled event (0-1000 characters)
-    entity_id: :class:`int`
+    entity_id: APINullable[:class:`int`]
         The id of an entity associated with a guild scheduled event
-    entity_metadata: :class:`str`
+    entity_metadata: APINullable[:class:`str`]
         Additional metadata for the guild scheduled event
-    creator: :class:`~pincer.objects.user.user.User`
+    creator: APINullable[:class:`~pincer.objects.user.user.User`]
         The user who created the scheduled event
-    user_count: :class:`int`
+    user_count: APINullable[:class:`int`]
         The number of users who have joined the scheduled event
     """
     id: Snowflake
@@ -112,3 +113,22 @@ class ScheduledEvent(APIObject):
     entity_metadata: APINullable[str] = MISSING
     creator: APINullable[User] = MISSING
     user_count: APINullable[int] = MISSING
+
+
+@dataclass
+class GuildScheduledEventUser(APIObject):
+    """
+    Represents a user who has joined a scheduled event.
+
+    Attributes
+    ----------
+    guild_scheduled_event_id: :class:`int`
+        the scheduled event id which the user subscribed to
+    user : :class:`~pincer.objects.user.user.User`
+        user which subscribed to an event
+    member : APINullable[:class:`~pincer.objects.guild.member.GuildMember`]
+        	guild member data for this user for the guild which this event belongs to, if any
+    """
+    guild_scheduled_event_id: Snowflake
+    user: User
+    member: APINullable[GuildMember] = MISSING