aurora/importers/aurora.py

@@ -1,20 +1,25 @@
 # pylint: disable=duplicate-code
 import json
-from datetime import timedelta
-from typing import Dict
+import os
+from time import time
+from typing import Dict, List
 
-from discord import ButtonStyle, Interaction, Message, ui
-from redbot.core import commands
-from redbot.core.utils.chat_formatting import box, warning
+from discord import ButtonStyle, File, Interaction, Message, ui
+from redbot.core import commands, data_manager
+from redbot.core.utils.chat_formatting import warning
 
-from ..utilities.database import connect, create_guild_table, mysql_log
+from ..models.moderation import Moderation
+from ..models.type import Type, type_registry
+from ..utilities.json import dump
+from ..utilities.utils import create_guild_table, timedelta_from_string
 
 
 class ImportAuroraView(ui.View):
-    def __init__(self, timeout, ctx, message):
+    def __init__(self, timeout, ctx, message, data: List[Dict[str, any]]):
         super().__init__()
         self.ctx: commands.Context = ctx
         self.message: Message = message
+        self.data: List[Dict[str, any]] = data
 
     @ui.button(label="Yes", style=ButtonStyle.success)
     async def import_button_y(
@@ -25,14 +30,8 @@ class ImportAuroraView(ui.View):
             "Deleting original table...", ephemeral=True
         )
 
-        database = connect()
-        cursor = database.cursor()
-
         query = f"DROP TABLE IF EXISTS moderation_{self.ctx.guild.id};"
-        cursor.execute(query)
-
-        cursor.close()
-        database.commit()
+        await Moderation.execute(query=query, return_obj=False)
 
         await interaction.edit_original_response(content="Creating new table...")
 
@@ -40,73 +39,102 @@ class ImportAuroraView(ui.View):
 
         await interaction.edit_original_response(content="Importing moderations...")
 
-        file = await self.ctx.message.attachments[0].read()
-        data: list[dict] = sorted(json.loads(file), key=lambda x: x["moderation_id"])
-
-        user_mod_types = ["NOTE", "WARN", "ADDROLE", "REMOVEROLE", "MUTE", "UNMUTE", "KICK", "TEMPBAN", "BAN", "UNBAN"]
-
-        channel_mod_types = ["SLOWMODE", "LOCKDOWN"]
-
         failed_cases = []
 
-        for case in data:
+        for case in self.data:
             if case["moderation_id"] == 0:
                 continue
 
+            moderation_type: Type = type_registry[case["moderation_type"].lower()]
             if "target_type" not in case or not case["target_type"]:
-                if case["moderation_type"] in user_mod_types:
-                    case["target_type"] = "USER"
-                elif case["moderation_type"] in channel_mod_types:
-                    case["target_type"] = "CHANNEL"
+                if moderation_type.channel:
+                    case["target_type"] = "channel"
                 else:
-                    case["target_type"] = "USER"
+                    case["target_type"] = "user"
 
             if "role_id" not in case or not case["role_id"]:
-                case["role_id"] = 0
+                case["role_id"] = None
+            else:
+                case["role_id"] = int(case["role_id"])
 
-            if "changes" not in case or not case["changes"]:
-                case["changes"] = []
+            case["target_id"] = int(case["target_id"])
+            case["moderator_id"] = int(case["moderator_id"])
+
+            changes = case.get("changes", None)
+            if not changes:
+                changes = []
+            else:
+                if not isinstance(changes, list):
+                    changes = json.loads(changes)
+                    if isinstance(changes, str):
+                        changes: list[dict] = json.loads(changes)
+
+            for change in changes:
+                if "bot" in change:
+                    del change["bot"]
 
             if "metadata" not in case:
                 metadata = {}
             else:
-                metadata: Dict[str, any] = json.loads(case["metadata"])
+                if isinstance(case["metadata"], str):
+                    metadata: Dict[str, any] = json.loads(case["metadata"])
+                else:
+                    metadata = case["metadata"]
             if not metadata.get("imported_from"):
                 metadata.update({"imported_from": "Aurora"})
+            metadata.update({"imported_timestamp": int(time())})
 
-            if case["duration"] != "NULL":
-                hours, minutes, seconds = map(int, case["duration"].split(":"))
-                duration = timedelta(hours=hours, minutes=minutes, seconds=seconds)
+            if case["duration"] != "NULL" and case["duration"] is not None:
+                duration = timedelta_from_string(case["duration"])
+                if moderation_type.key == "ban":
+                    moderation_type = type_registry["tempban"]
             else:
-                duration = "NULL"
+                duration = None
 
-            await mysql_log(
-                self.ctx.guild.id,
-                case["moderator_id"],
-                case["moderation_type"],
-                case["target_type"],
-                case["target_id"],
-                case["role_id"],
-                duration,
-                case["reason"],
-                timestamp=case["timestamp"],
-                resolved=case["resolved"],
-                resolved_by=case["resolved_by"],
-                resolved_reason=case["resolve_reason"],
-                expired=case["expired"],
-                changes=case["changes"],
-                metadata=metadata,
-                database=database,
-            )
+            try:
+                await Moderation.log(
+                    bot=interaction.client,
+                    guild_id=self.ctx.guild.id,
+                    moderator_id=case["moderator_id"],
+                    moderation_type=moderation_type,
+                    target_type=case["target_type"].lower(),
+                    target_id=case["target_id"],
+                    role_id=case["role_id"],
+                    duration=duration,
+                    reason=case["reason"],
+                    timestamp=case["timestamp"],
+                    resolved=case["resolved"],
+                    resolved_by=case["resolved_by"],
+                    resolved_reason=case["resolve_reason"],
+                    expired=case["expired"],
+                    changes=changes,
+                    metadata=metadata,
+                    return_obj=False
+                )
+            except Exception as e: # pylint: disable=broad-exception-caught
+                failed_cases.append(str(case["moderation_id"]) + f": {e}")
 
         await interaction.edit_original_response(content="Import complete.")
         if failed_cases:
-            await interaction.edit_original_response(
-                content="Import complete.\n"
-                + warning("Failed to import the following cases:\n")
-                + box(failed_cases)
+            filename = (
+                str(data_manager.cog_data_path(cog_instance=self))
+                + str(os.sep)
+                + f"failed_cases_{interaction.guild.id}.json"
             )
 
+            with open(filename, "w", encoding="utf-8") as f:
+                dump(obj=failed_cases, fp=f, indent=2)
+
+            await interaction.channel.send(
+                content="Import complete.\n"
+                + warning("Failed to import the following cases:\n"),
+                file=File(
+                        filename, f"failed_cases_{interaction.guild.id}.json"
+                    )
+            )
+
+            os.remove(filename)
+
     @ui.button(label="No", style=ButtonStyle.danger)
     async def import_button_n(
         self, interaction: Interaction, button: ui.Button

aurora/importers/galacticbot.py

@@ -1,18 +1,21 @@
 # pylint: disable=duplicate-code
 import json
 from datetime import timedelta
+from time import time
 
 from discord import ButtonStyle, Interaction, Message, ui
 from redbot.core import commands
 from redbot.core.utils.chat_formatting import box, warning
 
-from ..utilities.database import connect, create_guild_table, mysql_log
+from ..models.moderation import Change, Moderation
+from ..utilities.utils import create_guild_table
 
 
 class ImportGalacticBotView(ui.View):
     def __init__(self, timeout, ctx, message):
         super().__init__()
         self.ctx: commands.Context = ctx
+        self.timeout = timeout
         self.message: Message = message
 
     @ui.button(label="Yes", style=ButtonStyle.success)
@@ -24,14 +27,14 @@ class ImportGalacticBotView(ui.View):
             "Deleting original table...", ephemeral=True
         )
 
-        database = connect()
-        cursor = database.cursor()
+        database = await Moderation.connect()
+        cursor = await database.cursor()
 
         query = f"DROP TABLE IF EXISTS moderation_{self.ctx.guild.id};"
-        cursor.execute(query)
+        await cursor.execute(query)
 
-        cursor.close()
-        database.commit()
+        await cursor.close()
+        await database.commit()
 
         await interaction.edit_original_response(content="Creating new table...")
 
@@ -67,12 +70,12 @@ class ImportGalacticBotView(ui.View):
                 if case["duration"] is not None and float(case["duration"]) != 0:
                     duration = timedelta(seconds=round(float(case["duration"]) / 1000))
                 else:
-                    duration = "NULL"
+                    duration = None
             except OverflowError:
                 failed_cases.append(case["case"])
                 continue
 
-            metadata = {"imported_from": "GalacticBot"}
+            metadata = {"imported_from": "GalacticBot", "imported_timestamp": int(time())}
 
             if case["type"] == "SLOWMODE":
                 metadata["seconds"] = case["data"]["seconds"]
@@ -98,37 +101,37 @@ class ImportGalacticBotView(ui.View):
                 if resolved_timestamp is None:
                     resolved_timestamp = timestamp
                 changes = [
-                    {
+                    Change.from_dict(interaction.client, {
                         "type": "ORIGINAL",
                         "reason": case["reason"],
                         "user_id": case["executor"],
                         "timestamp": timestamp,
-                    },
-                    {
+                    }),
+                    Change.from_dict(interaction.client, {
                         "type": "RESOLVE",
                         "reason": resolved_reason,
                         "user_id": resolved_by,
                         "timestamp": resolved_timestamp,
-                    },
+                    }),
                 ]
             else:
-                resolved = 0
-                resolved_by = "NULL"
-                resolved_reason = "NULL"
-                changes = []
+                resolved = None
+                resolved_by = None
+                resolved_reason = None
+                changes = None
 
             if case["reason"] and case["reason"] != "N/A":
                 reason = case["reason"]
             else:
-                reason = "NULL"
+                reason = None
 
-            await mysql_log(
+            await Moderation.log(
                 self.ctx.guild.id,
                 case["executor"],
                 case["type"],
                 case["targetType"],
                 case["target"],
-                0,
+                None,
                 duration,
                 reason,
                 timestamp=timestamp,

aurora/info.json

@@ -1,5 +1,5 @@
 {
-  "author" : ["SeaswimmerTheFsh (seasw.)"],
+  "author" : ["Seaswimmer (cswimr)"],
   "install_msg" : "Thank you for installing Aurora!\nMost of this cog's functionality requires enabling slash commands.\nYou can find the source code of this cog [here](https://coastalcommits.com/SeaswimmerTheFsh/SeaCogs).",
   "name" : "Aurora",
   "short" : "A full replacement for Red's core Mod cogs.",
@@ -9,6 +9,7 @@
   "disabled": false,
   "min_bot_version": "3.5.0",
   "min_python_version": [3, 10, 0],
+  "requirements": ["pydantic", "aiosqlite", "phx-class-registry==5.0.0"],
   "tags": [
       "mod",
       "moderate",

aurora/menus/addrole.py

@@ -1,20 +1,24 @@
 from discord import ButtonStyle, Interaction, Message, ui
+from discord.errors import NotFound
 from redbot.core import commands
 from redbot.core.utils.chat_formatting import error
 
-from aurora.utilities.config import config
-from aurora.utilities.factory import addrole_embed
+from ..utilities.config import config
+from ..utilities.factory import addrole_embed
 
 
 class Addrole(ui.View):
-    def __init__(self, ctx: commands.Context, message: Message, timeout: int = None):
+    def __init__(self, ctx: commands.Context, message: Message, timeout: int | None = None):
         super().__init__()
         self.ctx = ctx
         self.message = message
         self.timeout = timeout
 
     async def on_timeout(self):
-        await self.message.edit(view=None)
+        try:
+            await self.message.edit(view=None)
+        except NotFound:
+            pass
 
     @ui.select(cls=ui.RoleSelect, placeholder="Select a role", min_values=0, max_values=25)
     async def addrole_select(self, interaction: Interaction, select: ui.RoleSelect):

aurora/menus/guild.py

@@ -1,20 +1,24 @@
 from discord import ButtonStyle, Interaction, Message, ui
+from discord.errors import NotFound
 from redbot.core import commands
 
-from aurora.utilities.config import config
-from aurora.utilities.factory import guild_embed
-from aurora.utilities.utils import create_pagesize_options
+from ..utilities.config import config
+from ..utilities.factory import guild_embed
+from ..utilities.utils import create_pagesize_options
 
 
 class Guild(ui.View):
-    def __init__(self, ctx: commands.Context, message: Message, timeout: int = None):
+    def __init__(self, ctx: commands.Context, message: Message, timeout: int | None = None):
         super().__init__()
         self.ctx = ctx
         self.message = message
         self.timeout = timeout
 
     async def on_timeout(self):
-        await self.message.edit(view=None)
+        try:
+            await self.message.edit(view=None)
+        except NotFound:
+            pass
 
     @ui.button(label="Show Moderator", style=ButtonStyle.green, row=0)
     async def show_moderator(self, interaction: Interaction, button: ui.Button): # pylint: disable=unused-argument

aurora/menus/immune.py

@@ -1,20 +1,24 @@
 from discord import ButtonStyle, Interaction, Message, ui
+from discord.errors import NotFound
 from redbot.core import commands
 from redbot.core.utils.chat_formatting import error
 
-from aurora.utilities.config import config
-from aurora.utilities.factory import immune_embed
+from ..utilities.config import config
+from ..utilities.factory import immune_embed
 
 
 class Immune(ui.View):
-    def __init__(self, ctx: commands.Context, message: Message, timeout: int = None):
+    def __init__(self, ctx: commands.Context, message: Message, timeout: int | None = None):
         super().__init__()
         self.ctx = ctx
         self.message = message
         self.timeout = timeout
 
     async def on_timeout(self):
-        await self.message.edit(view=None)
+        try:
+            await self.message.edit(view=None)
+        except NotFound:
+            pass
 
     @ui.select(cls=ui.RoleSelect, placeholder="Select a role", min_values=0, max_values=25)
     async def immune_select(self, interaction: Interaction, select: ui.RoleSelect):

aurora/menus/overrides.py

@@ -1,20 +1,24 @@
 from discord import ButtonStyle, Interaction, Message, ui
+from discord.errors import NotFound
 from redbot.core import commands
 
-from aurora.utilities.config import config
-from aurora.utilities.factory import overrides_embed
-from aurora.utilities.utils import create_pagesize_options
+from ..utilities.config import config
+from ..utilities.factory import overrides_embed
+from ..utilities.utils import create_pagesize_options
 
 
 class Overrides(ui.View):
-    def __init__(self, ctx: commands.Context, message: Message, timeout: int = None):
+    def __init__(self, ctx: commands.Context, message: Message, timeout: int | None = None):
         super().__init__()
         self.ctx = ctx
         self.message = message
         self.timeout = timeout
 
     async def on_timeout(self):
-        await self.message.edit(view=None)
+        try:
+            await self.message.edit(view=None)
+        except NotFound:
+            pass
 
     @ui.button(label="Auto Evidence Format", style=ButtonStyle.green, row=0)
     async def auto_evidenceformat(self, interaction: Interaction, button: ui.Button): # pylint: disable=unused-argument

aurora/menus/types.py

@@ -0,0 +1,71 @@
+from discord import ButtonStyle, Interaction, Message, ui
+from discord.errors import NotFound
+from redbot.core import commands
+
+from ..models.type import Type
+from ..utilities.config import config
+from ..utilities.factory import type_embed
+
+
+class Types(ui.View):
+    def __init__(self, ctx: commands.Context, message: Message, moderation_type: Type, timeout: int | None = None):
+        super().__init__()
+        self.ctx = ctx
+        self.message = message
+        self.type = moderation_type
+        self.timeout = timeout
+
+    async def on_timeout(self):
+        try:
+            await self.message.edit(view=None)
+        except NotFound:
+            pass
+
+    @ui.button(label="Show in History", style=ButtonStyle.green, row=0)
+    async def show_in_history(self, interaction: Interaction, button: ui.Button): # pylint: disable=unused-argument
+        if not interaction.user.guild_permissions.manage_guild and not interaction.user.guild_permissions.administrator:
+            await interaction.response.send_message("You must have the manage guild permission to change this setting.", ephemeral=True)
+            return
+        await interaction.response.defer()
+        current_setting = await config.custom("types", interaction.guild.id, self.type.key).show_in_history()
+        await config.custom("types", interaction.guild.id, self.type.key).show_in_history.set(not current_setting)
+        await interaction.message.edit(embed=await type_embed(self.ctx, self.type))
+
+    @ui.button(label="Show Moderator", style=ButtonStyle.green, row=0)
+    async def show_moderator(self, interaction: Interaction, button: ui.Button): # pylint: disable=unused-argument
+        if not interaction.user.guild_permissions.manage_guild and not interaction.user.guild_permissions.administrator:
+            await interaction.response.send_message("You must have the manage guild permission to change this setting.", ephemeral=True)
+            return
+        await interaction.response.defer()
+        current_setting = await config.custom("types", interaction.guild.id, self.type.key).show_moderator()
+        await config.custom("types", interaction.guild.id, self.type.key).show_moderator.set(not current_setting)
+        await interaction.message.edit(embed=await type_embed(self.ctx, self.type))
+
+    @ui.button(label="Use Discord Permissions", style=ButtonStyle.green, row=0)
+    async def use_discord_permissions(self, interaction: Interaction, button: ui.Button): # pylint: disable=unused-argument
+        if not interaction.user.guild_permissions.manage_guild and not interaction.user.guild_permissions.administrator:
+            await interaction.response.send_message("You must have the manage guild permission to change this setting.", ephemeral=True)
+            return
+        await interaction.response.defer()
+        current_setting = await config.custom("types", interaction.guild.id, self.type.key).use_discord_permissions()
+        await config.custom("types", interaction.guild.id, self.type.key).use_discord_permissions.set(not current_setting)
+        await interaction.message.edit(embed=await type_embed(self.ctx, self.type))
+
+    @ui.button(label="DM Users", style=ButtonStyle.green, row=0)
+    async def dm_users(self, interaction: Interaction, button: ui.Button): # pylint: disable=unused-argument
+        if not interaction.user.guild_permissions.manage_guild and not interaction.user.guild_permissions.administrator:
+            await interaction.response.send_message("You must have the manage guild permission to change this setting.", ephemeral=True)
+            return
+        await interaction.response.defer()
+        current_setting = await config.custom("types", interaction.guild.id, self.type.key).dm_users()
+        await config.custom("types", interaction.guild.id, self.type.key).dm_users.set(not current_setting)
+        await interaction.message.edit(embed=await type_embed(self.ctx, self.type))
+
+    @ui.button(label="Reset", style=ButtonStyle.red, row=1)
+    async def reset(self, interaction: Interaction, button: ui.Button): # pylint: disable=unused-argument
+        if not interaction.user.guild_permissions.manage_guild and not interaction.user.guild_permissions.administrator:
+            await interaction.response.send_message("You must have the manage guild permission to change this setting.", ephemeral=True)
+            return
+        await interaction.response.defer()
+        await config.custom("types", interaction.guild.id, self.type.key).clear()
+        await interaction.message.edit(embed=await type_embed(self.ctx, self.type))

aurora/models/__init__.py

@@ -0,0 +1,2 @@
+from .moderation_types import *  # noqa: F403
+# This just imports all the built-in moderation types so they can be registered, as they aren't imported anywhere else.

aurora/models/base.py

@@ -0,0 +1,30 @@
+from typing import Any, Optional
+
+from discord import Guild
+from pydantic import BaseModel, ConfigDict
+from redbot.core.bot import Red
+
+
+class AuroraBaseModel(BaseModel):
+    """Base class for all models in Aurora."""
+    model_config = ConfigDict(ignored_types=(Red,), arbitrary_types_allowed=True)
+    bot: Red
+
+    def dump(self) -> dict:
+        return self.model_dump(exclude={"bot"})
+
+    def to_json(self, indent: int | None = None, file: Any | None = None, **kwargs) -> str:
+        from ..utilities.json import dump, dumps  # pylint: disable=cyclic-import
+        return dump(self.dump(), file, indent=indent, **kwargs) if file else dumps(self.dump(), indent=indent, **kwargs)
+
+class AuroraGuildModel(AuroraBaseModel):
+    """Subclass of AuroraBaseModel that includes a guild_id attribute and a guild attribute."""
+    model_config = ConfigDict(ignored_types=(Red, Guild), arbitrary_types_allowed=True)
+    guild_id: int
+    guild: Optional[Guild] = None
+
+    def dump(self) -> dict:
+        return self.model_dump(exclude={"bot", "guild_id", "guild"})
+
+    def __repr__(self) -> str:
+        return f"<{self.__class__.__name__} guild_id={self.guild_id}>"

aurora/models/change.py

@@ -0,0 +1,83 @@
+import json
+from datetime import datetime, timedelta
+from typing import Literal, Optional
+
+from redbot.core.bot import Red
+
+from ..utilities.utils import timedelta_from_string
+from .base import AuroraBaseModel
+from .partials import PartialUser
+
+
+class Change(AuroraBaseModel):
+    type: Literal["ORIGINAL", "RESOLVE", "EDIT"]
+    timestamp: datetime
+    user_id: int
+    reason: Optional[str] = None
+    duration: Optional[timedelta] = None
+    end_timestamp: Optional[datetime] = None
+
+    @property
+    def unix_timestamp(self) -> int:
+        return int(self.timestamp.timestamp())
+
+    @property
+    def unix_end_timestamp(self) -> Optional[int]:
+        if self.end_timestamp:
+            return int(self.end_timestamp.timestamp())
+        return None
+
+    def __str__(self):
+        return f"{self.type} {self.user_id} {self.reason}"
+
+    def __repr__(self) -> str:
+        attrs = [
+            ('type', self.type),
+            ('timestamp', self.timestamp),
+            ('user_id', self.user_id),
+            ('reason', self.reason),
+            ('duration', self.duration),
+            ('end_timestamp', self.end_timestamp),
+        ]
+        joined = ' '.join(f'{key}={value!r}' for key, value in attrs)
+        return f"<{self.__class__.__name__} {joined}>"
+
+    async def get_user(self) -> "PartialUser":
+        return await PartialUser.from_id(self.bot, self.user_id)
+
+    @classmethod
+    def from_dict(cls, bot: Red, data: dict) -> "Change":
+        if isinstance(data, str):
+            data = json.loads(data)
+        if data.get('duration') and not isinstance(data["duration"], timedelta) and not data["duration"] == "NULL":
+            duration = timedelta_from_string(data["duration"])
+        elif data.get('duration') and isinstance(data["duration"], timedelta):
+            duration = data["duration"]
+        else:
+            duration = None
+
+        if data.get('end_timestamp') and not isinstance(data["end_timestamp"], datetime):
+            end_timestamp = datetime.fromtimestamp(data["end_timestamp"])
+        elif data.get('end_timestamp') and isinstance(data["end_timestamp"], datetime):
+            end_timestamp = data["end_timestamp"]
+        else:
+            end_timestamp = None
+
+        if not isinstance(data["timestamp"], datetime):
+            timestamp = datetime.fromtimestamp(data["timestamp"])
+        else:
+            timestamp = data["timestamp"]
+
+        try:
+            data["user_id"] = int(data["user_id"])
+        except ValueError:
+            data["user_id"] = 0
+
+        data.update({
+            "timestamp": timestamp,
+            "end_timestamp": end_timestamp,
+            "duration": duration
+        })
+        if "bot" in data:
+            del data["bot"]
+        return cls(bot=bot, **data)

aurora/models/moderation.py

@@ -0,0 +1,565 @@
+import json
+import sqlite3
+from datetime import datetime, timedelta
+from time import time
+from typing import Dict, Iterable, List, Optional, Tuple, Union
+
+import discord
+from aiosqlite import Connection, Cursor, OperationalError, Row
+from aiosqlite import connect as aiosqlite_connect
+from redbot.core import data_manager
+from redbot.core.bot import Red
+
+from ..utilities.logger import logger
+from ..utilities.utils import timedelta_to_string
+from .base import AuroraGuildModel
+from .change import Change
+from .partials import PartialChannel, PartialRole, PartialUser
+from .type import Type, type_registry
+
+
+class Moderation(AuroraGuildModel):
+    """This class represents a moderation case in the database.
+
+    Attributes:
+        bot (Red): The bot instance.
+        guild (discord.Guild): The guild the case belongs to.
+        moderation_id (int): The ID of the moderation case.
+        timestamp (datetime): The timestamp of the case.
+        moderation_type (Type): The type of moderation case.
+        target_type (str): The type of target. Should be either `user` or `channel`.
+        target_id (int): The ID of the target.
+        moderator_id (int): The ID of the moderator who issued the case.
+        role_id (int): The ID of the role, if applicable.
+        duration (timedelta): The duration of the case, if applicable.
+        end_timestamp (datetime): The end timestamp of the case, if applicable.
+        reason (str): The reason for the case.
+        resolved (bool): Whether the case is resolved.
+        resolved_by (int): The ID of the user who resolved the case.
+        resolve_reason (str): The reason the case was resolved.
+        expired (bool): Whether the case is expired.
+        changes (List[Change]): A list of changes to the case.
+        metadata (Dict): A dictionary of metadata stored with the case.
+
+    Properties:
+        id (int): The ID of the case.
+        type (Type): The type of the case.
+        unix_timestamp (int): The timestamp of the case as a Unix timestamp.
+
+    Methods:
+        get_moderator: Gets the moderator who issued the case.
+        get_target: Gets the target of the case.
+        get_resolved_by: Gets the user who resolved the case.
+        get_role: Gets the role, if applicable.
+        resolve: Resolves the case.
+        update: Updates the case in the database.
+
+    Class Methods:
+        from_dict: Creates a `Moderation` object from a dictionary.
+        from_result: Creates a `Moderation` object from a database result.
+        execute: Executes a query on the database.
+        get_latest: Gets the latest cases from the database.
+        get_next_case_number: Gets the next case number to use.
+        find_by_id: Finds a case by its ID.
+        find_by_target: Finds cases by the target.
+        find_by_moderator: Finds cases by the moderator.
+        log: Logs a moderation case in the database.
+
+    Static Methods:
+        connect: Connects to the SQLite database.
+    """
+
+    moderation_id: int
+    timestamp: datetime
+    moderation_type: Type
+    target_type: str
+    target_id: int
+    moderator_id: int
+    role_id: Optional[int] = None
+    duration: Optional[timedelta] = None
+    end_timestamp: Optional[datetime] = None
+    reason: Optional[str] = None
+    resolved: bool
+    resolved_by: Optional[int] = None
+    resolve_reason: Optional[str] = None
+    expired: bool
+    changes: List["Change"]
+    metadata: Dict
+
+    @property
+    def id(self) -> int:
+        return self.moderation_id
+
+    @property
+    def type(self) -> Type:
+        return self.moderation_type
+
+    @property
+    def unix_timestamp(self) -> int:
+        return int(self.timestamp.timestamp())
+
+    async def get_moderator(self) -> "PartialUser":
+        return await PartialUser.from_id(self.bot, self.moderator_id)
+
+    async def get_target(self) -> Union["PartialUser", "PartialChannel"]:
+        if self.target_type.lower() == "user":
+            return await PartialUser.from_id(self.bot, self.target_id)
+        return await PartialChannel.from_id(self.bot, self.target_id, self.guild)
+
+    async def get_resolved_by(self) -> Optional["PartialUser"]:
+        if self.resolved_by:
+            return await PartialUser.from_id(self.bot, self.resolved_by)
+        return None
+
+    async def get_role(self) -> Optional["PartialRole"]:
+        if self.role_id:
+            return await PartialRole.from_id(self.bot, self.guild, self.role_id)
+        return None
+
+    def __str__(self) -> str:
+        return f"{self.moderation_type} {self.target_type} {self.target_id} {self.reason}"
+
+    def __int__(self) -> int:
+        return self.moderation_id
+
+    def __repr__(self) -> str:
+        attrs = [
+            ('guild_id', self.guild_id),
+            ('moderation_id', self.moderation_id),
+            ('timestamp', self.timestamp),
+            ('type', self.type),
+            ('target_type', self.target_type),
+            ('target_id', self.target_id),
+            ('moderator_id', self.moderator_id),
+            ('role_id', self.role_id),
+            ('duration', self.duration),
+            ('end_timestamp', self.end_timestamp),
+            ('reason', self.reason),
+            ('resolved', self.resolved),
+            ('resolved_by', self.resolved_by),
+            ('resolve_reason', self.resolve_reason),
+            ('expired', self.expired),
+            ('changes', self.changes),
+            ('metadata', self.metadata),
+        ]
+        joined = ' '.join(f'{key}={value!r}' for key, value in attrs)
+        return f"<{self.__class__.__name__} {joined}>"
+
+    async def resolve(self, resolved_by: int, reason: str) -> Tuple[bool, str]:
+        if self.resolved:
+            raise ValueError("Case is already resolved!")
+
+        self.resolved = True
+        self.resolved_by = resolved_by
+        self.resolve_reason = reason
+
+        success, msg = await self.type.resolve_handler(moderation=self, reason=reason)
+
+        if not self.changes:
+            self.changes.append(Change.from_dict(self.bot, {
+                "type": "ORIGINAL",
+                "timestamp": self.timestamp,
+                "reason": self.reason,
+                "user_id": self.moderator_id,
+                "duration": self.duration,
+                "end_timestamp": self.end_timestamp,
+            }))
+        self.changes.append(Change.from_dict(self.bot, {
+            "type": "RESOLVE",
+            "timestamp": datetime.now(),
+            "reason": reason,
+            "user_id": resolved_by,
+        }))
+
+        await self.update()
+        return success, msg
+
+    async def update(self) -> None:
+        from ..utilities.json import dumps
+        query = f"UPDATE moderation_{self.guild_id} SET timestamp = ?, moderation_type = ?, target_type = ?, moderator_id = ?, role_id = ?, duration = ?, end_timestamp = ?, reason = ?, resolved = ?, resolved_by = ?, resolve_reason = ?, expired = ?, changes = ?, metadata = ? WHERE moderation_id = ?;"
+
+        await self.execute(query, (
+            self.timestamp.timestamp(),
+            self.moderation_type.key,
+            self.target_type,
+            self.moderator_id,
+            self.role_id,
+            timedelta_to_string(self.duration) if self.duration else None,
+            self.end_timestamp.timestamp() if self.end_timestamp else None,
+            self.reason,
+            self.resolved,
+            self.resolved_by,
+            self.resolve_reason,
+            self.expired,
+            dumps(self.changes),
+            dumps(self.metadata),
+            self.moderation_id,
+        ))
+
+        logger.verbose("Row updated in moderation_%s!\n%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s",
+            self.moderation_id,
+            self.guild_id,
+            self.timestamp.timestamp(),
+            self.moderation_type.key,
+            self.target_type,
+            self.moderator_id,
+            self.role_id,
+            timedelta_to_string(self.duration) if self.duration else None,
+            self.end_timestamp.timestamp() if self.end_timestamp else None,
+            self.reason,
+            self.resolved,
+            self.resolved_by,
+            self.resolve_reason,
+            self.expired,
+            dumps(self.changes),
+            dumps(self.metadata),
+        )
+
+    @classmethod
+    async def from_dict(cls, bot: Red, data: dict) -> "Moderation":
+        if data.get("guild_id"):
+            try:
+                guild = bot.get_guild(data["guild_id"])
+                if not guild:
+                    guild = await bot.fetch_guild(data["guild_id"])
+            except (discord.Forbidden, discord.HTTPException):
+                guild = None
+            data.update({"guild": guild})
+        return cls(bot=bot, **data)
+
+    @classmethod
+    async def from_result(cls, bot: Red, result: Iterable, guild_id: int) -> "Moderation":
+        if result[7] is not None and result[7] != "NULL":
+            try:
+                hours, minutes, seconds = map(int, result[7].split(':'))
+                duration = timedelta(hours=hours, minutes=minutes, seconds=seconds)
+            except ValueError as e:
+                logger.error("Error parsing duration for case %s: %s", result[0], result[7])
+                raise e
+        else:
+            duration = None
+
+        if result[14] is not None:
+            changes = json.loads(result[14])
+            change_obj_list = []
+            if changes:
+                for change in changes:
+                    change_obj_list.append(Change.from_dict(bot=bot, data=change))
+
+        if result[15] is not None:
+            metadata = json.loads(result[15])
+        else:
+            metadata = {}
+
+        moderation_type = str.lower(result[2])
+        if moderation_type in type_registry:
+            moderation_type = type_registry[moderation_type]
+        else:
+            logger.error("Unknown moderation type in case %s: %s", result[0], result[2])
+
+        case = {
+            "moderation_id": int(result[0]),
+            "guild_id": int(guild_id),
+            "timestamp": datetime.fromtimestamp(result[1]),
+            "moderation_type": moderation_type,
+            "target_type": str(result[3]),
+            "target_id": int(result[4]),
+            "moderator_id": int(result[5]),
+            "role_id": int(result[6]) if result[6] is not None else None,
+            "duration": duration,
+            "end_timestamp": datetime.fromtimestamp(result[8]) if result[8] is not None else None,
+            "reason": result[9],
+            "resolved": bool(result[10]),
+            "resolved_by": result[11],
+            "resolve_reason": result[12],
+            "expired": bool(result[13]),
+            "changes": change_obj_list,
+            "metadata": metadata if metadata else {},
+        }
+        return await cls.from_dict(bot=bot, data=case)
+
+    @staticmethod
+    async def connect() -> Connection:
+        """Connects to the SQLite database, and returns a connection object."""
+        try:
+            connection = await aiosqlite_connect(
+                database=data_manager.cog_data_path(raw_name="Aurora") / "aurora.db"
+            )
+            return connection
+
+        except OperationalError as e:
+            logger.error("Unable to access the SQLite database!\nError:\n%s", e.msg)
+            raise ConnectionRefusedError(
+                f"Unable to access the SQLite Database!\n{e.msg}"
+            ) from e
+
+    @classmethod
+    async def execute(cls, query: str, parameters: tuple | None = None, bot: Red | None = None, guild_id: int | None = None, cursor: Cursor | None = None, return_obj: bool = True) -> Union[Tuple["Moderation"], Iterable[Row]]:
+        """Executes a query on the database.
+
+        Arguments:
+            query (str): The query to execute.
+            parameters (tuple): The parameters to pass to the query.
+            bot (Red): The bot instance.
+            guild_id (int): The ID of the guild to execute the query on.
+            cursor (Cursor): The cursor to use for the query.
+            return_obj (bool): Whether to return the case object(s). Defaults to `True`. If `False`, returns a `Iterable` of `aiosqlite.Row` objects.
+        Returns: The result of the query, either as a `Tuple` of `Moderation` objects or an `Iterable` of `aiosqlite.Row` objects.
+        """
+        logger.trace("Executing query: \"%s\" with parameters \"%s\"", query, parameters)
+        if not parameters:
+            parameters = ()
+        if not cursor:
+            no_cursor = True
+            database = await cls.connect()
+            cursor = await database.cursor()
+        else:
+            no_cursor = False
+
+        try:
+            await cursor.execute(query, parameters)
+        except OperationalError as e:
+            logger.error("Error executing query: \"%s\" with parameters \"%s\"\nError:\n%s",
+                            query, parameters, e)
+            raise OperationalError(f"Error executing query: \"{query}\" with parameters \"{parameters}\"") from e
+        results = await cursor.fetchall()
+        await database.commit()
+        if no_cursor:
+            await cursor.close()
+            await database.close()
+
+        if results and return_obj and bot and guild_id:
+            cases = []
+            for result in results:
+                if result[0] == 0:
+                    continue
+                case = await cls.from_result(bot=bot, result=result, guild_id=guild_id)
+                cases.append(case)
+            return tuple(cases)
+        return results
+
+    @classmethod
+    async def get_latest(cls, bot: Red, guild_id: int, before: datetime | None = None, after: datetime | None = None, limit: int | None = None, offset: int = 0, types: Iterable[Type] | None = None, expired: bool | None = None, cursor: Cursor | None = None) -> Tuple["Moderation"]:
+        params = []
+        query = f"SELECT * FROM moderation_{guild_id}"
+        conditions = []
+
+        if types:
+            conditions.append(f"moderation_type IN ({', '.join(['?' for _ in types])})")
+            params.extend([t.key for t in types])
+        if before:
+            conditions.append("timestamp < ?")
+            params.append(int(before.timestamp()))
+        if after:
+            conditions.append("timestamp > ?")
+            params.append(int(after.timestamp()))
+        if expired is not None:
+            conditions.append("expired = ?")
+            params.append(int(expired))
+
+        if conditions:
+            query += " WHERE " + " AND ".join(conditions)
+
+        query += " ORDER BY moderation_id DESC"
+
+        if limit:
+            query += " LIMIT ? OFFSET ?"
+            params.extend((limit, offset))
+        query += ";"
+        return await cls.execute(bot=bot, guild_id=guild_id, query=query, parameters=tuple(params) if params else (), cursor=cursor)
+
+    @classmethod
+    async def get_next_case_number(cls, bot: Red, guild_id: int, cursor: Cursor | None = None) -> int:
+        result = await cls.get_latest(bot=bot, guild_id=guild_id, cursor=cursor, limit=1)
+        return (result[0].moderation_id + 1) if result else 1
+
+    @classmethod
+    async def find_by_id(cls, bot: Red, moderation_id: int, guild_id: int, cursor: Cursor | None = None) -> "Moderation":
+        query = f"SELECT * FROM moderation_{guild_id} WHERE moderation_id = ?;"
+        case = await cls.execute(bot=bot, guild_id=guild_id, query=query, parameters=(moderation_id,), cursor=cursor)
+        if case:
+            return case[0]
+        raise ValueError(f"Case {moderation_id} not found in moderation_{guild_id}!")
+
+    @classmethod
+    async def find_by_target(cls, bot: Red, guild_id: int, target: int, before: datetime = None, after: datetime = None, types: Iterable[Type] | None = None, expired: bool | None = None, cursor: Cursor | None = None) -> Tuple["Moderation"]:
+        query = f"SELECT * FROM moderation_{guild_id} WHERE target_id = ?"
+        params = [target]
+        if types:
+            query += f" AND moderation_type IN ({', '.join(['?' for _ in types])})"
+            for t in types:
+                params.append(t.key)
+        if before:
+            query += " AND timestamp < ?"
+            params.append(int(before.timestamp()))
+        if after:
+            query += " AND timestamp > ?"
+            params.append(int(after.timestamp()))
+        if expired is not None:
+            query += " AND expired = ?"
+            params.append(int(expired))
+
+        query += " ORDER BY moderation_id DESC;"
+
+        return await cls.execute(bot=bot, guild_id=guild_id, query=query, parameters=params, cursor=cursor)
+
+    @classmethod
+    async def find_by_moderator(cls, bot: Red, guild_id: int, moderator: int, before: datetime = None, after: datetime = None, types: Iterable[Type] | None = None, expired: bool | None = None, cursor: Cursor | None = None) -> Tuple["Moderation"]:
+        query = f"SELECT * FROM moderation_{guild_id} WHERE moderator_id = ?"
+        params = [moderator]
+        if types:
+            query += f" AND moderation_type IN ({', '.join(['?' for _ in types])})"
+            for t in types:
+                params.append(t.key)
+        if before:
+            query += " AND timestamp < ?"
+            params.append(int(before.timestamp()))
+        if after:
+            query += " AND timestamp > ?"
+            params.append(int(after.timestamp()))
+        if expired is not None:
+            query += " AND expired = ?"
+            params.append(int(expired))
+
+        query += " ORDER BY moderation_id DESC;"
+
+        return await cls.execute(bot=bot, guild_id=guild_id, query=query, parameters=params, cursor=cursor)
+
+    @classmethod
+    async def log(
+        cls,
+        bot: Red,
+        guild_id: int,
+        moderator_id: int,
+        moderation_type: Type,
+        target_type: str,
+        target_id: int,
+        role_id: int | None = None,
+        duration: timedelta | None = None,
+        reason: str | None = None,
+        database: sqlite3.Connection | None = None,
+        timestamp: datetime | None = None,
+        resolved: bool = False,
+        resolved_by: int | None = None,
+        resolved_reason: str | None = None,
+        expired: bool | None = None,
+        changes: list | None = None,
+        metadata: dict | None = None,
+        return_obj: bool = True,
+    ) -> Union["Moderation", int]:
+        """Logs a moderation case in the database.
+
+        Args:
+            bot (Red): The bot instance.
+            guild_id (int): The ID of the guild to log the case in.
+            moderator_id (int): The ID of the moderator who issued the case.
+            moderation_type (Type): The type of moderation case. See `aurora.models.moderation_types` for the built-in options.
+            target_type (str): The type of target. Should be either `user` or `channel`.
+            target_id (int): The ID of the target.
+            role_id (int): The ID of the role, if applicable.
+            duration (timedelta): The duration of the case, if applicable.
+            reason (str): The reason for the case.
+            database (sqlite3.Connection): The database connection to use to log the case. A connection will be automatically created if not provided.
+            timestamp (datetime): The timestamp of the case. Will be automatically generated if not provided.
+            resolved (bool): Whether the case is resolved.
+            resolved_by (int): The ID of the user who resolved the case.
+            resolved_reason (str): The reason the case was resolved.
+            expired (bool): Whether the case is expired.
+            changes (list): A list of changes to log. You usually shouldn't pass this, as it's automatically generated by the `/edit` and `/resolve` commands.
+            metadata (dict): A dictionary of metadata to store with the case.
+            return_obj (bool): Whether to return the case object. Defaults to `True`. If `False`, returns the case ID.
+
+        Returns:
+            Union[Moderation, int]: The `Moderation` object if `return_obj` is `True`, otherwise the case ID.
+        """
+        from ..utilities.json import dumps
+        if not timestamp:
+            timestamp = datetime.fromtimestamp(time())
+        elif not isinstance(timestamp, datetime):
+            timestamp = datetime.fromtimestamp(timestamp)
+
+        if duration == "NULL":
+            duration = None
+
+        if duration is not None:
+            end_timestamp = timestamp + duration
+        else:
+            duration = None
+            end_timestamp = None
+
+        if not expired:
+            if end_timestamp:
+                expired = bool(timestamp > end_timestamp)
+            else:
+                expired = False
+
+        if reason == "NULL":
+            reason = None
+
+        if resolved_by in ["NULL", "?"]:
+            resolved_by = None
+
+        if resolved_reason == "NULL":
+            resolved_reason = None
+
+        if role_id == 0:
+            role_id = None
+
+        if not database:
+            database = await cls.connect()
+            close_db = True
+        else:
+            close_db = False
+
+        moderation_id = await cls.get_next_case_number(bot=bot, guild_id=guild_id)
+
+        case = {
+            "moderation_id": moderation_id,
+            "timestamp": timestamp.timestamp(),
+            "moderation_type": moderation_type.key,
+            "target_type": target_type,
+            "target_id": target_id,
+            "moderator_id": moderator_id,
+            "role_id": role_id,
+            "duration": timedelta_to_string(duration) if duration else None,
+            "end_timestamp": end_timestamp.timestamp() if end_timestamp else None,
+            "reason": reason,
+            "resolved": resolved,
+            "resolved_by": resolved_by,
+            "resolve_reason": resolved_reason,
+            "expired": expired,
+            "changes": dumps(changes),
+            "metadata": dumps(metadata)
+        }
+
+        sql = f"INSERT INTO `moderation_{guild_id}` (moderation_id, timestamp, moderation_type, target_type, target_id, moderator_id, role_id, duration, end_timestamp, reason, resolved, resolved_by, resolve_reason, expired, changes, metadata) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
+        await database.execute(sql, tuple(case.values()))
+
+        await database.commit()
+        if close_db:
+            await database.close()
+
+        logger.verbose(
+            "Row inserted into moderation_%s!\n%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s",
+            guild_id,
+            case["moderation_id"],
+            case["timestamp"],
+            case["moderation_type"],
+            case["target_type"],
+            case["target_id"],
+            case["moderator_id"],
+            case["role_id"],
+            case["duration"],
+            case["end_timestamp"],
+            case["reason"],
+            case["resolved"],
+            case["resolved_by"],
+            case["resolve_reason"],
+            case["expired"],
+            case["changes"],
+            case["metadata"],
+        )
+
+        if return_obj:
+            return await cls.find_by_id(bot=bot, moderation_id=moderation_id, guild_id=guild_id)
+        return moderation_id

aurora/models/partials.py

@@ -0,0 +1,89 @@
+from discord import ChannelType, Forbidden, Guild, HTTPException, InvalidData, NotFound, Role, User
+from discord.abc import Messageable
+from redbot.core.bot import Red
+
+from .base import AuroraBaseModel, AuroraGuildModel
+
+
+class PartialUser(AuroraBaseModel):
+    id: int
+    username: str
+    discriminator: int
+    _obj: User | None
+
+    @property
+    def name(self):
+        return f"{self.username}#{self.discriminator}" if self.discriminator != 0 else self.username
+
+    def __str__(self):
+        return self.name
+
+    def __repr__(self):
+        return f"<{self.__class__.__name__} id={self.id}>"
+
+    @classmethod
+    async def from_id(cls, bot: Red, user_id: int) -> "PartialUser":
+        user = bot.get_user(user_id)
+        if not user:
+            try:
+                user = await bot.fetch_user(user_id)
+                return cls(bot=bot, id=user.id, username=user.name, discriminator=user.discriminator, _obj=user)
+            except NotFound:
+                return cls(bot=bot, id=user_id, username="Deleted User", discriminator=0, _obj=None)
+        return cls(bot=bot, id=user.id, username=user.name, discriminator=user.discriminator, _obj=user)
+
+
+class PartialChannel(AuroraGuildModel):
+    id: int
+    name: str
+    type: ChannelType
+    _obj: Messageable | None
+
+    @property
+    def mention(self):
+        if self.name in ["Deleted Channel", "Forbidden Channel"]:
+            return self.name
+        return f"<#{self.id}>"
+
+    def __str__(self):
+        return self.mention
+
+    def __repr__(self):
+        return f"<{self.__class__.__name__} id={self.id} guild_id={self.guild_id}>"
+
+    @classmethod
+    async def from_id(cls, bot: Red, channel_id: int, guild: Guild) -> "PartialChannel":
+        channel = bot.get_channel(channel_id)
+        if not channel:
+            try:
+                channel = await bot.fetch_channel(channel_id)
+                return cls(bot=bot, guild_id=channel.guild.id, guild=guild, id=channel.id, name=channel.name, type=channel.type, _obj=channel)
+            except (NotFound, InvalidData, HTTPException, Forbidden) as e:
+                if e == Forbidden:
+                    return cls(bot=bot, guild_id=0, id=channel_id, name="Forbidden Channel")
+                return cls(bot=bot, guild_id=0, id=channel_id, name="Deleted Channel", type=ChannelType.text, _obj=None)
+        return cls(bot=bot, guild_id=channel.guild.id, guild=guild, id=channel.id, name=channel.name, type=channel.type, _obj=channel)
+
+class PartialRole(AuroraGuildModel):
+    id: int
+    name: str
+    _obj: Role | None
+
+    @property
+    def mention(self):
+        if self.name in ["Deleted Role", "Forbidden Role"]:
+            return self.name
+        return f"<@&{self.id}>"
+
+    def __str__(self):
+        return self.mention
+
+    def __repr__(self) -> str:
+        return f"<{self.__class__.__name__} id={self.id} guild_id={self.guild_id}>"
+
+    @classmethod
+    async def from_id(cls, bot: Red, guild: Guild, role_id: int) -> "PartialRole":
+        role = guild.get_role(role_id)
+        if not role:
+            return cls(bot=bot, guild_id=guild.id, id=role_id, name="Deleted Role", _obj=None)
+        return cls(bot=bot, guild_id=guild.id, id=role.id, name=role.name, _obj=role)

aurora/models/type.py

@@ -0,0 +1,97 @@
+from abc import ABC, abstractmethod
+from typing import Any, Dict, Tuple
+
+from class_registry import ClassRegistry
+from class_registry.base import AutoRegister
+from discord import Interaction, Member, User
+from discord.abc import Messageable
+from redbot.core import commands
+
+type_registry: Dict['str', 'Type'] = ClassRegistry(attr_name='key', unique=True)
+
+class Type(AutoRegister(type_registry), ABC):
+    """This is a base class for moderation types.
+
+    Attributes:
+        key (str): The key to use for this type. This should be unique, as this is how the type is registered internally. Changing this key will break existing cases with this type.
+        string (str): The string to display for this type.
+        verb (str): The verb to use for this type.
+        embed_desc (str): The string to use for embed descriptions.
+        channel (bool): Whether this type targets channels or users. If this is `true` in a subclass, its overridden handler methods should be typed with `discord.abc.Messageable` instead of `discord.Member | discord.User`.
+        removes_from_guild (bool): Whether this type's handler removes the target from the guild, or if the moderation is expected to occur whenever the user is not in the guild. This does not actually remove the target from the guild, the handler method is responsible for that.
+
+    Properties:
+        name (str): The string to display for this type. This is the same as the `string` attribute.
+    """
+
+    key = "type"
+    string = "type"
+    verb = "typed"
+    embed_desc = "been "
+    channel = False
+    removes_from_guild = False
+
+    @abstractmethod
+    def void(self) -> Any:
+        """This method should be overridden by any child classes. This is a placeholder to allow for automatic class registration."""
+        raise NotImplementedError
+
+    @property
+    def name(self) -> str:
+        """Alias for the `string` attribute."""
+        return self.string
+
+    def __str__(self) -> str:
+        return self.string
+
+    def __repr__(self) -> str:
+        attrs = [
+            ('key', self.key),
+            ('channel', self.channel),
+        ]
+        joined = ' '.join(f'{key}={value!r}' for key, value in attrs)
+        return f"<{self.__class__.__name__} {joined}>"
+
+    @classmethod
+    async def handler(cls, ctx: commands.Context, target: Member | User | Messageable, silent: bool, **kwargs) -> 'Type': # pylint: disable=unused-argument
+        """This method should be overridden by any child classes, but should retain the same starting keyword arguments.
+
+        Arguments:
+            ctx (commands.Context): The context of the command.
+            target (discord.Member | discord.User | discord.abc.Messageable): The target of the moderation.
+            silent (bool): Whether details about the moderation should be DM'ed to the target of the moderation.
+        """
+        raise NotImplementedError
+
+    @classmethod
+    async def resolve_handler(cls, moderation, reason: str) -> Tuple[bool, str]: # pylint: disable=unused-argument
+        """This method should be overridden by any resolvable child classes, but should retain the same keyword arguments.
+        If your moderation type should not be resolvable, do not override this.
+
+        Arguments:
+            moderation (aurora.models.Moderation): The moderation to resolve.
+            reason (str): The reason for resolving the moderation.
+        """
+        raise NotImplementedError
+
+    @classmethod
+    async def expiry_handler(cls, moderation) -> int: # pylint: disable=unused-argument
+        """This method should be overridden by any expirable child classes, but should retain the same keyword arguments and return an integer.
+        If your moderation type should not expire, do not override this, but also do not set an `end_timestamp` when you log your moderation.
+
+        Arguments:
+            moderation (aurora.models.Moderation): The moderation that is expiring.
+        """
+        raise NotImplementedError
+
+    @classmethod
+    async def duration_edit_handler(cls, interaction: Interaction, old_moderation, new_moderation) -> bool: # pylint: disable=unused-argument
+        """This method should be overridden by any child classes with editable durations, but should retain the same keyword arguments and should return True if the duration was successfully modified, or False if it was not.
+        If your moderation type's duration should not be editable, do not override this.
+
+        Arguments:
+            interaction (discord.Interaction): The interaction that triggered the duration edit.
+            old_moderation (aurora.models.Moderation): The old moderation, from before the `/edit` command was invoked.
+            new_moderation (aurora.models.Moderation): The current state of the moderation.
+        """
+        raise NotImplementedError

aurora/utilities/config.py

@@ -27,3 +27,13 @@ def register_config(config_obj: Config):
         history_inline_pagesize=None,
         auto_evidenceformat=None,
     )
+
+    moderation_type = {
+        "show_in_history": True,
+        "show_moderator": None,
+        "use_discord_permissions": None,
+        "dm_users": None,
+    }
+
+    config_obj.init_custom("types", 2)
+    config_obj.register_custom("types", **moderation_type)

aurora/utilities/database.py

@@ -1,219 +0,0 @@
-# pylint: disable=cyclic-import
-import json
-import sqlite3
-import time
-from datetime import datetime, timedelta
-
-from discord import Guild
-from redbot.core import data_manager
-
-from .logger import logger
-from .utils import (convert_timedelta_to_str, generate_dict,
-                    get_next_case_number)
-
-
-def connect() -> sqlite3.Connection:
-    """Connects to the SQLite database, and returns a connection object."""
-    try:
-        connection = sqlite3.connect(
-            database=data_manager.cog_data_path(raw_name="Aurora") / "aurora.db"
-        )
-        return connection
-
-    except sqlite3.OperationalError as e:
-        logger.error("Unable to access the SQLite database!\nError:\n%s", e.msg)
-        raise ConnectionRefusedError(
-            f"Unable to access the SQLite Database!\n{e.msg}"
-        ) from e
-
-
-async def create_guild_table(guild: Guild):
-    database = connect()
-    cursor = database.cursor()
-
-    try:
-        cursor.execute(f"SELECT * FROM `moderation_{guild.id}`")
-        logger.debug("SQLite Table exists for server %s (%s)", guild.name, guild.id)
-
-    except sqlite3.OperationalError:
-        query = f"""
-            CREATE TABLE `moderation_{guild.id}` (
-                moderation_id INTEGER PRIMARY KEY,
-                timestamp INTEGER NOT NULL,
-                moderation_type TEXT NOT NULL,
-                target_type TEXT NOT NULL,
-                target_id TEXT NOT NULL,
-                moderator_id TEXT NOT NULL,
-                role_id TEXT,
-                duration TEXT,
-                end_timestamp INTEGER,
-                reason TEXT,
-                resolved INTEGER NOT NULL,
-                resolved_by TEXT,
-                resolve_reason TEXT,
-                expired INTEGER NOT NULL,
-                changes TEXT NOT NULL,
-                metadata TEXT NOT NULL
-            )
-        """
-        cursor.execute(query)
-
-        index_query_1 = f"CREATE INDEX IF NOT EXISTS idx_target_id ON moderation_{guild.id}(target_id);"
-        cursor.execute(index_query_1)
-
-        index_query_2 = f"CREATE INDEX IF NOT EXISTS idx_moderator_id ON moderation_{guild.id}(moderator_id);"
-        cursor.execute(index_query_2)
-
-        index_query_3 = f"CREATE INDEX IF NOT EXISTS idx_moderation_id ON moderation_{guild.id}(moderation_id);"
-        cursor.execute(index_query_3)
-
-        insert_query = f"""
-            INSERT INTO `moderation_{guild.id}`
-            (moderation_id, timestamp, moderation_type, target_type, target_id, moderator_id, role_id, duration, end_timestamp, reason, resolved, resolved_by, resolve_reason, expired, changes, metadata)
-            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
-        """
-        insert_values = (
-            0,
-            0,
-            "NULL",
-            "NULL",
-            0,
-            0,
-            0,
-            "NULL",
-            0,
-            "NULL",
-            0,
-            "NULL",
-            "NULL",
-            0,
-            json.dumps([]),
-            json.dumps({}),
-        )
-        cursor.execute(insert_query, insert_values)
-
-        database.commit()
-
-        logger.debug(
-            "SQLite Table (moderation_%s) created for %s (%s)",
-            guild.id,
-            guild.name,
-            guild.id,
-        )
-
-    database.close()
-
-
-async def mysql_log(
-    guild_id: str,
-    author_id: str,
-    moderation_type: str,
-    target_type: str,
-    target_id: int,
-    role_id: int,
-    duration: timedelta,
-    reason: str,
-    database: sqlite3.Connection = None,
-    timestamp: int = None,
-    resolved: bool = False,
-    resolved_by: str = None,
-    resolved_reason: str = None,
-    expired: bool = None,
-    changes: list = None,
-    metadata: dict = None,
-) -> int:
-    if not timestamp:
-        timestamp = int(time.time())
-
-    if duration != "NULL":
-        end_timedelta = datetime.fromtimestamp(timestamp) + duration
-        end_timestamp = int(end_timedelta.timestamp())
-
-        duration = convert_timedelta_to_str(duration)
-    else:
-        end_timestamp = 0
-
-    if not expired:
-        if int(time.time()) > end_timestamp:
-            expired = 1
-        else:
-            expired = 0
-
-    if resolved_by is None:
-        resolved_by = "NULL"
-
-    if resolved_reason is None:
-        resolved_reason = "NULL"
-
-    if not database:
-        database = connect()
-        close_db = True
-    else:
-        close_db = False
-    cursor = database.cursor()
-
-    moderation_id = await get_next_case_number(guild_id=guild_id, cursor=cursor)
-
-    sql = f"INSERT INTO `moderation_{guild_id}` (moderation_id, timestamp, moderation_type, target_type, target_id, moderator_id, role_id, duration, end_timestamp, reason, resolved, resolved_by, resolve_reason, expired, changes, metadata) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
-    val = (
-        moderation_id,
-        timestamp,
-        moderation_type,
-        target_type,
-        target_id,
-        author_id,
-        role_id,
-        duration,
-        end_timestamp,
-        reason,
-        int(resolved),
-        resolved_by,
-        resolved_reason,
-        expired,
-        json.dumps(changes if changes else []),
-        json.dumps(metadata if metadata else {}),
-    )
-    cursor.execute(sql, val)
-
-    cursor.close()
-    database.commit()
-    if close_db:
-        database.close()
-
-    logger.debug(
-        "Row inserted into moderation_%s!\n%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s",
-        guild_id,
-        moderation_id,
-        timestamp,
-        moderation_type,
-        target_type,
-        target_id,
-        author_id,
-        role_id,
-        duration,
-        end_timestamp,
-        reason,
-        int(resolved),
-        resolved_by,
-        resolved_reason,
-        expired,
-        changes,
-        metadata,
-    )
-
-    return moderation_id
-
-
-async def fetch_case(moderation_id: int, guild_id: str) -> dict:
-    """This method fetches a case from the database and returns the case's dictionary."""
-    database = connect()
-    cursor = database.cursor()
-
-    query = f"SELECT * FROM moderation_{guild_id} WHERE moderation_id = ?;"
-    cursor.execute(query, (moderation_id,))
-    result = cursor.fetchone()
-
-    cursor.close()
-    database.close()
-
-    return generate_dict(result)

aurora/utilities/factory.py

@@ -2,78 +2,68 @@
 from datetime import datetime, timedelta
 from typing import Union
 
-from discord import Color, Embed, Guild, Interaction, InteractionMessage, Member, Role, User
+from discord import Color, Embed, Guild, Interaction, Member, Message, Role, User
 from redbot.core import commands
-from redbot.core.utils.chat_formatting import bold, box, error, humanize_timedelta, warning
+from redbot.core.bot import Red
+from redbot.core.utils.chat_formatting import bold, box, error, humanize_timedelta, inline, warning
 
-from aurora.utilities.config import config
-from aurora.utilities.utils import fetch_channel_dict, fetch_user_dict, get_bool_emoji, get_next_case_number, get_pagesize_str
+from ..models.moderation import Moderation
+from ..models.partials import PartialUser
+from ..models.type import Type
+from .config import config
+from .utils import get_bool_emoji, get_pagesize_str
 
 
 async def message_factory(
+    bot: Red,
     color: Color,
     guild: Guild,
     reason: str,
-    moderation_type: str,
-    moderator: Union[Member, User] = None,
-    duration: timedelta = None,
-    response: InteractionMessage = None,
-    role: Role = None,
+    moderation_type: Type,
+    moderator: Union[Member, User] | None = None,
+    duration: timedelta | None = None,
+    response: Message | None = None,
+    case: bool = True,
 ) -> Embed:
     """This function creates a message from set parameters, meant for contacting the moderated user.
 
     Args:
+        bot (Red): The bot instance.
         color (Color): The color of the embed.
         guild (Guild): The guild the moderation occurred in.
         reason (str): The reason for the moderation.
-        moderation_type (str): The type of moderation.
+        moderation_type (Type): The type of moderation.
         moderator (Union[Member, User], optional): The moderator who performed the moderation. Defaults to None.
         duration (timedelta, optional): The duration of the moderation. Defaults to None.
-        response (InteractionMessage, optional): The response message. Defaults to None.
-        role (Role, optional): The role that was added or removed. Defaults to None.
+        response (Message, optional): The response message. Defaults to None.
+        case (bool, optional): Whether the message is for a moderation case. Defaults to True.
 
 
     Returns:
         embed: The message embed.
     """
-    if response is not None and moderation_type not in [
-        "kicked",
-        "banned",
-        "tempbanned",
-        "unbanned",
-    ]:
+    if response is not None and not moderation_type.removes_from_guild:
         guild_name = f"[{guild.name}]({response.jump_url})"
     else:
         guild_name = guild.name
 
-    title = moderation_type
-
-    if moderation_type in ["tempbanned", "muted"] and duration:
+    if duration:
         embed_duration = f" for {humanize_timedelta(timedelta=duration)}"
     else:
         embed_duration = ""
 
-    if moderation_type == "note":
-        embed_desc = "received a"
-    elif moderation_type == "addrole":
-        embed_desc = f"received the {role.name} role"
-        title = "Role Added"
-        moderation_type = ""
-    elif moderation_type == "removerole":
-        embed_desc = f"lost the {role.name} role"
-        title = "Role Removed"
-        moderation_type = ""
-    else:
-        embed_desc = "been"
-
     embed = Embed(
-        title=str.title(title),
-        description=f"You have {embed_desc} {moderation_type}{embed_duration} in {guild_name}.",
+        title=str.title(moderation_type.verb),
+        description=f"You have {moderation_type.embed_desc}{moderation_type.verb}{embed_duration} in {guild_name}.",
         color=color,
         timestamp=datetime.now(),
     )
 
-    if await config.guild(guild).show_moderator() and moderator is not None:
+    show_moderator = await config.custom("types", guild.id, moderation_type.key).show_moderator()
+    if show_moderator is None:
+        show_moderator = await config.guild(guild).show_moderator()
+
+    if show_moderator and moderator is not None:
         embed.add_field(
             name="Moderator", value=f"`{moderator.name} ({moderator.id})`", inline=False
         )
@@ -85,312 +75,243 @@ async def message_factory(
     else:
         embed.set_author(name=guild.name)
 
+    if case:
+        embed.set_footer(
+            text=f"Case #{await Moderation.get_next_case_number(bot=bot, guild_id=guild.id):,}",
+            icon_url="attachment://arrow.png",
+        )
+
+    return embed
+
+async def resolve_factory(moderation: Moderation, reason: str) -> Embed:
+    """This function creates a resolved embed from set parameters, meant for contacting the moderated user.
+
+    Args:
+        moderation (aurora.models.Moderation): The moderation object.
+        reason (str): The reason for resolving the moderation.
+    Returns: `discord.Embed`
+    """
+
+    embed = Embed(
+        title=str.title(moderation.type.name) + " Resolved",
+        description=f"Your {moderation.type.name} in {moderation.guild.name} has been resolved.",
+        color=await moderation.bot.get_embed_color(moderation.guild.channels[0]),
+        timestamp=datetime.now(),
+    )
+
+    embed.add_field(name="Reason", value=f"`{reason}`", inline=False)
+
+    if moderation.guild.icon.url is not None:
+        embed.set_author(name=moderation.guild.name, icon_url=moderation.guild.icon.url)
+    else:
+        embed.set_author(name=moderation.guild.name)
     embed.set_footer(
-        text=f"Case #{await get_next_case_number(guild.id):,}",
+        text=f"Case #{moderation.id:,}",
         icon_url="attachment://arrow.png",
     )
 
     return embed
 
-
 async def log_factory(
-    interaction: Interaction, case_dict: dict, resolved: bool = False
+    ctx: commands.Context, moderation: Moderation, resolved: bool = False
 ) -> Embed:
     """This function creates a log embed from set parameters, meant for moderation logging.
 
     Args:
-        interaction (Interaction): The interaction object.
-        case_dict (dict): The case dictionary.
+        ctx (commands.Context): The ctx object.
+        moderation (aurora.models.Moderation): The moderation object.
         resolved (bool, optional): Whether the case is resolved or not. Defaults to False.
     """
+    target = await moderation.get_target()
+    moderator = await moderation.get_moderator()
     if resolved:
-        if case_dict["target_type"] == "USER":
-            target_user = await fetch_user_dict(interaction.client, case_dict["target_id"])
-            target_name = (
-                f"`{target_user['name']}`"
-                if target_user["discriminator"] == "0"
-                else f"`{target_user['name']}#{target_user['discriminator']}`"
-            )
-        elif case_dict["target_type"] == "CHANNEL":
-            target_user = await fetch_channel_dict(interaction.guild, case_dict["target_id"])
-            if target_user["mention"]:
-                target_name = f"{target_user['mention']}"
-            else:
-                target_name = f"`{target_user['name']}`"
-
-        moderator_user = await fetch_user_dict(interaction.client, case_dict["moderator_id"])
-        moderator_name = (
-            f"`{moderator_user['name']}`"
-            if moderator_user["discriminator"] == "0"
-            else f"`{moderator_user['name']}#{moderator_user['discriminator']}`"
-        )
-
         embed = Embed(
-            title=f"📕 Case #{case_dict['moderation_id']:,} Resolved",
-            color=await interaction.client.get_embed_color(interaction.channel),
+            title=f"📕 Case #{moderation.id:,} Resolved",
+            color=await ctx.bot.get_embed_color(ctx.channel),
         )
 
-        embed.description = f"**Type:** {str.title(case_dict['moderation_type'])}\n**Target:** {target_name} ({target_user['id']})\n**Moderator:** {moderator_name} ({moderator_user['id']})\n**Timestamp:** <t:{case_dict['timestamp']}> | <t:{case_dict['timestamp']}:R>"
+        resolved_by = await moderation.get_resolved_by()
+        embed.description = f"**Type:** {str.title(moderation.type.string)}\n**Target:** {target.name} ({target.id})\n**Moderator:** {moderator.name} ({moderator.id})\n**Timestamp:** <t:{moderation.unix_timestamp}> | <t:{moderation.unix_timestamp}:R>"
 
-        if case_dict["duration"] != "NULL":
-            td = timedelta(
-                **{
-                    unit: int(val)
-                    for unit, val in zip(
-                        ["hours", "minutes", "seconds"],
-                        case_dict["duration"].split(":"),
-                    )
-                }
-            )
+        if moderation.duration is not None:
             duration_embed = (
-                f"{humanize_timedelta(timedelta=td)} | <t:{case_dict['end_timestamp']}:R>"
-                if case_dict["expired"] == "0"
-                else str(humanize_timedelta(timedelta=td))
+                f"{humanize_timedelta(timedelta=moderation.duration)} | <t:{moderation.end_timestamp}:R>"
+                if not moderation.expired
+                else str(humanize_timedelta(timedelta=moderation.duration))
             )
             embed.description = (
                 embed.description
-                + f"\n**Duration:** {duration_embed}\n**Expired:** {bool(case_dict['expired'])}"
+                + f"\n**Duration:** {duration_embed}\n**Expired:** {moderation.expired}"
             )
 
-        embed.add_field(name="Reason", value=box(case_dict["reason"]), inline=False)
+        if moderation.metadata.items():
+            for key, value in moderation.metadata.items():
+                embed.description += f"\n**{key.title()}:** {value}"
+
+        embed.add_field(name="Reason", value=box(moderation.reason), inline=False)
 
-        resolved_user = await fetch_user_dict(interaction.client, case_dict["resolved_by"])
-        resolved_name = (
-            resolved_user["name"]
-            if resolved_user["discriminator"] == "0"
-            else f"{resolved_user['name']}#{resolved_user['discriminator']}"
-        )
         embed.add_field(
             name="Resolve Reason",
-            value=f"Resolved by `{resolved_name}` ({resolved_user['id']}) for:\n"
-            + box(case_dict["resolve_reason"]),
+            value=f"Resolved by `{resolved_by.name}` ({resolved_by.id}) for:\n"
+            + box(moderation.resolve_reason),
             inline=False,
         )
     else:
-        if case_dict["target_type"] == "USER":
-            target_user = await fetch_user_dict(interaction.client, case_dict["target_id"])
-            target_name = (
-                f"`{target_user['name']}`"
-                if target_user["discriminator"] == "0"
-                else f"`{target_user['name']}#{target_user['discriminator']}`"
-            )
-        elif case_dict["target_type"] == "CHANNEL":
-            target_user = await fetch_channel_dict(interaction.guild, case_dict["target_id"])
-            if target_user["mention"]:
-                target_name = target_user["mention"]
-            else:
-                target_name = f"`{target_user['name']}`"
-
-        moderator_user = await fetch_user_dict(interaction.client, case_dict["moderator_id"])
-        moderator_name = (
-            f"`{moderator_user['name']}`"
-            if moderator_user["discriminator"] == "0"
-            else f"`{moderator_user['name']}#{moderator_user['discriminator']}`"
-        )
-
         embed = Embed(
-            title=f"📕 Case #{case_dict['moderation_id']:,}",
-            color=await interaction.client.get_embed_color(interaction.channel),
+            title=f"📕 Case #{moderation.id:,}",
+            color=await ctx.bot.get_embed_color(ctx.channel),
         )
-        embed.description = f"**Type:** {str.title(case_dict['moderation_type'])}\n**Target:** {target_name} ({target_user['id']})\n**Moderator:** {moderator_name} ({moderator_user['id']})\n**Timestamp:** <t:{case_dict['timestamp']}> | <t:{case_dict['timestamp']}:R>"
+        embed.description = f"**Type:** {str.title(moderation.type.string)}\n**Target:** {target.name} ({target.id})\n**Moderator:** {moderator.name} ({moderator.id})\n**Timestamp:** <t:{moderation.unix_timestamp}> | <t:{moderation.unix_timestamp}:R>"
 
-        if case_dict["duration"] != "NULL":
-            td = timedelta(
-                **{
-                    unit: int(val)
-                    for unit, val in zip(
-                        ["hours", "minutes", "seconds"],
-                        case_dict["duration"].split(":"),
-                    )
-                }
-            )
+        if moderation.duration:
             embed.description = (
                 embed.description
-                + f"\n**Duration:** {humanize_timedelta(timedelta=td)} | <t:{case_dict['end_timestamp']}:R>"
+                + f"\n**Duration:** {humanize_timedelta(timedelta=moderation.duration)} | <t:{moderation.unix_timestamp}:R>"
             )
 
-        embed.add_field(name="Reason", value=box(case_dict["reason"]), inline=False)
+        if moderation.metadata.items():
+            for key, value in moderation.metadata.items():
+                embed.description += f"\n**{key.title()}:** {value}"
+
+        embed.add_field(name="Reason", value=box(moderation.reason), inline=False)
     return embed
 
 
-async def case_factory(interaction: Interaction, case_dict: dict) -> Embed:
+async def case_factory(interaction: Interaction, moderation: Moderation) -> Embed:
     """This function creates a case embed from set parameters.
 
     Args:
-        interaction (Interaction): The interaction object.
-        case_dict (dict): The case dictionary.
+        interaction (discord.Interaction): The interaction object.
+        moderation (aurora.models.Moderation): The moderation object.
     """
-    if case_dict["target_type"] == "USER":
-        target_user = await fetch_user_dict(interaction.client, case_dict["target_id"])
-        target_name = (
-            f"`{target_user['name']}`"
-            if target_user["discriminator"] == "0"
-            else f"`{target_user['name']}#{target_user['discriminator']}`"
-        )
-    elif case_dict["target_type"] == "CHANNEL":
-        target_user = await fetch_channel_dict(interaction.guild, case_dict["target_id"])
-        if target_user["mention"]:
-            target_name = f"{target_user['mention']}"
-        else:
-            target_name = f"`{target_user['name']}`"
-
-    moderator_user = await fetch_user_dict(interaction.client, case_dict["moderator_id"])
-    moderator_name = (
-        f"`{moderator_user['name']}`"
-        if moderator_user["discriminator"] == "0"
-        else f"`{moderator_user['name']}#{moderator_user['discriminator']}`"
-    )
+    target = await moderation.get_target()
+    moderator = await moderation.get_moderator()
 
     embed = Embed(
-        title=f"📕 Case #{case_dict['moderation_id']:,}",
+        title=f"📕 Case #{moderation.id:,}",
         color=await interaction.client.get_embed_color(interaction.channel),
     )
-    embed.description = f"**Type:** {str.title(case_dict['moderation_type'])}\n**Target:** {target_name} ({target_user['id']})\n**Moderator:** {moderator_name} ({moderator_user['id']})\n**Resolved:** {bool(case_dict['resolved'])}\n**Timestamp:** <t:{case_dict['timestamp']}> | <t:{case_dict['timestamp']}:R>"
+    embed.description = f"**Type:** {str.title(moderation.type.string)}\n**Target:** `{target.name}` ({target.id})\n**Moderator:** `{moderator.name}` ({moderator.id})\n**Resolved:** {moderation.resolved}\n**Timestamp:** <t:{moderation.unix_timestamp}> | <t:{moderation.unix_timestamp}:R>"
 
-    if case_dict["duration"] != "NULL":
-        td = timedelta(
-            **{
-                unit: int(val)
-                for unit, val in zip(
-                    ["hours", "minutes", "seconds"], case_dict["duration"].split(":")
-                )
-            }
-        )
+    if moderation.duration:
         duration_embed = (
-            f"{humanize_timedelta(timedelta=td)} | <t:{case_dict['end_timestamp']}:R>"
-            if bool(case_dict["expired"]) is False
-            else str(humanize_timedelta(timedelta=td))
+            f"{humanize_timedelta(timedelta=moderation.duration)} | <t:{moderation.unix_timestamp}:R>"
+            if moderation.expired is False
+            else str(humanize_timedelta(timedelta=moderation.duration))
         )
-        embed.description += f"\n**Duration:** {duration_embed}\n**Expired:** {bool(case_dict['expired'])}"
+        embed.description += f"\n**Duration:** {duration_embed}\n**Expired:** {moderation.expired}"
 
     embed.description += (
-        f"\n**Changes:** {len(case_dict['changes']) - 1}"
-        if case_dict["changes"]
+        f"\n**Changes:** {len(moderation.changes) - 1}"
+        if moderation.changes
         else "\n**Changes:** 0"
     )
 
-    if case_dict["role_id"]:
-        embed.description += f"\n**Role:** <@&{case_dict['role_id']}>"
+    if moderation.role_id:
+        role = await moderation.get_role()
+        embed.description += f"\n**Role:** {role.name}"
 
-    if case_dict["metadata"]:
-        if case_dict["metadata"]["imported_from"]:
+    if moderation.metadata:
+        if moderation.metadata.get("imported_from"):
             embed.description += (
-                f"\n**Imported From:** {case_dict['metadata']['imported_from']}"
+                f"\n**Imported From:** {moderation.metadata['imported_from']}"
             )
+            moderation.metadata.pop("imported_from")
+        if moderation.metadata.get("imported_timestamp"):
+            embed.description += (
+                f"\n**Imported Timestamp:** <t:{moderation.metadata['imported_timestamp']}> | <t:{moderation.metadata['imported_timestamp']}:R>"
+            )
+            moderation.metadata.pop("imported_timestamp")
+        if moderation.metadata.items():
+            for key, value in moderation.metadata.items():
+                embed.description += f"\n**{key.title()}:** {value}"
 
-    embed.add_field(name="Reason", value=box(case_dict["reason"]), inline=False)
+    embed.add_field(name="Reason", value=box(moderation.reason), inline=False)
 
-    if case_dict["resolved"] == 1:
-        resolved_user = await fetch_user_dict(interaction.client, case_dict["resolved_by"])
-        resolved_name = (
-            f"`{resolved_user['name']}`"
-            if resolved_user["discriminator"] == "0"
-            else f"`{resolved_user['name']}#{resolved_user['discriminator']}`"
-        )
+    if moderation.resolved:
+        resolved_user = await moderation.get_resolved_by()
+        if not resolved_user:
+            resolved_user = PartialUser(bot=interaction.client, id=0, username="Deleted User", discriminator="0")
         embed.add_field(
             name="Resolve Reason",
-            value=f"Resolved by {resolved_name} ({resolved_user['id']}) for:\n{box(case_dict['resolve_reason'])}",
+            value=f"Resolved by `{resolved_user.name or 'Deleted User'}` ({resolved_user.id or '0'}) for:\n{box(moderation.resolve_reason)}",
             inline=False,
         )
 
     return embed
 
 
-async def changes_factory(interaction: Interaction, case_dict: dict) -> Embed:
+async def changes_factory(interaction: Interaction, moderation: Moderation) -> Embed:
     """This function creates a changes embed from set parameters.
 
     Args:
-        interaction (Interaction): The interaction object.
-        case_dict (dict): The case dictionary.
+        interaction (discord.Interaction): The interaction object.
+        moderation (aurora.models.Moderation): The moderation object.
     """
     embed = Embed(
-        title=f"📕 Case #{case_dict['moderation_id']:,} Changes",
+        title=f"📕 Case #{moderation.id:,} Changes",
         color=await interaction.client.get_embed_color(interaction.channel),
     )
 
     memory_dict = {}
 
-    if case_dict["changes"]:
-        for change in case_dict["changes"]:
-            if change["user_id"] not in memory_dict:
-                memory_dict[str(change["user_id"])] = await fetch_user_dict(
-                    interaction.client, change["user_id"]
-                )
+    if moderation.changes:
+        for change in moderation.changes:
+            if change.user_id not in memory_dict:
+                memory_dict[str(change.user_id)] = await change.get_user()
 
-            user = memory_dict[str(change["user_id"])]
-            name = (
-                user["name"]
-                if user["discriminator"] == "0"
-                else f"{user['name']}#{user['discriminator']}"
+            user: PartialUser = memory_dict[str(change.user_id)]
+
+            timestamp = f"<t:{change.unix_timestamp}> | <t:{change.unix_timestamp}:R>"
+            end_timestamp = f"<t:{change.unix_end_timestamp}> | <t:{change.unix_end_timestamp}:R>" if change.end_timestamp else None
+
+            change_str = [
+                f"{bold('User:')} {inline(user.name)} ({user.id})",
+                f"{bold('Reason:')} {change.reason}" if change.reason else "",
+                f"{bold('Duration:')} {humanize_timedelta(timedelta=change.duration)}" if change.duration else "",
+                f"{bold('End Timestamp:')} {end_timestamp}" if end_timestamp else "",
+                f"{bold('Timestamp')} {timestamp}",
+            ]
+            copy = change_str.copy()
+            for string in change_str:
+                if string == "":
+                    copy.remove(string)
+
+            embed.add_field(
+                name=change.type.title(),
+                value="\n".join(copy),
+                inline=False,
             )
 
-            timestamp = f"<t:{change['timestamp']}> | <t:{change['timestamp']}:R>"
-
-            if change["type"] == "ORIGINAL":
-                embed.add_field(
-                    name="Original",
-                    value=f"**User:** `{name}` ({user['id']})\n**Reason:** {change['reason']}\n**Timestamp:** {timestamp}",
-                    inline=False,
-                )
-
-            elif change["type"] == "EDIT":
-                embed.add_field(
-                    name="Edit",
-                    value=f"**User:** `{name}` ({user['id']})\n**Reason:** {change['reason']}\n**Timestamp:** {timestamp}",
-                    inline=False,
-                )
-
-            elif change["type"] == "RESOLVE":
-                embed.add_field(
-                    name="Resolve",
-                    value=f"**User:** `{name}` ({user['id']})\n**Reason:** {change['reason']}\n**Timestamp:** {timestamp}",
-                    inline=False,
-                )
-
     else:
         embed.description = "*No changes have been made to this case.* 🙁"
 
     return embed
 
 
-async def evidenceformat_factory(interaction: Interaction, case_dict: dict) -> str:
+async def evidenceformat_factory(moderation: Moderation) -> str:
     """This function creates a codeblock in evidence format from set parameters.
 
     Args:
-        interaction (Interaction): The interaction object.
-        case_dict (dict): The case dictionary.
+        interaction (discord.Interaction): The interaction object.
+        moderation (aurora.models.Moderation): The moderation object.
     """
-    if case_dict["target_type"] == "USER":
-        target_user = await fetch_user_dict(interaction.client, case_dict["target_id"])
-        target_name = (
-            target_user["name"]
-            if target_user["discriminator"] == "0"
-            else f"{target_user['name']}#{target_user['discriminator']}"
-        )
+    target = await moderation.get_target()
+    moderator = await moderation.get_moderator()
 
-    elif case_dict["target_type"] == "CHANNEL":
-        target_user = await fetch_channel_dict(interaction.guild, case_dict["target_id"])
-        target_name = target_user["name"]
+    content = f"Case: {moderation.id:,} ({str.title(moderation.type.string)})\nTarget: {target.name} ({target.id})\nModerator: {moderator.name} ({moderator.id})"
 
-    moderator_user = await fetch_user_dict(interaction.client, case_dict["moderator_id"])
-    moderator_name = (
-        moderator_user["name"]
-        if moderator_user["discriminator"] == "0"
-        else f"{moderator_user['name']}#{moderator_user['discriminator']}"
-    )
+    if moderation.duration is not None:
+        content += f"\nDuration: {humanize_timedelta(timedelta=moderation.duration)}"
 
-    content = f"Case: {case_dict['moderation_id']:,} ({str.title(case_dict['moderation_type'])})\nTarget: {target_name} ({target_user['id']})\nModerator: {moderator_name} ({moderator_user['id']})"
+    if moderation.role_id:
+        role = await moderation.get_role()
+        content += "\nRole: " + (role.name if role is not None else moderation.role_id)
 
-    if case_dict["role_id"] != "0":
-        role = interaction.guild.get_role(int(case_dict["role_id"]))
-        content += "\nRole: " + (role.name if role is not None else case_dict["role_id"])
+    content += f"\nReason: {moderation.reason}"
 
-    if case_dict["duration"] != "NULL":
-        hours, minutes, seconds = map(int, case_dict["duration"].split(":"))
-        td = timedelta(hours=hours, minutes=minutes, seconds=seconds)
-        content += f"\nDuration: {humanize_timedelta(timedelta=td)}"
-
-    content += f"\nReason: {case_dict['reason']}"
+    for key, value in moderation.metadata.items():
+        content += f"\n{key.title()}: {value}"
 
     return box(content, "prolog")
 
@@ -622,3 +543,41 @@ async def immune_embed(ctx: commands.Context) -> Embed:
         e.description += "\n\n" + immune_str
 
     return e
+
+async def type_embed(ctx: commands.Context, moderation_type = Type) -> Embed:
+    """Generates a configuration menu field value for a guild's settings."""
+
+    type_settings = {
+        "show_in_history": await config.custom("types", ctx.guild.id, moderation_type.key).show_in_history(),
+        "show_moderator": await config.custom("types", ctx.guild.id, moderation_type.key).show_moderator(),
+        "use_discord_permissions": await config.custom("types", ctx.guild.id, moderation_type.key).use_discord_permissions(),
+        "dm_users": await config.custom("types", ctx.guild.id, moderation_type.key).dm_users(),
+    }
+
+    guild_str = [
+        "- "
+        + bold("Show in History: ")
+        + get_bool_emoji(type_settings["show_in_history"]),
+        "- "
+        + bold("Show Moderator: ")
+        + get_bool_emoji(type_settings["show_moderator"]),
+        "- "
+        + bold("Use Discord Permissions: ")
+        + get_bool_emoji(type_settings["use_discord_permissions"]),
+        "- "
+        + bold("DM Users: ")
+        + get_bool_emoji(type_settings["dm_users"]),
+    ]
+    guild_str = "\n".join(guild_str)
+
+    e = await _config(ctx)
+    e.title += f": {moderation_type.string.title()} Configuration"
+    e.description = (
+    f"""
+    Use the buttons below to manage Aurora's configuration for the {bold(moderation_type.string)} moderation type.
+    If an option has a question mark (\N{BLACK QUESTION MARK ORNAMENT}\N{VARIATION SELECTOR-16}) next to it, Aurora will default to the guild level setting instead.
+    See `{ctx.prefix}aurora set guild` for more information.\n
+    """
+        + guild_str
+    )
+    return e

aurora/utilities/json.py

@@ -0,0 +1,135 @@
+import json
+from datetime import datetime, timedelta
+from typing import Any
+
+from redbot.core.bot import Red
+
+from ..models.base import AuroraBaseModel
+from ..models.type import Type
+
+
+class JSONEncoder(json.JSONEncoder):
+    def default(self, o) -> Any:
+        match o:
+            case datetime():
+                return int(o.timestamp())
+            case timedelta():
+                from ..utilities.utils import timedelta_to_string
+                return timedelta_to_string(o)
+            case AuroraBaseModel():
+                return o.dump()
+            case Type():
+                return o.key
+            case Red():
+                return None
+            case _:
+                return super().default(o)
+
+
+# This is a wrapper around the json module's dumps function that uses our custom JSONEncoder class
+def dumps(obj, *, skipkeys=False, ensure_ascii=True, check_circular=True,
+        allow_nan=True, indent=None, separators=None,
+        default=None, sort_keys=False, **kw) -> str:
+    """Serialize ``obj`` to a JSON formatted ``str``.
+
+    If ``skipkeys`` is true then ``dict`` keys that are not basic types
+    (``str``, ``int``, ``float``, ``bool``, ``None``) will be skipped
+    instead of raising a ``TypeError``.
+
+    If ``ensure_ascii`` is false, then the return value can contain non-ASCII
+    characters if they appear in strings contained in ``obj``. Otherwise, all
+    such characters are escaped in JSON strings.
+
+    If ``check_circular`` is false, then the circular reference check
+    for container types will be skipped and a circular reference will
+    result in an ``RecursionError`` (or worse).
+
+    If ``allow_nan`` is false, then it will be a ``ValueError`` to
+    serialize out of range ``float`` values (``nan``, ``inf``, ``-inf``) in
+    strict compliance of the JSON specification, instead of using the
+    JavaScript equivalents (``NaN``, ``Infinity``, ``-Infinity``).
+
+    If ``indent`` is a non-negative integer, then JSON array elements and
+    object members will be pretty-printed with that indent level. An indent
+    level of 0 will only insert newlines. ``None`` is the most compact
+    representation.
+
+    If specified, ``separators`` should be an ``(item_separator, key_separator)``
+    tuple.  The default is ``(', ', ': ')`` if *indent* is ``None`` and
+    ``(',', ': ')`` otherwise.  To get the most compact JSON representation,
+    you should specify ``(',', ':')`` to eliminate whitespace.
+
+    ``default(obj)`` is a function that should return a serializable version
+    of obj or raise TypeError. The default simply raises TypeError.
+
+    If *sort_keys* is true (default: ``False``), then the output of
+    dictionaries will be sorted by key.
+    """
+    return json.dumps(
+        obj,
+        cls=JSONEncoder,
+        skipkeys=skipkeys,
+        ensure_ascii=ensure_ascii,
+        check_circular=check_circular,
+        allow_nan=allow_nan,
+        indent=indent,
+        separators=separators,
+        default=default,
+        sort_keys=sort_keys,
+        **kw
+    )
+
+# This is a wrapper around the json module's dump function that uses our custom JSONEncoder class
+def dump(obj, fp, *, skipkeys=False, ensure_ascii=True, check_circular=True,
+        allow_nan=True, indent=None, separators=None,
+        default=None, sort_keys=False, **kw) -> str:
+    """Serialize ``obj`` as a JSON formatted stream to ``fp`` (a
+    ``.write()``-supporting file-like object).
+
+    If ``skipkeys`` is true then ``dict`` keys that are not basic types
+    (``str``, ``int``, ``float``, ``bool``, ``None``) will be skipped
+    instead of raising a ``TypeError``.
+
+    If ``ensure_ascii`` is false, then the strings written to ``fp`` can
+    contain non-ASCII characters if they appear in strings contained in
+    ``obj``. Otherwise, all such characters are escaped in JSON strings.
+
+    If ``check_circular`` is false, then the circular reference check
+    for container types will be skipped and a circular reference will
+    result in an ``RecursionError`` (or worse).
+
+    If ``allow_nan`` is false, then it will be a ``ValueError`` to
+    serialize out of range ``float`` values (``nan``, ``inf``, ``-inf``)
+    in strict compliance of the JSON specification, instead of using the
+    JavaScript equivalents (``NaN``, ``Infinity``, ``-Infinity``).
+
+    If ``indent`` is a non-negative integer, then JSON array elements and
+    object members will be pretty-printed with that indent level. An indent
+    level of 0 will only insert newlines. ``None`` is the most compact
+    representation.
+
+    If specified, ``separators`` should be an ``(item_separator, key_separator)``
+    tuple.  The default is ``(', ', ': ')`` if *indent* is ``None`` and
+    ``(',', ': ')`` otherwise.  To get the most compact JSON representation,
+    you should specify ``(',', ':')`` to eliminate whitespace.
+
+    ``default(obj)`` is a function that should return a serializable version
+    of obj or raise TypeError. The default simply raises TypeError.
+
+    If *sort_keys* is true (default: ``False``), then the output of
+    dictionaries will be sorted by key.
+    """
+    return json.dump(
+        obj,
+        fp,
+        cls=JSONEncoder,
+        skipkeys=skipkeys,
+        ensure_ascii=ensure_ascii,
+        check_circular=check_circular,
+        allow_nan=allow_nan,
+        indent=indent,
+        separators=separators,
+        default=default,
+        sort_keys=sort_keys,
+        **kw
+    )

aurora/utilities/utils.py

@@ -1,23 +1,25 @@
 # pylint: disable=cyclic-import
-import json
-from datetime import datetime
-from datetime import timedelta as td
-from typing import Optional, Union
+from datetime import datetime, timedelta
+from typing import Optional, Tuple, Union
 
+import aiosqlite
 from dateutil.relativedelta import relativedelta as rd
-from discord import File, Guild, Interaction, Member, SelectOption, User
-from discord.errors import Forbidden, NotFound
+from discord import File, Guild, Interaction, Member, SelectOption, TextChannel, User
+from discord.errors import Forbidden
 from redbot.core import commands, data_manager
 from redbot.core.utils.chat_formatting import error
 
-from .config import config
+from ..models.type import Type
+from ..utilities.config import config
+from ..utilities.json import dumps
+from ..utilities.logger import logger
 
 
 def check_permissions(
     user: User,
-    permissions: list,
-    ctx: Union[commands.Context, Interaction] = None,
-    guild: Guild = None,
+    permissions: Tuple[str],
+    ctx: commands.Context | Interaction | None = None,
+    guild: Guild | None = None,
 ) -> Union[bool, str]:
     """Checks if a user has a specific permission (or a list of permissions) in a channel."""
     if ctx:
@@ -42,11 +44,17 @@ def check_permissions(
 
 
 async def check_moddable(
-    target: Union[User, Member], interaction: Interaction, permissions: list
+    target: Union[User, Member, TextChannel], ctx: commands.Context, permissions: Tuple[str], moderation_type: Type,
 ) -> bool:
     """Checks if a moderator can moderate a target."""
-    if check_permissions(interaction.client.user, permissions, guild=interaction.guild):
-        await interaction.response.send_message(
+    is_channel = isinstance(target, TextChannel)
+
+    use_discord_permissions = await config.custom("types", ctx.guild.id, moderation_type.key).use_discord_permissions()
+    if use_discord_permissions is None:
+        use_discord_permissions = await config.guild(ctx.guild).use_discord_permissions()
+
+    if check_permissions(ctx.bot.user, permissions, guild=ctx.guild):
+        await ctx.send(
             error(
                 f"I do not have the `{permissions}` permission, required for this action."
             ),
@@ -54,9 +62,9 @@ async def check_moddable(
         )
         return False
 
-    if await config.guild(interaction.guild).use_discord_permissions() is True:
-        if check_permissions(interaction.user, permissions, guild=interaction.guild):
-            await interaction.response.send_message(
+    if use_discord_permissions is True:
+        if check_permissions(ctx.author, permissions, guild=ctx.guild):
+            await ctx.send(
                 error(
                     f"You do not have the `{permissions}` permission, required for this action."
                 ),
@@ -64,21 +72,21 @@ async def check_moddable(
             )
             return False
 
-    if interaction.user.id == target.id:
-        await interaction.response.send_message(
+    if ctx.author.id == target.id:
+        await ctx.send(
             content="You cannot moderate yourself!", ephemeral=True
         )
         return False
 
-    if target.bot:
-        await interaction.response.send_message(
+    if not is_channel and target.bot:
+        await ctx.send(
             content="You cannot moderate bots!", ephemeral=True
         )
         return False
 
     if isinstance(target, Member):
-        if interaction.user.top_role <= target.top_role and await config.guild(interaction.guild).respect_hierarchy() is True:
-            await interaction.response.send_message(
+        if ctx.author.top_role <= target.top_role and await config.guild(ctx.guild).respect_hierarchy() is True:
+            await ctx.send(
                 content=error(
                     "You cannot moderate members with a higher role than you!"
                 ),
@@ -86,11 +94,17 @@ async def check_moddable(
             )
             return False
 
+        if target.guild_permissions.administrator:
+            await ctx.send(
+                content="You cannot moderate members with the Administrator permission!", ephemeral=True
+            )
+            return False
+
         if (
-            interaction.guild.get_member(interaction.client.user.id).top_role
+            ctx.guild.get_member(ctx.bot.user.id).top_role
             <= target.top_role
         ):
-            await interaction.response.send_message(
+            await ctx.send(
                 content=error(
                     "You cannot moderate members with a role higher than the bot!"
                 ),
@@ -102,7 +116,7 @@ async def check_moddable(
 
         for role in target.roles:
             if role.id in immune_roles:
-                await interaction.response.send_message(
+                await ctx.send(
                     content=error("You cannot moderate members with an immune role!"),
                     ephemeral=True,
                 )
@@ -111,152 +125,56 @@ async def check_moddable(
     return True
 
 
-async def get_next_case_number(guild_id: str, cursor=None) -> int:
-    """This function returns the next case number from the MySQL table for a specific guild."""
-    from .database import connect
-
-    if not cursor:
-        database = connect()
-        cursor = database.cursor()
-    cursor.execute(
-        f"SELECT moderation_id FROM `moderation_{guild_id}` ORDER BY moderation_id DESC LIMIT 1"
-    )
-    result = cursor.fetchone()
-    return (result[0] + 1) if result else 1
-
-
-def generate_dict(result) -> dict:
-    case = {
-        "moderation_id": result[0],
-        "timestamp": result[1],
-        "moderation_type": result[2],
-        "target_type": result[3],
-        "target_id": result[4],
-        "moderator_id": result[5],
-        "role_id": result[6],
-        "duration": result[7],
-        "end_timestamp": result[8],
-        "reason": result[9],
-        "resolved": result[10],
-        "resolved_by": result[11],
-        "resolve_reason": result[12],
-        "expired": result[13],
-        "changes": json.loads(result[14]),
-        "metadata": json.loads(result[15]),
-    }
-    return case
-
-
-async def fetch_user_dict(client: commands.Bot, user_id: str) -> dict:
-    """This function returns a dictionary containing either user information or a standard deleted user template."""
-    if user_id == "?":
-        user_dict = {"id": "?", "name": "Unknown User", "discriminator": "0"}
-
-    else:
-        try:
-            user = client.get_user(int(user_id))
-            if user is None:
-                user = await client.fetch_user(int(user_id))
-
-            user_dict = {
-                "id": user.id,
-                "name": user.name,
-                "discriminator": user.discriminator,
-            }
-
-        except NotFound:
-            user_dict = {
-                "id": user_id,
-                "name": "Deleted User",
-                "discriminator": "0",
-            }
-
-
-    return user_dict
-
-
-async def fetch_channel_dict(guild: Guild, channel_id: int) -> dict:
-    """This function returns a dictionary containing either channel information or a standard deleted channel template."""
-    try:
-        channel = guild.get_channel(int(channel_id))
-        if not channel:
-            channel = await guild.fetch_channel(channel_id)
-
-        channel_dict = {
-            "id": channel.id,
-            "name": channel.name,
-            "mention": channel.mention,
-        }
-
-    except NotFound:
-        channel_dict = {"id": channel_id, "name": "Deleted Channel", "mention": None}
-
-    return channel_dict
-
-
-async def fetch_role_dict(guild: Guild, role_id: int) -> dict:
-    """This function returns a dictionary containing either role information or a standard deleted role template."""
-    role = guild.get_role(int(role_id))
-    if not role:
-        role_dict = {"id": role_id, "name": "Deleted Role"}
-
-    role_dict = {"id": role.id, "name": role.name}
-
-    return role_dict
-
-
-async def log(interaction: Interaction, moderation_id: int, resolved: bool = False) -> None:
+async def log(ctx: commands.Context, moderation_id: int, resolved: bool = False) -> None:
     """This function sends a message to the guild's configured logging channel when an infraction takes place."""
-    from .database import fetch_case
+    from ..models.moderation import Moderation
     from .factory import log_factory
 
-    logging_channel_id = await config.guild(interaction.guild).log_channel()
+    logging_channel_id = await config.guild(ctx.guild).log_channel()
     if logging_channel_id != " ":
-        logging_channel = interaction.guild.get_channel(logging_channel_id)
+        logging_channel = ctx.guild.get_channel(logging_channel_id)
 
-        case = await fetch_case(moderation_id, interaction.guild.id)
-        if case:
+        try:
+            moderation = await Moderation.find_by_id(ctx.bot, moderation_id, ctx.guild.id)
             embed = await log_factory(
-                interaction=interaction, case_dict=case, resolved=resolved
+                ctx=ctx, moderation=moderation, resolved=resolved
             )
             try:
                 await logging_channel.send(embed=embed)
             except Forbidden:
                 return
+        except ValueError:
+            return
 
 
-async def send_evidenceformat(interaction: Interaction, case_dict: dict) -> None:
+async def send_evidenceformat(ctx: commands.Context, moderation_id: int) -> None:
     """This function sends an ephemeral message to the moderator who took the moderation action, with a pre-made codeblock for use in the mod-evidence channel."""
+    from ..models.moderation import Moderation
     from .factory import evidenceformat_factory
 
     send_evidence_bool = (
-        await config.user(interaction.user).auto_evidenceformat()
-        or await config.guild(interaction.guild).auto_evidenceformat()
+        await config.user(ctx.author).auto_evidenceformat()
+        or await config.guild(guild=ctx.guild).auto_evidenceformat()
         or False
     )
-    if send_evidence_bool is False:
-        return
-
-    content = await evidenceformat_factory(interaction=interaction, case_dict=case_dict)
-    await interaction.followup.send(content=content, ephemeral=True)
-
-
-def convert_timedelta_to_str(timedelta: td) -> str:
-    """This function converts a timedelta object to a string."""
-    total_seconds = int(timedelta.total_seconds())
-    hours = total_seconds // 3600
-    minutes = (total_seconds % 3600) // 60
-    seconds = total_seconds % 60
-    return f"{hours}:{minutes}:{seconds}"
+    if send_evidence_bool is True:
+        moderation = await Moderation.find_by_id(ctx.bot, moderation_id, ctx.guild.id)
+        content = await evidenceformat_factory(moderation=moderation)
+        if not ctx.interaction:
+            await ctx.author.send(content=content)
+        else:
+            await ctx.send(content=content, ephemeral=True)
 
 
 def get_bool_emoji(value: Optional[bool]) -> str:
     """Returns a unicode emoji based on a boolean value."""
-    if value is True:
-        return "\N{WHITE HEAVY CHECK MARK}"
-    if value is False:
-        return "\N{NO ENTRY SIGN}"
-    return "\N{BLACK QUESTION MARK ORNAMENT}\N{VARIATION SELECTOR-16}"
+    match value:
+        case True:
+            return "\N{WHITE HEAVY CHECK MARK}"
+        case False:
+            return "\N{NO ENTRY SIGN}"
+        case _:
+            return "\N{BLACK QUESTION MARK ORNAMENT}\N{VARIATION SELECTOR-16}"
 
 
 def get_pagesize_str(value: Union[int, None]) -> str:
@@ -286,13 +204,91 @@ def create_pagesize_options() -> list[SelectOption]:
         )
     return options
 
-def timedelta_from_relativedelta(relativedelta: rd) -> td:
+def timedelta_from_relativedelta(relativedelta: rd) -> timedelta:
     """Converts a relativedelta object to a timedelta object."""
     now = datetime.now()
     then = now - relativedelta
     return now - then
 
+def timedelta_from_string(string: str) -> timedelta:
+    """Converts a string to a timedelta object."""
+    hours, minutes, seconds = map(int, string.split(":"))
+    return timedelta(hours=hours, minutes=minutes, seconds=seconds)
+
+def timedelta_to_string(td: timedelta) -> str:
+    """Converts a timedelta object to a string."""
+    days = td.days * 24
+    hours, remainder = divmod(td.seconds, 3600)
+    minutes, seconds = divmod(remainder, 60)
+    return f"{days + hours}:{minutes:02}:{seconds:02}"
+
 def get_footer_image(coginstance: commands.Cog) -> File:
     """Returns the footer image for the embeds."""
     image_path = data_manager.bundled_data_path(coginstance) / "arrow.png"
     return File(image_path, filename="arrow.png", description="arrow")
+
+async def create_guild_table(guild: Guild) -> None:
+    from ..models.moderation import Moderation
+
+    try:
+        await Moderation.execute(f"SELECT * FROM `moderation_{guild.id}`", return_obj=False)
+        logger.trace("SQLite Table exists for server %s (%s)", guild.name, guild.id)
+
+    except aiosqlite.OperationalError:
+        query = f"""
+            CREATE TABLE `moderation_{guild.id}` (
+                moderation_id INTEGER PRIMARY KEY NOT NULL,
+                timestamp INTEGER NOT NULL,
+                moderation_type TEXT NOT NULL,
+                target_type TEXT NOT NULL,
+                target_id INTEGER NOT NULL,
+                moderator_id INTEGER NOT NULL,
+                role_id INTEGER,
+                duration TEXT,
+                end_timestamp INTEGER,
+                reason TEXT,
+                resolved INTEGER NOT NULL,
+                resolved_by TEXT,
+                resolve_reason TEXT,
+                expired INTEGER NOT NULL,
+                changes JSON NOT NULL,
+                metadata JSON NOT NULL
+            )
+        """
+        await Moderation.execute(query=query, return_obj=False)
+
+        index_query_1 = f"CREATE INDEX IF NOT EXISTS idx_target_id ON moderation_{guild.id}(target_id);"
+        await Moderation.execute(query=index_query_1, return_obj=False)
+
+        index_query_2 = f"CREATE INDEX IF NOT EXISTS idx_moderator_id ON moderation_{guild.id}(moderator_id);"
+        await Moderation.execute(query=index_query_2, return_obj=False)
+
+        index_query_3 = f"CREATE INDEX IF NOT EXISTS idx_moderation_id ON moderation_{guild.id}(moderation_id);"
+        await Moderation.execute(query=index_query_3, return_obj=False)
+
+        insert_query = f"""
+            INSERT INTO `moderation_{guild.id}`
+            (moderation_id, timestamp, moderation_type, target_type, target_id, moderator_id, role_id, duration, end_timestamp, reason, resolved, resolved_by, resolve_reason, expired, changes, metadata)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+        """
+        insert_values = (
+            0,
+            0,
+            "NULL",
+            "NULL",
+            0,
+            0,
+            None,
+            None,
+            None,
+            None,
+            0,
+            None,
+            None,
+            0,
+            dumps([]),
+            dumps({}),
+        )
+        await Moderation.execute(query=insert_query, parameters=insert_values, return_obj=False)
+
+        logger.trace("SQLite Table created for server %s (%s)", guild.name, guild.id)

pyproject.toml

@@ -9,14 +9,17 @@ package-mode = false
 
 [tool.poetry.dependencies]
 python = ">=3.11,<3.12"
-Red-DiscordBot = "^3.5.5"
+Red-DiscordBot = "^3.5.9"
 py-dactyl = "^2.0.4"
 websockets = "^12.0"
 pillow = "^10.3.0"
 numpy = "^1.26.4"
+pydantic = "^2.7.1"
 colorthief = "^0.2.1"
 beautifulsoup4 = "^4.12.3"
 markdownify = "^0.12.1"
+aiosqlite = "^0.20.0"
+phx-class-registry = "^5.0.0"
 
 [tool.poetry.group.dev]
 optional = true
@@ -24,6 +27,7 @@ optional = true
 [tool.poetry.group.dev.dependencies]
 ruff = "^0.3.1"
 pylint = "^3.1.0"
+pipx = "^1.5.0"
 sqlite-web = "^0.6.4"
 
 [tool.poetry.group.docs]