Skip to content

Managers API Reference

classes.managers.GuildManager

GuildManager

Manage guild-level data, settings materialization, and caching.

Responsibilities
  • Ensure a guild document exists in MongoDB.
  • Merge module-declared default settings with DB values.
  • Maintain an in-memory settings cache per guild.
Source code in classes/managers/GuildManager.py
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
class GuildManager:
    """Manage guild-level data, settings materialization, and caching.

    Responsibilities:
        - Ensure a guild document exists in MongoDB.
        - Merge module-declared default settings with DB values.
        - Maintain an in-memory settings cache per guild.
    """

    def __init__(self, client: ExtendedClient, logger: logging.Logger):
        self.client = client
        self.setting_cache: Dict[str, Dict[str, Setting]] = {}
        self.logger = logger

    async def fetch_or_create(self, guild_id: str, force: bool = False) -> Guild:
        """Fetch the Guild object with materialized settings, creating DB docs as needed.

        Loads the Discord guild from cache or API, ensures a guild profile exists
        in the `guilds` collection, and resolves the full settings map by combining
        module defaults and stored values.

        Args:
            guild_id: The Discord guild id (stringable).
            force: Reserved for future use (e.g., bypass certain caches).

        Returns:
            A :class:`Guild` wrapper with `guild`, `data`, and `settings` populated.

        Raises:
            ValueError: If the Discord guild cannot be found.
        """
        guild_id = str(guild_id)
        if not self.client.is_ready():
            await self.client.wait_until_ready()

        # Fetch or load the guild
        guild = self.client.get_guild(int(guild_id)) or await self.client.fetch_guild(int(guild_id))
        if not guild:
            raise ValueError(f"Guild with ID {guild_id} not found.")


        # Fetch or initialize guild data from database
        guild_data = await self.fetch_guild_data(guild_id)
        if not guild_data:
            guild_data = {"_id": guild_id, "settings": {}, "permissionsOverrides": {}}
            await self.create_guild_data(guild_data)

        # Fetch settings
        settings = self.client.setting_cache.get(guild_id) or await self._get_all_settings(guild_data, guild)
        self.client.setting_cache[guild_id] = settings

        return Guild(self.client, guild, guild_data, settings)

    async def fetch_all_members(self, guild_id: int) -> List[Dict[str, Any]]:
        """Fetching all members for a given guild.

        Args:
            guild_id: The Discord guild id.

        Returns:
            A list of member profile dicts or manager-defined structures.
        """
        gid = str(guild_id)
        return await self.client.db.find("members", {"guildId": gid})

    async def fetch_guild_data(self, guild_id: str) -> Dict[str, Any]:
        """Fetch the guild document from MongoDB.

        Args:
            guild_id: The Discord guild id.

        Returns:
            The guild document or None if not found.
        """
        return await self.client.db.find_one("guilds", {"_id": guild_id})

    async def create_guild_data(self, guild_data: Dict[str, Any]):
        """Insert a new guild document in MongoDB.

        Args:
            guild_data: Initial document to insert.

        Returns:
            The inserted id or driver-specific result.
        """
        return await self.client.db.insert_one("guilds", guild_data)

    async def get_language(self, guild_id: str) -> str:
        """Return the guild's language from settings, defaulting to 'en'.

        Args:
            guild_id: The Discord guild id.

        Returns:
            The ISO code of the language, e.g., "en".
        """
        guild_data = await self.fetch_or_create(guild_id)

        return guild_data.data.get("settings", {}).get("language", "en")

    async def _get_all_settings(self, guild_data: Dict[str, Any], guild: DiscordGuild) -> Dict[str, Setting]:
        """Materialize the full settings map for a guild.

        Combines module-declared settings with values retrieved from the database.
        If a default value is missing in the DB, parses and stores the default.

        Args:
            guild_data: The guild document from MongoDB.
            guild: The Discord guild.

        Returns:
            A mapping from setting id to `Setting` instances with `value` resolved.
        """
        settings_map = {}
        db_settings = guild_data.get("settings", {})

        for module in self.client.modules.values():
            for default_setting in module.settings:
                setting = default_setting.clone()
                self.logger.info(f"Cloned setting: {setting}")

                db_value = db_settings.get(setting.id)
                if db_value is not None:
                    try:
                        setting.value = await setting.parse(db_value, self.client, guild_data, guild)
                        self.logger.debug(f"Loaded setting '{setting.id}' with value: {setting.value}")
                    except Exception as e:
                        self.logger.error(f"Failed to parse setting '{setting.id}' from database: {e}")
                else:
                    try:
                        default_value = await setting.parse(setting.value, self.client, guild_data, guild)
                        self.logger.debug(f"Default value for '{setting.id}': {default_value}")

                        if not isinstance(default_value, (str, int, float, list, dict, bool, type(None))):
                            raise ValueError(f"Invalid default_value type for setting '{setting.id}': {type(default_value).__name__}")

                        query = {f"settings.{setting.id}": default_value}
                        self.logger.debug(f"Update query for guild {guild.id}: {query}")

                        await self.client.db.update_one(
                            "guilds",
                            {"_id": guild_data["_id"]},  # Filtro para encontrar o documento
                            query,  # Atualização com $set
                            upsert=True  # Garantir que ele crie o documento se não existir
                        )
                        setting.value = setting.value  # Valor padrão
                        self.logger.info(f"Created default setting '{setting.id}' for guild {guild.id}")
                    except Exception as e:
                        self.logger.error(f"Failed to create default setting '{setting.id}': {e}")

                settings_map[setting.id] = setting

        return settings_map


    async def find_by_kv(self, filter: Dict[str, Any]) -> List[Guild]:
        """Find guilds matching a filter and return their materialized profiles.

        Prefers `_id` as the guild id key but tolerates legacy documents using `id`.
        Each result is hydrated with settings (merged defaults + DB values) and
        cached under `setting_cache` keyed by the string guild id.

        Args:
            filter: Mongo‑style filter applied to the `guilds` collection.

        Returns:
            A list of `Guild` wrappers with `guild`, `data`, and `settings` populated.
        """
        guild_profiles = await self.client.db.find("guilds", filter)
        guilds = []
        for profile in guild_profiles:
            gid = str(profile.get("_id") or profile.get("id"))  # tolerate both, prefer _id
            if not gid:
                continue
            guild = self.client.get_guild(int(gid)) or await self.client.fetch_guild(int(gid))
            if not guild:
                continue
            settings = self.setting_cache.get(gid) or await self._get_all_settings(profile, guild)
            self.setting_cache[gid] = settings
            guilds.append(Guild(self.client, guild, profile, settings))

        self.logger.info(f"Found {len(guilds)} guilds matching filter {filter}.")
        return guilds

    def invalidate_cache(self, guild_id: str):
        """Invalidate the in-memory settings cache for a guild.

        Args:
            guild_id: The Discord guild id (stringable).
        """
        if guild_id in self.setting_cache:
            del self.setting_cache[guild_id]
            self.logger.debug(f"Invalidated cache for guild {guild_id}.")

create_guild_data(guild_data) async

Insert a new guild document in MongoDB.

Parameters:

Name Type Description Default
guild_data Dict[str, Any]

Initial document to insert.

required

Returns:

Type Description

The inserted id or driver-specific result.

Source code in classes/managers/GuildManager.py
86
87
88
89
90
91
92
93
94
95
async def create_guild_data(self, guild_data: Dict[str, Any]):
    """Insert a new guild document in MongoDB.

    Args:
        guild_data: Initial document to insert.

    Returns:
        The inserted id or driver-specific result.
    """
    return await self.client.db.insert_one("guilds", guild_data)

fetch_all_members(guild_id) async

Fetching all members for a given guild.

Parameters:

Name Type Description Default
guild_id int

The Discord guild id.

required

Returns:

Type Description
List[Dict[str, Any]]

A list of member profile dicts or manager-defined structures.

Source code in classes/managers/GuildManager.py
63
64
65
66
67
68
69
70
71
72
73
async def fetch_all_members(self, guild_id: int) -> List[Dict[str, Any]]:
    """Fetching all members for a given guild.

    Args:
        guild_id: The Discord guild id.

    Returns:
        A list of member profile dicts or manager-defined structures.
    """
    gid = str(guild_id)
    return await self.client.db.find("members", {"guildId": gid})

fetch_guild_data(guild_id) async

Fetch the guild document from MongoDB.

Parameters:

Name Type Description Default
guild_id str

The Discord guild id.

required

Returns:

Type Description
Dict[str, Any]

The guild document or None if not found.

Source code in classes/managers/GuildManager.py
75
76
77
78
79
80
81
82
83
84
async def fetch_guild_data(self, guild_id: str) -> Dict[str, Any]:
    """Fetch the guild document from MongoDB.

    Args:
        guild_id: The Discord guild id.

    Returns:
        The guild document or None if not found.
    """
    return await self.client.db.find_one("guilds", {"_id": guild_id})

fetch_or_create(guild_id, force=False) async

Fetch the Guild object with materialized settings, creating DB docs as needed.

Loads the Discord guild from cache or API, ensures a guild profile exists in the guilds collection, and resolves the full settings map by combining module defaults and stored values.

Parameters:

Name Type Description Default
guild_id str

The Discord guild id (stringable).

required
force bool

Reserved for future use (e.g., bypass certain caches).

False

Returns:

Name Type Description
A Guild

class:Guild wrapper with guild, data, and settings populated.

Raises:

Type Description
ValueError

If the Discord guild cannot be found.

Source code in classes/managers/GuildManager.py
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
async def fetch_or_create(self, guild_id: str, force: bool = False) -> Guild:
    """Fetch the Guild object with materialized settings, creating DB docs as needed.

    Loads the Discord guild from cache or API, ensures a guild profile exists
    in the `guilds` collection, and resolves the full settings map by combining
    module defaults and stored values.

    Args:
        guild_id: The Discord guild id (stringable).
        force: Reserved for future use (e.g., bypass certain caches).

    Returns:
        A :class:`Guild` wrapper with `guild`, `data`, and `settings` populated.

    Raises:
        ValueError: If the Discord guild cannot be found.
    """
    guild_id = str(guild_id)
    if not self.client.is_ready():
        await self.client.wait_until_ready()

    # Fetch or load the guild
    guild = self.client.get_guild(int(guild_id)) or await self.client.fetch_guild(int(guild_id))
    if not guild:
        raise ValueError(f"Guild with ID {guild_id} not found.")


    # Fetch or initialize guild data from database
    guild_data = await self.fetch_guild_data(guild_id)
    if not guild_data:
        guild_data = {"_id": guild_id, "settings": {}, "permissionsOverrides": {}}
        await self.create_guild_data(guild_data)

    # Fetch settings
    settings = self.client.setting_cache.get(guild_id) or await self._get_all_settings(guild_data, guild)
    self.client.setting_cache[guild_id] = settings

    return Guild(self.client, guild, guild_data, settings)

find_by_kv(filter) async

Find guilds matching a filter and return their materialized profiles.

Prefers _id as the guild id key but tolerates legacy documents using id. Each result is hydrated with settings (merged defaults + DB values) and cached under setting_cache keyed by the string guild id.

Parameters:

Name Type Description Default
filter Dict[str, Any]

Mongo‑style filter applied to the guilds collection.

required

Returns:

Type Description
List[Guild]

A list of Guild wrappers with guild, data, and settings populated.

Source code in classes/managers/GuildManager.py
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
async def find_by_kv(self, filter: Dict[str, Any]) -> List[Guild]:
    """Find guilds matching a filter and return their materialized profiles.

    Prefers `_id` as the guild id key but tolerates legacy documents using `id`.
    Each result is hydrated with settings (merged defaults + DB values) and
    cached under `setting_cache` keyed by the string guild id.

    Args:
        filter: Mongo‑style filter applied to the `guilds` collection.

    Returns:
        A list of `Guild` wrappers with `guild`, `data`, and `settings` populated.
    """
    guild_profiles = await self.client.db.find("guilds", filter)
    guilds = []
    for profile in guild_profiles:
        gid = str(profile.get("_id") or profile.get("id"))  # tolerate both, prefer _id
        if not gid:
            continue
        guild = self.client.get_guild(int(gid)) or await self.client.fetch_guild(int(gid))
        if not guild:
            continue
        settings = self.setting_cache.get(gid) or await self._get_all_settings(profile, guild)
        self.setting_cache[gid] = settings
        guilds.append(Guild(self.client, guild, profile, settings))

    self.logger.info(f"Found {len(guilds)} guilds matching filter {filter}.")
    return guilds

get_language(guild_id) async

Return the guild's language from settings, defaulting to 'en'.

Parameters:

Name Type Description Default
guild_id str

The Discord guild id.

required

Returns:

Type Description
str

The ISO code of the language, e.g., "en".

Source code in classes/managers/GuildManager.py
 97
 98
 99
100
101
102
103
104
105
106
107
108
async def get_language(self, guild_id: str) -> str:
    """Return the guild's language from settings, defaulting to 'en'.

    Args:
        guild_id: The Discord guild id.

    Returns:
        The ISO code of the language, e.g., "en".
    """
    guild_data = await self.fetch_or_create(guild_id)

    return guild_data.data.get("settings", {}).get("language", "en")

invalidate_cache(guild_id)

Invalidate the in-memory settings cache for a guild.

Parameters:

Name Type Description Default
guild_id str

The Discord guild id (stringable).

required
Source code in classes/managers/GuildManager.py
194
195
196
197
198
199
200
201
202
def invalidate_cache(self, guild_id: str):
    """Invalidate the in-memory settings cache for a guild.

    Args:
        guild_id: The Discord guild id (stringable).
    """
    if guild_id in self.setting_cache:
        del self.setting_cache[guild_id]
        self.logger.debug(f"Invalidated cache for guild {guild_id}.")

classes.managers.MemberManager

MemberManager

Manage member profiles, hydration, and per-user settings resolution.

Responsibilities
  • Ensure a member document exists in MongoDB for (guild_id, member_id).
  • Fetch the Discord Member object.
  • Build a Member domain object with resolved user settings from modules.
Source code in classes/managers/MemberManager.py
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
class MemberManager:
    """Manage member profiles, hydration, and per-user settings resolution.

    Responsibilities:
        - Ensure a member document exists in MongoDB for (guild_id, member_id).
        - Fetch the Discord Member object.
        - Build a `Member` domain object with resolved user settings from modules.
    """
    def __init__(self, client: ExtendedClient, logger: Logger):
        self.client = client
        self.logger = logger

    async def fetch(self, member_id: str, guild_id: str) -> Member:
        """Fetch an existing member profile and hydrate a `Member` object.

        Args:
            member_id: Discord user id (stringable).
            guild_id: Discord guild id (stringable).

        Returns:
            A hydrated :class:`Member` object.

        Raises:
            CommandError: If the profile, guild, or member cannot be found.
        """

        member_id = str(member_id)
        guild_id = str(guild_id)
        if self.client.global_lock.is_locked():
            await self.client.global_lock.acquire()

        member_profile = await self.client.db.members.find_one({"id": member_id, "guildId": guild_id})
        if not member_profile:
            raise CommandError("No member profile!")

        guild_obj = await self.client.guild_manager.fetch_or_create(guild_id)
        if not guild_obj:
            raise CommandError("No guild found with the ID!")

        member = await guild_obj.guild.fetch_member(int(member_id))
        if not member:
            raise CommandError("No member!")

        settings = await get_all_settings(self.client, member_profile, guild_obj.guild, self.logger, member)
        return Member(self.client, member, guild_obj, settings, member_profile)

    async def fetch_or_create(self, member_id: str, guild_id: str) -> Member:
        """Fetch or create a member profile, then hydrate a `Member` object.

        Args:
            member_id: Discord user id (stringable).
            guild_id: Discord guild id (stringable).

        Returns:
            A hydrated :class:`Member` object.

        Raises:
            CommandError: If the guild/member cannot be found.
        """

        guild_id = str(guild_id)
        member_id = str(member_id)
        async with self.client.global_lock:
            self.logger.debug(f"Fetching member {member_id} from guild {guild_id}")

            member_profile = await self.client.db.members.find_one({"id": member_id, "guildId": guild_id})
            guild_obj = await self.client.guild_manager.fetch_or_create(guild_id)
            if not guild_obj:
                raise CommandError("No guild!")

            if not member_profile:
                member_profile = {"id": member_id, "guildId": guild_id}
                await self.client.db.members.insert_one(member_profile)

            member = await guild_obj.guild.fetch_member(int(member_id))
            if not member:
                await self.client.db.members.delete_one({"id": member_id, "guildId": guild_id})
                raise CommandError("No member!")

            settings = await get_all_settings(self.client, member_profile, guild_obj.guild, self.logger, member)
            return Member(self.client, member, guild_obj, settings, member_profile)

    async def delete(self, member_id: str, guild_id: str):
        """Delete a member profile document.

        Args:
            member_id: Discord user id (stringable).
            guild_id: Discord guild id (stringable).

        Returns:
            The deleted document or None.

        Raises:
            CommandError: If the profile does not exist.
        """
        member_id = str(member_id)
        guild_id = str(guild_id)
        async with self.client.global_lock:
            member_profile = await self.client.db.members.find_one_and_delete({"id": member_id, "guildId": guild_id})
            if not member_profile:
                raise CommandError("No member profile!")

            return member_profile

    async def create(self, member_id: str, guild_id: str) -> Member:
        """Create a new member profile and hydrate a `Member` object.

        Args:
            member_id: Discord user id (stringable).
            guild_id: Discord guild id (stringable).

        Returns:
            A hydrated :class:`Member` object.

        Raises:
            CommandError: If the guild/member cannot be found or creation fails.
        """
        member_id = str(member_id)
        guild_id = str(guild_id)
        async with self.client.global_lock:
            await self.client.global_lock.acquire()

            existing_member = await self.client.db.members.find_one({"id": member_id, "guildId": guild_id})
            if existing_member:
                raise CommandError("Member already exists!")

            member_profile = {"id": member_id, "guildId": guild_id}
            await self.client.db.members.insert_one(member_profile)

            guild_obj = await self.client.guild_manager.fetch_or_create(guild_id)
            if not guild_obj:
                raise CommandError("No guild!")

            member = await guild_obj.guild.fetch_member(int(member_id))
            if not member:
                await self.client.db.members.delete_one({"id": member_id, "guildId": guild_id})
                raise CommandError("No member!")

            settings = await get_all_settings(self.client, member_profile, guild_obj.guild, self.logger, member)
            return Member(self.client, member, guild_obj, settings, member_profile)

    async def find_or_create_profile(self, member_id: str, guild_id: str) -> Any:
        """Find a member profile document by (guild, user), creating it if missing.

        Args:
            member_id: Discord user id (stringable).
            guild_id: Discord guild id (stringable).

        Returns:
            The member profile document.
        """
        member_id = str(member_id)
        guild_id = str(guild_id)
        async with self.client.global_lock:

            member_profile = await self.client.db.members.find_one({"id": member_id, "guildId": guild_id})
            if not member_profile:
                member_profile = {"id": member_id, "guildId": guild_id}
                await self.client.db.members.insert_one(member_profile)

            return member_profile

    async def find_by_kv(self, filter: Dict[str, Any]) -> List[Member]:
        """Find member profiles by filter and hydrate `Member` objects.

        Args:
            filter: Mongo-style filter applied to the `members` collection.

        Returns:
            A list of hydrated :class:`Member` objects.

        Raises:
            CommandError: If no member profiles match the filter.
        """
        async with self.client.global_lock:
            member_profiles = await self.client.db.members.find(filter).to_list(length=None)
            if not member_profiles:
                raise CommandError("No member profiles!")

            guild_ids = list(set(profile["guildId"] for profile in member_profiles))
            guild_array = await get_guilds(self.client, guild_ids)

            member_array: List[Member] = []
            for guild_obj in guild_array:
                member_ids = [profile["id"] for profile in member_profiles if profile["guildId"] == str(guild_obj.guild.id)]
                members = await guild_obj.guild.fetch_members(user_ids=member_ids)
                for member in members:
                    member_profile = next((profile for profile in member_profiles if profile["id"] == str(member.id)), None)
                    settings = await get_all_settings(self.client, member_profile, guild_obj.guild, self.logger, member)
                    member_array.append(Member(self.client, member, guild_obj, settings, member_profile))

            return member_array

    async def update_member(self, member_id: str, guild_id: str, update: Dict[str, Any]) -> int:
        """Update fields in a member profile document via `$set`.

        Args:
            member_id: Discord user id.
            guild_id: Discord guild id.
            update: Field map to be applied under `$set`, e.g. {"settings.foo": 1}.

        Returns:
            The number of modified documents (0 or 1).
        """
        member_id, guild_id = str(member_id), str(guild_id)
        async with self.client.global_lock:
            return await self.client.db.update_one(
                "members",
                {"id": member_id, "guildId": guild_id},
                {"$set": update},
                upsert=False,
            )

create(member_id, guild_id) async

Create a new member profile and hydrate a Member object.

Parameters:

Name Type Description Default
member_id str

Discord user id (stringable).

required
guild_id str

Discord guild id (stringable).

required

Returns:

Type Description
Member

A hydrated :class:Member object.

Raises:

Type Description
CommandError

If the guild/member cannot be found or creation fails.

Source code in classes/managers/MemberManager.py
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
async def create(self, member_id: str, guild_id: str) -> Member:
    """Create a new member profile and hydrate a `Member` object.

    Args:
        member_id: Discord user id (stringable).
        guild_id: Discord guild id (stringable).

    Returns:
        A hydrated :class:`Member` object.

    Raises:
        CommandError: If the guild/member cannot be found or creation fails.
    """
    member_id = str(member_id)
    guild_id = str(guild_id)
    async with self.client.global_lock:
        await self.client.global_lock.acquire()

        existing_member = await self.client.db.members.find_one({"id": member_id, "guildId": guild_id})
        if existing_member:
            raise CommandError("Member already exists!")

        member_profile = {"id": member_id, "guildId": guild_id}
        await self.client.db.members.insert_one(member_profile)

        guild_obj = await self.client.guild_manager.fetch_or_create(guild_id)
        if not guild_obj:
            raise CommandError("No guild!")

        member = await guild_obj.guild.fetch_member(int(member_id))
        if not member:
            await self.client.db.members.delete_one({"id": member_id, "guildId": guild_id})
            raise CommandError("No member!")

        settings = await get_all_settings(self.client, member_profile, guild_obj.guild, self.logger, member)
        return Member(self.client, member, guild_obj, settings, member_profile)

delete(member_id, guild_id) async

Delete a member profile document.

Parameters:

Name Type Description Default
member_id str

Discord user id (stringable).

required
guild_id str

Discord guild id (stringable).

required

Returns:

Type Description

The deleted document or None.

Raises:

Type Description
CommandError

If the profile does not exist.

Source code in classes/managers/MemberManager.py
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
async def delete(self, member_id: str, guild_id: str):
    """Delete a member profile document.

    Args:
        member_id: Discord user id (stringable).
        guild_id: Discord guild id (stringable).

    Returns:
        The deleted document or None.

    Raises:
        CommandError: If the profile does not exist.
    """
    member_id = str(member_id)
    guild_id = str(guild_id)
    async with self.client.global_lock:
        member_profile = await self.client.db.members.find_one_and_delete({"id": member_id, "guildId": guild_id})
        if not member_profile:
            raise CommandError("No member profile!")

        return member_profile

fetch(member_id, guild_id) async

Fetch an existing member profile and hydrate a Member object.

Parameters:

Name Type Description Default
member_id str

Discord user id (stringable).

required
guild_id str

Discord guild id (stringable).

required

Returns:

Type Description
Member

A hydrated :class:Member object.

Raises:

Type Description
CommandError

If the profile, guild, or member cannot be found.

Source code in classes/managers/MemberManager.py
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
async def fetch(self, member_id: str, guild_id: str) -> Member:
    """Fetch an existing member profile and hydrate a `Member` object.

    Args:
        member_id: Discord user id (stringable).
        guild_id: Discord guild id (stringable).

    Returns:
        A hydrated :class:`Member` object.

    Raises:
        CommandError: If the profile, guild, or member cannot be found.
    """

    member_id = str(member_id)
    guild_id = str(guild_id)
    if self.client.global_lock.is_locked():
        await self.client.global_lock.acquire()

    member_profile = await self.client.db.members.find_one({"id": member_id, "guildId": guild_id})
    if not member_profile:
        raise CommandError("No member profile!")

    guild_obj = await self.client.guild_manager.fetch_or_create(guild_id)
    if not guild_obj:
        raise CommandError("No guild found with the ID!")

    member = await guild_obj.guild.fetch_member(int(member_id))
    if not member:
        raise CommandError("No member!")

    settings = await get_all_settings(self.client, member_profile, guild_obj.guild, self.logger, member)
    return Member(self.client, member, guild_obj, settings, member_profile)

fetch_or_create(member_id, guild_id) async

Fetch or create a member profile, then hydrate a Member object.

Parameters:

Name Type Description Default
member_id str

Discord user id (stringable).

required
guild_id str

Discord guild id (stringable).

required

Returns:

Type Description
Member

A hydrated :class:Member object.

Raises:

Type Description
CommandError

If the guild/member cannot be found.

Source code in classes/managers/MemberManager.py
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
async def fetch_or_create(self, member_id: str, guild_id: str) -> Member:
    """Fetch or create a member profile, then hydrate a `Member` object.

    Args:
        member_id: Discord user id (stringable).
        guild_id: Discord guild id (stringable).

    Returns:
        A hydrated :class:`Member` object.

    Raises:
        CommandError: If the guild/member cannot be found.
    """

    guild_id = str(guild_id)
    member_id = str(member_id)
    async with self.client.global_lock:
        self.logger.debug(f"Fetching member {member_id} from guild {guild_id}")

        member_profile = await self.client.db.members.find_one({"id": member_id, "guildId": guild_id})
        guild_obj = await self.client.guild_manager.fetch_or_create(guild_id)
        if not guild_obj:
            raise CommandError("No guild!")

        if not member_profile:
            member_profile = {"id": member_id, "guildId": guild_id}
            await self.client.db.members.insert_one(member_profile)

        member = await guild_obj.guild.fetch_member(int(member_id))
        if not member:
            await self.client.db.members.delete_one({"id": member_id, "guildId": guild_id})
            raise CommandError("No member!")

        settings = await get_all_settings(self.client, member_profile, guild_obj.guild, self.logger, member)
        return Member(self.client, member, guild_obj, settings, member_profile)

find_by_kv(filter) async

Find member profiles by filter and hydrate Member objects.

Parameters:

Name Type Description Default
filter Dict[str, Any]

Mongo-style filter applied to the members collection.

required

Returns:

Type Description
List[Member]

A list of hydrated :class:Member objects.

Raises:

Type Description
CommandError

If no member profiles match the filter.

Source code in classes/managers/MemberManager.py
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
async def find_by_kv(self, filter: Dict[str, Any]) -> List[Member]:
    """Find member profiles by filter and hydrate `Member` objects.

    Args:
        filter: Mongo-style filter applied to the `members` collection.

    Returns:
        A list of hydrated :class:`Member` objects.

    Raises:
        CommandError: If no member profiles match the filter.
    """
    async with self.client.global_lock:
        member_profiles = await self.client.db.members.find(filter).to_list(length=None)
        if not member_profiles:
            raise CommandError("No member profiles!")

        guild_ids = list(set(profile["guildId"] for profile in member_profiles))
        guild_array = await get_guilds(self.client, guild_ids)

        member_array: List[Member] = []
        for guild_obj in guild_array:
            member_ids = [profile["id"] for profile in member_profiles if profile["guildId"] == str(guild_obj.guild.id)]
            members = await guild_obj.guild.fetch_members(user_ids=member_ids)
            for member in members:
                member_profile = next((profile for profile in member_profiles if profile["id"] == str(member.id)), None)
                settings = await get_all_settings(self.client, member_profile, guild_obj.guild, self.logger, member)
                member_array.append(Member(self.client, member, guild_obj, settings, member_profile))

        return member_array

find_or_create_profile(member_id, guild_id) async

Find a member profile document by (guild, user), creating it if missing.

Parameters:

Name Type Description Default
member_id str

Discord user id (stringable).

required
guild_id str

Discord guild id (stringable).

required

Returns:

Type Description
Any

The member profile document.

Source code in classes/managers/MemberManager.py
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
async def find_or_create_profile(self, member_id: str, guild_id: str) -> Any:
    """Find a member profile document by (guild, user), creating it if missing.

    Args:
        member_id: Discord user id (stringable).
        guild_id: Discord guild id (stringable).

    Returns:
        The member profile document.
    """
    member_id = str(member_id)
    guild_id = str(guild_id)
    async with self.client.global_lock:

        member_profile = await self.client.db.members.find_one({"id": member_id, "guildId": guild_id})
        if not member_profile:
            member_profile = {"id": member_id, "guildId": guild_id}
            await self.client.db.members.insert_one(member_profile)

        return member_profile

update_member(member_id, guild_id, update) async

Update fields in a member profile document via $set.

Parameters:

Name Type Description Default
member_id str

Discord user id.

required
guild_id str

Discord guild id.

required
update Dict[str, Any]

Field map to be applied under $set, e.g. {"settings.foo": 1}.

required

Returns:

Type Description
int

The number of modified documents (0 or 1).

Source code in classes/managers/MemberManager.py
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
async def update_member(self, member_id: str, guild_id: str, update: Dict[str, Any]) -> int:
    """Update fields in a member profile document via `$set`.

    Args:
        member_id: Discord user id.
        guild_id: Discord guild id.
        update: Field map to be applied under `$set`, e.g. {"settings.foo": 1}.

    Returns:
        The number of modified documents (0 or 1).
    """
    member_id, guild_id = str(member_id), str(guild_id)
    async with self.client.global_lock:
        return await self.client.db.update_one(
            "members",
            {"id": member_id, "guildId": guild_id},
            {"$set": update},
            upsert=False,
        )

get_all_settings(client, member_data, guild, logger, user) async

Resolve all user-scoped settings defined by modules for a specific member.

Copies the Setting templates from modules, preserves hooks (save, load, parse, parse_to_database, condition), then loads either from load(), from the member profile data, or uses the default.

Parameters:

Name Type Description Default
client ExtendedClient

Extended client.

required
member_data Any

Member profile document from DB.

required
guild Guild

Discord guild object.

required
logger Logger

Logger instance.

required
user Member

Discord member to resolve settings for.

required

Returns:

Type Description

A mapping from setting id to Setting instances with value resolved.

Source code in classes/managers/MemberManager.py
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
async def get_all_settings(client: ExtendedClient, member_data: Any, guild: DiscordGuild, logger: Logger, user: GuildMember):
    """Resolve all user-scoped settings defined by modules for a specific member.

    Copies the `Setting` templates from modules, preserves hooks (`save`, `load`,
    `parse`, `parse_to_database`, `condition`), then loads either from `load()`,
    from the member profile data, or uses the default.

    Args:
        client: Extended client.
        member_data: Member profile document from DB.
        guild: Discord guild object.
        logger: Logger instance.
        user: Discord member to resolve settings for.

    Returns:
        A mapping from setting id to `Setting` instances with `value` resolved.
    """
    settings = [setting for module in client.modules.values() for setting in module.user_settings]
    settings_map: Dict[str, Setting[Any]] = {}

    for original_setting in settings:
        setting = original_setting.clone()
        setting.save = original_setting.save
        setting.load = original_setting.load
        setting.parse = original_setting.parse
        setting.parse_to_database = original_setting.parse_to_database
        setting.condition = original_setting.condition

        setting_data = member_data.settings.get(setting.id)

        if setting.load:
            setting.value = await setting.load(guild, member_data, user)
        elif setting_data:
            if setting.parse:
                setting.value = await setting.parse(setting_data, client, member_data, guild, user)
            else:
                setting.value = setting_data
        settings_map[setting.id] = setting

    return settings_map

get_guilds(client, guilds) async

Helper to materialize a list of guild wrappers by id.

Parameters:

Name Type Description Default
client ExtendedClient

Extended client.

required
guilds List[str]

List of guild ids.

required

Returns:

Type Description
List[Guild]

A list of :class:Guild wrappers.

Source code in classes/managers/MemberManager.py
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
async def get_guilds(client: ExtendedClient, guilds: List[str]) -> List[Guild]:
    """Helper to materialize a list of guild wrappers by id.

    Args:
        client: Extended client.
        guilds: List of guild ids.

    Returns:
        A list of :class:`Guild` wrappers.
    """
    guild_array: List[Guild] = []
    for guild_id in guilds:
        if any(guild_obj.guild.id == int(guild_id) for guild_obj in guild_array):
            continue
        guild_array.append(await client.guild_manager.fetch_or_create(guild_id))
    return guild_array

classes.managers.SettingsManager

SettingsManager

Facade for loading and saving guild- and member-level settings.

Delegates to GuildManager and MemberManager to materialize domain objects and persist updates.

Source code in classes/managers/SettingsManager.py
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
class SettingsManager:
    """Facade for loading and saving guild- and member-level settings.

    Delegates to `GuildManager` and `MemberManager` to materialize domain objects
    and persist updates.
    """

    def __init__(self, client: ExtendedClient, logger: Optional[logging.Logger] = None):
        self.client = client
        self.logger = logger or logging.getLogger("SettingsManager")
        self.guild_manager = client.guild_manager  # Assuming GuildManager is attached to client
        self.member_manager = client.member_manager    # Assuming memberManager is attached to client

    async def load_guild_settings(self, guild_id: int) -> Optional[Guild]:
        """Load and return a `Guild` object with settings materialized.

        Args:
            guild_id: Discord guild id.

        Returns:
            A `Guild` object or None if loading fails.
        """
        self.logger.debug(f"Loading settings for guild {guild_id}...")
        guild = await self.guild_manager.fetch_or_create(guild_id)
        if guild:
            self.logger.debug(f"Settings loaded for guild {guild_id}.")
            return guild
        else:
            self.logger.error(f"Failed to load settings for guild {guild_id}.")
            return None

    async def save_guild_setting(self, guild_id: int, setting_id: str, value: Any):
        """Persist a single guild setting.

        Args:
            guild_id: Discord guild id.
            setting_id: Setting identifier.
            value: New value to store.

        Raises:
            ValueError: If the guild cannot be loaded.
            KeyError: If the setting id is unknown for that guild.
        """
        guild = await self.guild_manager.fetch_or_create(guild_id)
        if not guild:
            raise ValueError(f"Guild with ID {guild_id} not found.")

        try:
            await guild.update_setting(setting_id, value)
            self.logger.debug(f"Saved setting '{setting_id}' for guild {guild_id} with value '{value}'.")
        except KeyError as e:
            self.logger.error(str(e))
            raise

    async def load_member_settings(self, member_id: int, guild_id: int) -> Optional[Member]:
        """Load and return a hydrated `Member` object with user settings.

        This delegates to `MemberManager.fetch_or_create`, which ensures the
        profile exists and resolves per‑user settings from module definitions.

        Args:
            member_id: Discord user id.
            guild_id: Discord guild id.

        Returns:
            Hydrated `Member` or `None` if loading fails.
        """
        self.logger.debug(f"Loading settings for member {member_id} in guild {guild_id}...")
        try:
            # This already returns a hydrated Member object with settings:
            member = await self.member_manager.fetch_or_create(str(member_id), str(guild_id))
            self.logger.debug(f"Settings loaded for member {member_id} in guild {guild_id}.")
            return member
        except Exception as e:
            self.logger.error(f"Failed to load settings for member {member_id} in guild {guild_id}: {e}")
            return None

    async def save_member_setting(self, member_id: int, guild_id: int, setting_id: str, value: Any):
        """Persist a single member (user-scoped) setting.

        Args:
            member_id: Discord user id.
            guild_id: Discord guild id.
            setting_id: Setting identifier.
            value: New value to store.

        Raises:
            ValueError: If the member cannot be loaded.
            Exception: If persistence fails.
        """
        member = await self.member_manager.fetch_or_create(member_id, guild_id)
        if not member:
            raise ValueError(f"member with ID {member_id} in guild {guild_id} not found.")

        try:
            await self.member_manager.update_member(member_id, guild_id, {f"settings.{setting_id}": value})
            self.logger.debug(f"Saved setting '{setting_id}' for member {member_id} in guild {guild_id} with value '{value}'.")
        except Exception as e:
            self.logger.error(f"Error saving setting '{setting_id}' for member {member_id} in guild {guild_id}: {e}")
            raise

load_guild_settings(guild_id) async

Load and return a Guild object with settings materialized.

Parameters:

Name Type Description Default
guild_id int

Discord guild id.

required

Returns:

Type Description
Optional[Guild]

A Guild object or None if loading fails.

Source code in classes/managers/SettingsManager.py
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
async def load_guild_settings(self, guild_id: int) -> Optional[Guild]:
    """Load and return a `Guild` object with settings materialized.

    Args:
        guild_id: Discord guild id.

    Returns:
        A `Guild` object or None if loading fails.
    """
    self.logger.debug(f"Loading settings for guild {guild_id}...")
    guild = await self.guild_manager.fetch_or_create(guild_id)
    if guild:
        self.logger.debug(f"Settings loaded for guild {guild_id}.")
        return guild
    else:
        self.logger.error(f"Failed to load settings for guild {guild_id}.")
        return None

load_member_settings(member_id, guild_id) async

Load and return a hydrated Member object with user settings.

This delegates to MemberManager.fetch_or_create, which ensures the profile exists and resolves per‑user settings from module definitions.

Parameters:

Name Type Description Default
member_id int

Discord user id.

required
guild_id int

Discord guild id.

required

Returns:

Type Description
Optional[Member]

Hydrated Member or None if loading fails.

Source code in classes/managers/SettingsManager.py
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
async def load_member_settings(self, member_id: int, guild_id: int) -> Optional[Member]:
    """Load and return a hydrated `Member` object with user settings.

    This delegates to `MemberManager.fetch_or_create`, which ensures the
    profile exists and resolves per‑user settings from module definitions.

    Args:
        member_id: Discord user id.
        guild_id: Discord guild id.

    Returns:
        Hydrated `Member` or `None` if loading fails.
    """
    self.logger.debug(f"Loading settings for member {member_id} in guild {guild_id}...")
    try:
        # This already returns a hydrated Member object with settings:
        member = await self.member_manager.fetch_or_create(str(member_id), str(guild_id))
        self.logger.debug(f"Settings loaded for member {member_id} in guild {guild_id}.")
        return member
    except Exception as e:
        self.logger.error(f"Failed to load settings for member {member_id} in guild {guild_id}: {e}")
        return None

save_guild_setting(guild_id, setting_id, value) async

Persist a single guild setting.

Parameters:

Name Type Description Default
guild_id int

Discord guild id.

required
setting_id str

Setting identifier.

required
value Any

New value to store.

required

Raises:

Type Description
ValueError

If the guild cannot be loaded.

KeyError

If the setting id is unknown for that guild.

Source code in classes/managers/SettingsManager.py
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
async def save_guild_setting(self, guild_id: int, setting_id: str, value: Any):
    """Persist a single guild setting.

    Args:
        guild_id: Discord guild id.
        setting_id: Setting identifier.
        value: New value to store.

    Raises:
        ValueError: If the guild cannot be loaded.
        KeyError: If the setting id is unknown for that guild.
    """
    guild = await self.guild_manager.fetch_or_create(guild_id)
    if not guild:
        raise ValueError(f"Guild with ID {guild_id} not found.")

    try:
        await guild.update_setting(setting_id, value)
        self.logger.debug(f"Saved setting '{setting_id}' for guild {guild_id} with value '{value}'.")
    except KeyError as e:
        self.logger.error(str(e))
        raise

save_member_setting(member_id, guild_id, setting_id, value) async

Persist a single member (user-scoped) setting.

Parameters:

Name Type Description Default
member_id int

Discord user id.

required
guild_id int

Discord guild id.

required
setting_id str

Setting identifier.

required
value Any

New value to store.

required

Raises:

Type Description
ValueError

If the member cannot be loaded.

Exception

If persistence fails.

Source code in classes/managers/SettingsManager.py
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
async def save_member_setting(self, member_id: int, guild_id: int, setting_id: str, value: Any):
    """Persist a single member (user-scoped) setting.

    Args:
        member_id: Discord user id.
        guild_id: Discord guild id.
        setting_id: Setting identifier.
        value: New value to store.

    Raises:
        ValueError: If the member cannot be loaded.
        Exception: If persistence fails.
    """
    member = await self.member_manager.fetch_or_create(member_id, guild_id)
    if not member:
        raise ValueError(f"member with ID {member_id} in guild {guild_id} not found.")

    try:
        await self.member_manager.update_member(member_id, guild_id, {f"settings.{setting_id}": value})
        self.logger.debug(f"Saved setting '{setting_id}' for member {member_id} in guild {guild_id} with value '{value}'.")
    except Exception as e:
        self.logger.error(f"Error saving setting '{setting_id}' for member {member_id} in guild {guild_id}: {e}")
        raise

classes.managers.PermissionsManager

PermissionsManager

Hierarchical permission registry and resolver.

Stores permission nodes in a nested mapping keyed by dot‑separated paths, for example "Role.*", "User.1234", "Channel.5678", or custom trees like "Feature.Admin.Purge".

Resolution rules
  1. Exact path match returns the first terminal node found.
  2. If a segment is missing, the resolver remembers a sibling literal "*" at that level, which acts as a wildcard. The last wildcard seen becomes the fallback.
  3. If the traversal ends on a terminal node, it is returned. Otherwise the last wildcard wins.

The manager also evaluates override documents with "allow" and "deny" lists by calling into registered nodes.

Parameters:

Name Type Description Default
client Bot

Discord bot client.

required
logger Logger

Logger instance.

required
Source code in classes/managers/PermissionsManager.py
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
class PermissionsManager:
    """Hierarchical permission registry and resolver.

    Stores permission nodes in a nested mapping keyed by dot‑separated paths, for example
    "Role.*", "User.1234", "Channel.5678", or custom trees like "Feature.Admin.Purge".

    Resolution rules:
      1. Exact path match returns the first terminal node found.
      2. If a segment is missing, the resolver remembers a sibling literal "*" at that level,
         which acts as a wildcard. The last wildcard seen becomes the fallback.
      3. If the traversal ends on a terminal node, it is returned. Otherwise the last wildcard wins.

    The manager also evaluates override documents with "allow" and "deny" lists by calling into
    registered nodes.

    Args:
        client: Discord bot client.
        logger: Logger instance.
    """
    def __init__(self, client: Bot, logger: logging.Logger):
        self.client = client
        self.logger = logger
        self.permissions: RecursiveMap = {}

    def register_node(self, permission: str, result: PermissionNode):
        """Register a terminal permission node.

        The permission string is a dot‑separated path. Intermediate namespaces are created as dicts.
        The final segment becomes the terminal node and must be a callable that returns an awaitable bool.

        Example:
            register_node("Role.*", role_checker)
            register_node("Feature.Admin.Purge", purge_checker)

        Args:
            permission: Permission path, for example "Role.*" or "Feature.Admin.Purge".
            result: Async callable of type PermissionNode.
        """
        namespaces = permission.split('.')
        current = self.permissions
        last = namespaces.pop() if namespaces else None

        if not last:
            self.logger.warning("No namespaces provided for permission registration.")
            return

        for namespace in namespaces:
            if namespace not in current:
                current[namespace] = {}
            elif not isinstance(current[namespace], dict):
                self.logger.error(f"Cannot create namespace '{namespace}' as it's already an end node.")
                return
            current = current[namespace]

        current[last] = result

    def get_node(self, permission: str) -> Optional[PermissionNode]:
        """Resolve a permission path into a terminal node.

        Traverses the internal tree segment by segment and returns the matching callable.
        Supports a literal "*" at any level as a wildcard fallback.

        Args:
            permission: Permission path to resolve.

        Returns:
            The resolved PermissionNode or None if nothing matches.
        """
        namespaces = permission.split('.')
        current = self.permissions
        last_global: Optional[PermissionNode] = None

        for namespace in namespaces:
            if namespace in current:
                node = current[namespace]
                if is_end_node(node):
                    return node
                current = node
            elif '*' in current:
                last_global = current['*']
            else:
                return last_global

        return current if is_end_node(current) else last_global

    async def check_permission_for(self, node: str, member: GuildMember, channel: TextChannel) -> bool:
        """Evaluate a permission node for a given member and channel.

        Args:
            node: Permission path to evaluate.
            member: Guild member being checked.
            channel: Text channel context for the check.

        Returns:
            True if the node grants permission, False otherwise.
        """
        permission_node = self.get_node(node)
        if not permission_node:
            self.logger.warning(f"Permission node '{node}' not found.")
            return False

        try:
            result = await permission_node(self.client, node, member, channel)
            return result
        except Exception as e:
            self.logger.error(f"Error executing permission node '{node}': {e}")
            return False

    async def compute_permissions(self, override: OverrideNode, member: GuildMember, channel: TextChannel) -> Optional[bool]:
        """Evaluate an override document with allow and deny lists.

        The override contains two lists of permission paths:
            - "allow": if any path evaluates to True, returns True
            - "deny": if any path evaluates to True, returns False

        Args:
            override: Override document with "allow" and "deny".
            member: Guild member being checked.
            channel: Text channel context.

        Returns:
            True, False, or None if no rule matched.
        """
        for allow_perm in override.get('allow', []):
            if await self.check_permission_for(allow_perm, member, channel):
                return True

        for deny_perm in override.get('deny', []):
            if await self.check_permission_for(deny_perm, member, channel):
                return False

        return None

    async def has_permission(self, node: str, member: GuildMember, channel: TextChannel, override: OverrideNode) -> bool:
        """High‑level permission check that combines Discord flags and overrides.

        Order:
            1) If the last segment of `node` matches a Discord permission flag and the member has it,
               return True.
            2) Evaluate the override document. If it resolves, return that result.
            3) Default to False.

        Note:
            This method is intended for cases where `node` maps to a Discord built‑in permission,
            for example "Permissions.manage_messages". For custom namespaces like "Role.*", prefer
            `compute_permissions` or `check_permission_for` with explicit nodes.

        Args:
            node: Permission path to check.
            member: Guild member being checked.
            channel: Text channel context.
            override: Override document with "allow" and "deny".

        Returns:
            True if permission is granted, otherwise False.
        """
        discord_permission = getattr(member.guild_permissions, node.split('.')[-1], False)
        if discord_permission:
            return True

        override_result = await self.compute_permissions(override, member, channel)
        if override_result is not None:
            return override_result

        return False

check_permission_for(node, member, channel) async

Evaluate a permission node for a given member and channel.

Parameters:

Name Type Description Default
node str

Permission path to evaluate.

required
member Member

Guild member being checked.

required
channel TextChannel

Text channel context for the check.

required

Returns:

Type Description
bool

True if the node grants permission, False otherwise.

Source code in classes/managers/PermissionsManager.py
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
async def check_permission_for(self, node: str, member: GuildMember, channel: TextChannel) -> bool:
    """Evaluate a permission node for a given member and channel.

    Args:
        node: Permission path to evaluate.
        member: Guild member being checked.
        channel: Text channel context for the check.

    Returns:
        True if the node grants permission, False otherwise.
    """
    permission_node = self.get_node(node)
    if not permission_node:
        self.logger.warning(f"Permission node '{node}' not found.")
        return False

    try:
        result = await permission_node(self.client, node, member, channel)
        return result
    except Exception as e:
        self.logger.error(f"Error executing permission node '{node}': {e}")
        return False

compute_permissions(override, member, channel) async

Evaluate an override document with allow and deny lists.

The override contains two lists of permission paths
  • "allow": if any path evaluates to True, returns True
  • "deny": if any path evaluates to True, returns False

Parameters:

Name Type Description Default
override OverrideNode

Override document with "allow" and "deny".

required
member Member

Guild member being checked.

required
channel TextChannel

Text channel context.

required

Returns:

Type Description
Optional[bool]

True, False, or None if no rule matched.

Source code in classes/managers/PermissionsManager.py
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
async def compute_permissions(self, override: OverrideNode, member: GuildMember, channel: TextChannel) -> Optional[bool]:
    """Evaluate an override document with allow and deny lists.

    The override contains two lists of permission paths:
        - "allow": if any path evaluates to True, returns True
        - "deny": if any path evaluates to True, returns False

    Args:
        override: Override document with "allow" and "deny".
        member: Guild member being checked.
        channel: Text channel context.

    Returns:
        True, False, or None if no rule matched.
    """
    for allow_perm in override.get('allow', []):
        if await self.check_permission_for(allow_perm, member, channel):
            return True

    for deny_perm in override.get('deny', []):
        if await self.check_permission_for(deny_perm, member, channel):
            return False

    return None

get_node(permission)

Resolve a permission path into a terminal node.

Traverses the internal tree segment by segment and returns the matching callable. Supports a literal "*" at any level as a wildcard fallback.

Parameters:

Name Type Description Default
permission str

Permission path to resolve.

required

Returns:

Type Description
Optional[PermissionNode]

The resolved PermissionNode or None if nothing matches.

Source code in classes/managers/PermissionsManager.py
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def get_node(self, permission: str) -> Optional[PermissionNode]:
    """Resolve a permission path into a terminal node.

    Traverses the internal tree segment by segment and returns the matching callable.
    Supports a literal "*" at any level as a wildcard fallback.

    Args:
        permission: Permission path to resolve.

    Returns:
        The resolved PermissionNode or None if nothing matches.
    """
    namespaces = permission.split('.')
    current = self.permissions
    last_global: Optional[PermissionNode] = None

    for namespace in namespaces:
        if namespace in current:
            node = current[namespace]
            if is_end_node(node):
                return node
            current = node
        elif '*' in current:
            last_global = current['*']
        else:
            return last_global

    return current if is_end_node(current) else last_global

has_permission(node, member, channel, override) async

High‑level permission check that combines Discord flags and overrides.

Order

1) If the last segment of node matches a Discord permission flag and the member has it, return True. 2) Evaluate the override document. If it resolves, return that result. 3) Default to False.

Note

This method is intended for cases where node maps to a Discord built‑in permission, for example "Permissions.manage_messages". For custom namespaces like "Role.*", prefer compute_permissions or check_permission_for with explicit nodes.

Parameters:

Name Type Description Default
node str

Permission path to check.

required
member Member

Guild member being checked.

required
channel TextChannel

Text channel context.

required
override OverrideNode

Override document with "allow" and "deny".

required

Returns:

Type Description
bool

True if permission is granted, otherwise False.

Source code in classes/managers/PermissionsManager.py
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
async def has_permission(self, node: str, member: GuildMember, channel: TextChannel, override: OverrideNode) -> bool:
    """High‑level permission check that combines Discord flags and overrides.

    Order:
        1) If the last segment of `node` matches a Discord permission flag and the member has it,
           return True.
        2) Evaluate the override document. If it resolves, return that result.
        3) Default to False.

    Note:
        This method is intended for cases where `node` maps to a Discord built‑in permission,
        for example "Permissions.manage_messages". For custom namespaces like "Role.*", prefer
        `compute_permissions` or `check_permission_for` with explicit nodes.

    Args:
        node: Permission path to check.
        member: Guild member being checked.
        channel: Text channel context.
        override: Override document with "allow" and "deny".

    Returns:
        True if permission is granted, otherwise False.
    """
    discord_permission = getattr(member.guild_permissions, node.split('.')[-1], False)
    if discord_permission:
        return True

    override_result = await self.compute_permissions(override, member, channel)
    if override_result is not None:
        return override_result

    return False

register_node(permission, result)

Register a terminal permission node.

The permission string is a dot‑separated path. Intermediate namespaces are created as dicts. The final segment becomes the terminal node and must be a callable that returns an awaitable bool.

Example

register_node("Role.*", role_checker) register_node("Feature.Admin.Purge", purge_checker)

Parameters:

Name Type Description Default
permission str

Permission path, for example "Role.*" or "Feature.Admin.Purge".

required
result PermissionNode

Async callable of type PermissionNode.

required
Source code in classes/managers/PermissionsManager.py
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def register_node(self, permission: str, result: PermissionNode):
    """Register a terminal permission node.

    The permission string is a dot‑separated path. Intermediate namespaces are created as dicts.
    The final segment becomes the terminal node and must be a callable that returns an awaitable bool.

    Example:
        register_node("Role.*", role_checker)
        register_node("Feature.Admin.Purge", purge_checker)

    Args:
        permission: Permission path, for example "Role.*" or "Feature.Admin.Purge".
        result: Async callable of type PermissionNode.
    """
    namespaces = permission.split('.')
    current = self.permissions
    last = namespaces.pop() if namespaces else None

    if not last:
        self.logger.warning("No namespaces provided for permission registration.")
        return

    for namespace in namespaces:
        if namespace not in current:
            current[namespace] = {}
        elif not isinstance(current[namespace], dict):
            self.logger.error(f"Cannot create namespace '{namespace}' as it's already an end node.")
            return
        current = current[namespace]

    current[last] = result

is_end_node(node)

Return True if the node is a terminal permission node.

A terminal node is a callable that resolves an access decision and is not a dict.

Source code in classes/managers/PermissionsManager.py
11
12
13
14
15
16
def is_end_node(node: Union[PermissionNode, RecursiveMap]) -> bool:
    """Return True if the node is a terminal permission node.

    A terminal node is a callable that resolves an access decision and is not a dict.
    """
    return not isinstance(node, dict)

classes.managers.FlagsManager

FlagsManager

Central registry for feature flags and defaults.

This manager holds global default flag values and exposes helpers to read per-guild flag values via the guild's own ObjectFlags facade.

Typical usage
  • Call :meth:register_flag at startup to declare a flag and its default.
  • Use :meth:get_flag inside your features to read the value for a given guild, falling back to the global default when the guild did not override it.
  • Use :meth:has_flag to quickly check if a flag exists either locally (guild) or globally.
Notes

Guild-level reads are delegated to :class:classes.structs.ObjectFlags.ObjectFlags via guild.flags (which is created in :class:classes.structs.Guild.Guild).

Source code in classes/managers/FlagsManager.py
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
class FlagsManager:
    """
    Central registry for feature flags and defaults.

    This manager holds **global default flag values** and exposes helpers to read
    per-guild flag values via the guild's own `ObjectFlags` facade.

    Typical usage:
        - Call :meth:`register_flag` at startup to declare a flag and its default.
        - Use :meth:`get_flag` inside your features to read the value for a given guild,
          falling back to the global default when the guild did not override it.
        - Use :meth:`has_flag` to quickly check if a flag exists either locally (guild)
          or globally.

    Notes
    -----
    Guild-level reads are delegated to :class:`classes.structs.ObjectFlags.ObjectFlags`
    via ``guild.flags`` (which is created in :class:`classes.structs.Guild.Guild`).
    """

    def __init__(self, client: Bot, logger: logging.Logger):
        """
        Parameters
        ----------
        client:
            The Discord client/bot instance.
        logger:
            Logger used for diagnostics.
        """
        self.client = client
        self.flags = {}  # Stores global default flags
        self.logger = logger

    def register_flag(self, flag: str, default_value: Union[str, bool, list]):
        """
        Register (or overwrite) a global flag and its default value.

        Parameters
        ----------
        flag:
            Unique flag identifier (e.g., ``"features.xp_enabled"``).
        default_value:
            Default value used when the guild has no explicit override.

        Returns
        -------
        FlagsManager
            Self, to allow call chaining.

        Examples
        --------
        >>> flags.register_flag("features.xp_enabled", True)
        """
        if flag in self.flags:
            self.logger.warning(f"Flag {flag} already exists, overwriting it.")
        self.flags[flag] = default_value
        return self

    def delete_flag(self, flag: str):
        """
        Remove a global flag definition.

        If the flag is not present, this is a no-op.

        Parameters
        ----------
        flag:
            The flag identifier to remove.

        Returns
        -------
        FlagsManager
            Self, to allow call chaining.
        """
        if flag in self.flags:
            del self.flags[flag]
        return self

    def get_flag(self, guild: Union[Guild, Any], flag: str) -> Any:
        """
        Resolve a flag value for a specific guild.

        The lookup order is:
        1) Guild override via ``guild.flags.get(flag)``.
        2) Global default from this manager.

        Parameters
        ----------
        guild:
            The target :class:`classes.structs.Guild.Guild` instance.
        flag:
            The flag identifier.

        Returns
        -------
        Any
            The resolved flag value (guild override if present, otherwise global default).

        Raises
        ------
        TypeError
            If ``guild`` is not an instance of :class:`classes.structs.Guild.Guild`.
        """
        if not isinstance(guild, Guild):
            self.logger.error(f"Expected a Guild instance, got {type(guild).__name__}.")
            raise TypeError("Invalid guild type.")
        return guild.flags.get(flag) or self.flags.get(flag)

    def has_flag(self, guild: Union[Guild, Any], flag: str) -> bool:
        """
        Check if a flag exists for a guild (override) or globally.

        Parameters
        ----------
        guild:
            The target :class:`classes.structs.Guild.Guild` instance.
        flag:
            The flag identifier.

        Returns
        -------
        bool
            ``True`` if the flag is available either in the guild override or in
            the manager's global defaults; ``False`` otherwise.

        Raises
        ------
        TypeError
            If ``guild`` is not an instance of :class:`classes.structs.Guild.Guild`.
        """
        if not isinstance(guild, Guild):
            self.logger.error(f"Expected a Guild instance, got {type(guild).__name__}.")
            raise TypeError("Invalid guild type.")
        return guild.flags.has(flag) or flag in self.flags

__init__(client, logger)

Parameters

client: The Discord client/bot instance. logger: Logger used for diagnostics.

Source code in classes/managers/FlagsManager.py
28
29
30
31
32
33
34
35
36
37
38
39
def __init__(self, client: Bot, logger: logging.Logger):
    """
    Parameters
    ----------
    client:
        The Discord client/bot instance.
    logger:
        Logger used for diagnostics.
    """
    self.client = client
    self.flags = {}  # Stores global default flags
    self.logger = logger

delete_flag(flag)

Remove a global flag definition.

If the flag is not present, this is a no-op.

Parameters

flag: The flag identifier to remove.

Returns

FlagsManager Self, to allow call chaining.

Source code in classes/managers/FlagsManager.py
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def delete_flag(self, flag: str):
    """
    Remove a global flag definition.

    If the flag is not present, this is a no-op.

    Parameters
    ----------
    flag:
        The flag identifier to remove.

    Returns
    -------
    FlagsManager
        Self, to allow call chaining.
    """
    if flag in self.flags:
        del self.flags[flag]
    return self

get_flag(guild, flag)

Resolve a flag value for a specific guild.

The lookup order is: 1) Guild override via guild.flags.get(flag). 2) Global default from this manager.

Parameters

guild: The target :class:classes.structs.Guild.Guild instance. flag: The flag identifier.

Returns

Any The resolved flag value (guild override if present, otherwise global default).

Raises

TypeError If guild is not an instance of :class:classes.structs.Guild.Guild.

Source code in classes/managers/FlagsManager.py
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def get_flag(self, guild: Union[Guild, Any], flag: str) -> Any:
    """
    Resolve a flag value for a specific guild.

    The lookup order is:
    1) Guild override via ``guild.flags.get(flag)``.
    2) Global default from this manager.

    Parameters
    ----------
    guild:
        The target :class:`classes.structs.Guild.Guild` instance.
    flag:
        The flag identifier.

    Returns
    -------
    Any
        The resolved flag value (guild override if present, otherwise global default).

    Raises
    ------
    TypeError
        If ``guild`` is not an instance of :class:`classes.structs.Guild.Guild`.
    """
    if not isinstance(guild, Guild):
        self.logger.error(f"Expected a Guild instance, got {type(guild).__name__}.")
        raise TypeError("Invalid guild type.")
    return guild.flags.get(flag) or self.flags.get(flag)

has_flag(guild, flag)

Check if a flag exists for a guild (override) or globally.

Parameters

guild: The target :class:classes.structs.Guild.Guild instance. flag: The flag identifier.

Returns

bool True if the flag is available either in the guild override or in the manager's global defaults; False otherwise.

Raises

TypeError If guild is not an instance of :class:classes.structs.Guild.Guild.

Source code in classes/managers/FlagsManager.py
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def has_flag(self, guild: Union[Guild, Any], flag: str) -> bool:
    """
    Check if a flag exists for a guild (override) or globally.

    Parameters
    ----------
    guild:
        The target :class:`classes.structs.Guild.Guild` instance.
    flag:
        The flag identifier.

    Returns
    -------
    bool
        ``True`` if the flag is available either in the guild override or in
        the manager's global defaults; ``False`` otherwise.

    Raises
    ------
    TypeError
        If ``guild`` is not an instance of :class:`classes.structs.Guild.Guild`.
    """
    if not isinstance(guild, Guild):
        self.logger.error(f"Expected a Guild instance, got {type(guild).__name__}.")
        raise TypeError("Invalid guild type.")
    return guild.flags.has(flag) or flag in self.flags

register_flag(flag, default_value)

Register (or overwrite) a global flag and its default value.

Parameters

flag: Unique flag identifier (e.g., "features.xp_enabled"). default_value: Default value used when the guild has no explicit override.

Returns

FlagsManager Self, to allow call chaining.

Examples

flags.register_flag("features.xp_enabled", True)

Source code in classes/managers/FlagsManager.py
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def register_flag(self, flag: str, default_value: Union[str, bool, list]):
    """
    Register (or overwrite) a global flag and its default value.

    Parameters
    ----------
    flag:
        Unique flag identifier (e.g., ``"features.xp_enabled"``).
    default_value:
        Default value used when the guild has no explicit override.

    Returns
    -------
    FlagsManager
        Self, to allow call chaining.

    Examples
    --------
    >>> flags.register_flag("features.xp_enabled", True)
    """
    if flag in self.flags:
        self.logger.warning(f"Flag {flag} already exists, overwriting it.")
    self.flags[flag] = default_value
    return self

classes.managers.SlashManager

SlashManager

Helper for registering :class:classes.structs.SlashCommand.SlashCommand objects.

This manager wires slash commands into the client's command tree and performs the necessary sync operations (globally or per-guild).

Source code in classes/managers/SlashManager.py
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
class SlashManager:
    """
    Helper for registering :class:`classes.structs.SlashCommand.SlashCommand` objects.

    This manager wires slash commands into the client's command tree and performs
    the necessary sync operations (globally or per-guild).
    """
    def __init__(self, client: Client):
        """
        Parameters
        ----------
        client:
            The Discord client/bot that owns the command tree.
        """
        self.client = client
        self.logger = logging.getLogger("SlashManager")

    async def register_global_commands(self, commands: List[SlashCommand]):
        """
        Register and sync a list of slash commands **globally**.

        Parameters
        ----------
        commands:
            List of :class:`classes.structs.SlashCommand.SlashCommand` instances
            to be added to the application's global command tree.

        Notes
        -----
        Discord caches global commands and propagation can take some minutes.
        """
        try:
            if not isinstance(commands, list):
                raise TypeError("Commands must be a list of SlashCommand instances.")

            # Add each slash command to the bot's command tree
            for cmd in commands:
                if not isinstance(cmd, SlashCommand):
                    self.logger.warning(f"Invalid command type: {type(cmd)}. Skipping.")
                    continue
                self.client.tree.add_command(cmd.data)
                self.logger.debug(f"Added global slash command: {cmd.data.name}")

            # Sync the tree globally
            await self.client.tree.sync()
            self.logger.info(f"Registered {len(commands)} global slash commands.")
        except Exception as e:
            self.logger.exception("Failed to register global commands:", exc_info=e)

    async def register_commands_for_guild(self, commands: List[SlashCommand], guild_ids: List[int]):
        """
        Register and sync slash commands **for specific guilds**.

        Parameters
        ----------
        commands:
            List of :class:`classes.structs.SlashCommand.SlashCommand` instances.
        guild_ids:
            Guild IDs where the commands should be registered.

        Notes
        -----
        Guild-specific sync is near-instant and ideal for testing.
        """
        try:
            if not isinstance(commands, list):
                raise TypeError("Commands must be a list of SlashCommand instances.")

            for cmd in commands:
                if not isinstance(cmd, SlashCommand):
                    self.logger.warning(f"Invalid command type: {type(cmd)}. Skipping.")
                    continue
                # Register each command for the specified guilds
                for guild_id in guild_ids:
                    guild = self.client.get_guild(guild_id)
                    if not guild:
                        self.logger.warning(f"Guild with ID {guild_id} not found. Skipping.")
                        continue
                    self.client.tree.add_command(cmd.data, guild=guild)
                    self.logger.debug(f"Added guild-specific slash command: {cmd.data.name} to guild ID: {guild_id}")

            # Sync the tree for each guild
            for guild_id in guild_ids:
                guild = self.client.get_guild(guild_id)
                if guild:
                    await self.client.tree.sync(guild=guild)
                    self.logger.info(f"Registered {len(commands)} slash commands for guild ID {guild_id}.")
        except Exception as e:
            self.logger.exception("Failed to register guild-specific commands:", exc_info=e)

__init__(client)

Parameters

client: The Discord client/bot that owns the command tree.

Source code in classes/managers/SlashManager.py
13
14
15
16
17
18
19
20
21
def __init__(self, client: Client):
    """
    Parameters
    ----------
    client:
        The Discord client/bot that owns the command tree.
    """
    self.client = client
    self.logger = logging.getLogger("SlashManager")

register_commands_for_guild(commands, guild_ids) async

Register and sync slash commands for specific guilds.

Parameters

commands: List of :class:classes.structs.SlashCommand.SlashCommand instances. guild_ids: Guild IDs where the commands should be registered.

Notes

Guild-specific sync is near-instant and ideal for testing.

Source code in classes/managers/SlashManager.py
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
async def register_commands_for_guild(self, commands: List[SlashCommand], guild_ids: List[int]):
    """
    Register and sync slash commands **for specific guilds**.

    Parameters
    ----------
    commands:
        List of :class:`classes.structs.SlashCommand.SlashCommand` instances.
    guild_ids:
        Guild IDs where the commands should be registered.

    Notes
    -----
    Guild-specific sync is near-instant and ideal for testing.
    """
    try:
        if not isinstance(commands, list):
            raise TypeError("Commands must be a list of SlashCommand instances.")

        for cmd in commands:
            if not isinstance(cmd, SlashCommand):
                self.logger.warning(f"Invalid command type: {type(cmd)}. Skipping.")
                continue
            # Register each command for the specified guilds
            for guild_id in guild_ids:
                guild = self.client.get_guild(guild_id)
                if not guild:
                    self.logger.warning(f"Guild with ID {guild_id} not found. Skipping.")
                    continue
                self.client.tree.add_command(cmd.data, guild=guild)
                self.logger.debug(f"Added guild-specific slash command: {cmd.data.name} to guild ID: {guild_id}")

        # Sync the tree for each guild
        for guild_id in guild_ids:
            guild = self.client.get_guild(guild_id)
            if guild:
                await self.client.tree.sync(guild=guild)
                self.logger.info(f"Registered {len(commands)} slash commands for guild ID {guild_id}.")
    except Exception as e:
        self.logger.exception("Failed to register guild-specific commands:", exc_info=e)

register_global_commands(commands) async

Register and sync a list of slash commands globally.

Parameters

commands: List of :class:classes.structs.SlashCommand.SlashCommand instances to be added to the application's global command tree.

Notes

Discord caches global commands and propagation can take some minutes.

Source code in classes/managers/SlashManager.py
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
async def register_global_commands(self, commands: List[SlashCommand]):
    """
    Register and sync a list of slash commands **globally**.

    Parameters
    ----------
    commands:
        List of :class:`classes.structs.SlashCommand.SlashCommand` instances
        to be added to the application's global command tree.

    Notes
    -----
    Discord caches global commands and propagation can take some minutes.
    """
    try:
        if not isinstance(commands, list):
            raise TypeError("Commands must be a list of SlashCommand instances.")

        # Add each slash command to the bot's command tree
        for cmd in commands:
            if not isinstance(cmd, SlashCommand):
                self.logger.warning(f"Invalid command type: {type(cmd)}. Skipping.")
                continue
            self.client.tree.add_command(cmd.data)
            self.logger.debug(f"Added global slash command: {cmd.data.name}")

        # Sync the tree globally
        await self.client.tree.sync()
        self.logger.info(f"Registered {len(commands)} global slash commands.")
    except Exception as e:
        self.logger.exception("Failed to register global commands:", exc_info=e)