Add guild scheduled event support (#42)

This commit is contained in:
Slipstream 2025-06-10 21:09:47 -06:00 committed by GitHub
parent 2d6c2cb0be
commit c250c22737
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 207 additions and 0 deletions

View File

@ -51,6 +51,7 @@ if TYPE_CHECKING:
Thread,
DMChannel,
Webhook,
ScheduledEvent,
AuditLogEntry,
Invite,
)
@ -709,6 +710,20 @@ class Client:
self._webhooks[webhook.id] = webhook
return webhook
def parse_scheduled_event(self, data: Dict[str, Any]) -> "ScheduledEvent":
"""Parses scheduled event data and updates cache."""
from .models import ScheduledEvent
event = ScheduledEvent(data, client_instance=self)
# Cache by ID under guild if guild cache exists
guild = self._guilds.get(event.guild_id)
if guild is not None:
events = getattr(guild, "_scheduled_events", {})
events[event.id] = event
setattr(guild, "_scheduled_events", events)
return event
def parse_audit_log_entry(self, data: Dict[str, Any]) -> "AuditLogEntry":
"""Parses audit log entry data."""
from .models import AuditLogEntry
@ -1298,6 +1313,64 @@ class Client:
await self._http.delete_webhook(webhook_id)
async def fetch_scheduled_events(
self, guild_id: Snowflake
) -> List["ScheduledEvent"]:
"""|coro| Fetch all scheduled events for a guild."""
if self._closed:
raise DisagreementException("Client is closed.")
data = await self._http.get_guild_scheduled_events(guild_id)
return [self.parse_scheduled_event(ev) for ev in data]
async def fetch_scheduled_event(
self, guild_id: Snowflake, event_id: Snowflake
) -> Optional["ScheduledEvent"]:
"""|coro| Fetch a single scheduled event."""
if self._closed:
raise DisagreementException("Client is closed.")
try:
data = await self._http.get_guild_scheduled_event(guild_id, event_id)
return self.parse_scheduled_event(data)
except DisagreementException as e:
print(f"Failed to fetch scheduled event {event_id}: {e}")
return None
async def create_scheduled_event(
self, guild_id: Snowflake, payload: Dict[str, Any]
) -> "ScheduledEvent":
"""|coro| Create a scheduled event in a guild."""
if self._closed:
raise DisagreementException("Client is closed.")
data = await self._http.create_guild_scheduled_event(guild_id, payload)
return self.parse_scheduled_event(data)
async def edit_scheduled_event(
self, guild_id: Snowflake, event_id: Snowflake, payload: Dict[str, Any]
) -> "ScheduledEvent":
"""|coro| Edit an existing scheduled event."""
if self._closed:
raise DisagreementException("Client is closed.")
data = await self._http.edit_guild_scheduled_event(guild_id, event_id, payload)
return self.parse_scheduled_event(data)
async def delete_scheduled_event(
self, guild_id: Snowflake, event_id: Snowflake
) -> None:
"""|coro| Delete a scheduled event."""
if self._closed:
raise DisagreementException("Client is closed.")
await self._http.delete_guild_scheduled_event(guild_id, event_id)
async def create_invite(
self, channel_id: Snowflake, payload: Dict[str, Any]
) -> "Invite":

View File

@ -278,6 +278,32 @@ class GuildFeature(str, Enum): # Changed from IntEnum to Enum
return str(value)
# --- Guild Scheduled Event Enums ---
class GuildScheduledEventPrivacyLevel(IntEnum):
"""Privacy level for a scheduled event."""
GUILD_ONLY = 2
class GuildScheduledEventStatus(IntEnum):
"""Status of a scheduled event."""
SCHEDULED = 1
ACTIVE = 2
COMPLETED = 3
CANCELED = 4
class GuildScheduledEventEntityType(IntEnum):
"""Entity type for a scheduled event."""
STAGE_INSTANCE = 1
VOICE = 2
EXTERNAL = 3
class VoiceRegion(str, Enum):
"""Voice region identifier."""

View File

@ -628,6 +628,49 @@ class HTTPClient:
"""Fetches a guild object for a given guild ID."""
return await self.request("GET", f"/guilds/{guild_id}")
async def get_guild_scheduled_events(
self, guild_id: "Snowflake"
) -> List[Dict[str, Any]]:
"""Returns a list of scheduled events for the guild."""
return await self.request("GET", f"/guilds/{guild_id}/scheduled-events")
async def get_guild_scheduled_event(
self, guild_id: "Snowflake", event_id: "Snowflake"
) -> Dict[str, Any]:
"""Returns a guild scheduled event."""
return await self.request(
"GET", f"/guilds/{guild_id}/scheduled-events/{event_id}"
)
async def create_guild_scheduled_event(
self, guild_id: "Snowflake", payload: Dict[str, Any]
) -> Dict[str, Any]:
"""Creates a guild scheduled event."""
return await self.request(
"POST", f"/guilds/{guild_id}/scheduled-events", payload=payload
)
async def edit_guild_scheduled_event(
self, guild_id: "Snowflake", event_id: "Snowflake", payload: Dict[str, Any]
) -> Dict[str, Any]:
"""Edits a guild scheduled event."""
return await self.request(
"PATCH",
f"/guilds/{guild_id}/scheduled-events/{event_id}",
payload=payload,
)
async def delete_guild_scheduled_event(
self, guild_id: "Snowflake", event_id: "Snowflake"
) -> None:
"""Deletes a guild scheduled event."""
await self.request("DELETE", f"/guilds/{guild_id}/scheduled-events/{event_id}")
async def get_audit_logs(
self, guild_id: "Snowflake", **filters: Any
) -> Dict[str, Any]:

View File

@ -23,6 +23,9 @@ from .enums import ( # These enums will need to be defined in disagreement/enum
ChannelType,
ComponentType,
ButtonStyle, # Added for Button
GuildScheduledEventPrivacyLevel,
GuildScheduledEventStatus,
GuildScheduledEventEntityType,
# SelectMenuType will be part of ComponentType or a new enum if needed
)
from .permissions import Permissions
@ -2058,6 +2061,42 @@ class Reaction:
return f"<Reaction message_id='{self.message_id}' user_id='{self.user_id}' emoji='{emoji_value}'>"
class ScheduledEvent:
"""Represents a guild scheduled event."""
def __init__(
self, data: Dict[str, Any], client_instance: Optional["Client"] = None
):
self._client = client_instance
self.id: str = data["id"]
self.guild_id: str = data["guild_id"]
self.channel_id: Optional[str] = data.get("channel_id")
self.creator_id: Optional[str] = data.get("creator_id")
self.name: str = data["name"]
self.description: Optional[str] = data.get("description")
self.scheduled_start_time: str = data["scheduled_start_time"]
self.scheduled_end_time: Optional[str] = data.get("scheduled_end_time")
self.privacy_level: GuildScheduledEventPrivacyLevel = (
GuildScheduledEventPrivacyLevel(data["privacy_level"])
)
self.status: GuildScheduledEventStatus = GuildScheduledEventStatus(
data["status"]
)
self.entity_type: GuildScheduledEventEntityType = GuildScheduledEventEntityType(
data["entity_type"]
)
self.entity_id: Optional[str] = data.get("entity_id")
self.entity_metadata: Optional[Dict[str, Any]] = data.get("entity_metadata")
self.creator: Optional[User] = (
User(data["creator"]) if data.get("creator") else None
)
self.user_count: Optional[int] = data.get("user_count")
self.image: Optional[str] = data.get("image")
def __repr__(self) -> str:
return f"<ScheduledEvent id='{self.id}' name='{self.name}'>"
@dataclass
class Invite:
"""Represents a Discord invite."""

26
docs/scheduled_events.md Normal file
View File

@ -0,0 +1,26 @@
# Guild Scheduled Events
The `Client` provides helpers to manage guild scheduled events.
```python
from disagreement.client import Client
client = Client(token="TOKEN")
payload = {
"name": "Movie Night",
"scheduled_start_time": "2024-05-01T20:00:00Z",
"privacy_level": 2,
"entity_type": 3,
"entity_metadata": {"location": "https://discord.gg/example"},
}
event = await client.create_scheduled_event(123456789012345678, payload)
print(event.id, event.name)
```
## Next Steps
- [Commands](commands.md)
- [Caching](caching.md)
- [Voice Features](voice_features.md)