wlinator/tux
1
Fork 0
mirror of https://github.com/allthingslinux/tux.git synced 2024-10-02 16:43:12 +00:00
Code Issues Releases Wiki Activity

Refactor database models and clean up archive

Browse source 
Temp moving moderation and misc files to .archive for cleanup purposes.
Updated `schema.prisma` to use environment variables for production.
Removed redundant user, role, and emoji statistics database models and associated controllers.
Integrated a new `Case` model to consolidate infraction-related data.
Modified reminder, snippet, and note models, and their corresponding controllers to streamline and standardize the database structure.
Removed outdated database seeding commands.
This commit is contained in:
kzndotsh 2024-06-29 06:08:48 -04:00
parent 3137677cd8
commit 692855e6be
24 changed files with 529 additions and 1615 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

0
tux/cogs/moderation/ban.py → .archive/ban.py
View file

0
tux/cogs/misc/emojistats.py → .archive/emojistats.py
View file

0
tux/cogs/moderation/kick.py → .archive/kick.py
View file

0
tux/cogs/moderation/notes.py → .archive/notes.py
View file

0
tux/cogs/moderation/timeout.py → .archive/timeout.py
View file

0
tux/cogs/moderation/unban.py → .archive/unban.py
View file

0
tux/cogs/moderation/warn.py → .archive/warn.py
View file

227
prisma/schema.prisma
View file

@ -1,7 +1,7 @@
datasource db {
provider = "postgresql"
url = env("DATABASE_URL")
directUrl = env("DIRECT_URL")
url = env("PROD_DATABASE_URL")
directUrl = env("PROD_DIRECT_URL")
}
generator client {
@ -13,181 +13,78 @@ generator client {
recursive_type_depth = -1
}
// General settings for the system
model Settings {
// Key of the setting
key String @id
// Value of the setting
value String
// Optional description of what the setting does
description String?
enum CaseType {
BAN
HACK_BAN
TEMP_BAN
UNBAN
KICK
TIMEOUT_ADD
TIMEOUT_REMOVE
WARN
}
model Users {
// The user’s unique ID (via Discord)
id BigInt @id
// The user’s username.
name String
// The user’s global nickname, taking precedence over the username in display.
global_name String?
// Returns the user’s display name. For regular users this is just their global name or their username, but if they have a guild specific nickname then that is returned instead.
display_name String
// Returns a string that allows you to mention the given user.
mention String
// Specifies if the user is a bot account.
bot Boolean @default(false)
// Returns the user’s creation time in UTC. This is when the user’s Discord account was created.
created_at DateTime?
// True if user is a member of a guild (not a discord.py attribute)
is_member Boolean @default(true)
// The guild specific nickname of the user. Takes precedence over the global name.
nick String?
// An aware datetime object that specifies the date and time in UTC that the member joined the guild. If the member left and rejoined the guild, this will be the latest date. In certain cases, this can be None.
joined_at DateTime?
model Guild {
guild_id BigInt @id
guild_joined_at DateTime? @default(now())
cases Case[]
snippets Snippet[]
notes Note[]
reminders Reminder[]
// This is a relation field and is a list of roles that the user has, linking to the `UserRoles` table. If you fetch a user from the database and include this field, you will get all the roles associated with that user.
roles UserRoles[]
// This represents all the infractions that this user has given out when acting as a moderator. It has a `relation` annotation to make clear that for these infractions, this user is referred to in the `moderator` field of the `Infractions` table.
infractions_given Infractions[] @relation("Moderator")
// This is all the infractions that this user has received. It has a `relation` annotation to make clear that for these infractions, this user is referred to in the `user` field of the `Infractions` table.
infractions_received Infractions[] @relation("User")
snippets Snippets[]
afk Boolean @default(false)
afk_reason String?
afk_since DateTime?
notes_given Notes[] @relation("Moderator")
notes_received Notes[] @relation("User")
reminders Reminders[]
CommandStats CommandStats[]
@@unique([guild_id])
}
model Roles {
// The ID for the role (via Discord)
id BigInt @id
// The name of the role
name String
// Indicates if the role will be displayed separately from other members.
hoist Boolean @default(false)
// Indicates if the role is managed by the guild through some form of integrations such as Twitch.
managed Boolean @default(false)
// Indicates if the role is mentionable.
mentionable Boolean @default(false)
// The role’s creation time in UTC.
created_at DateTime?
// Returns a string that allows you to mention a role.
mention String? @default("")
// The role’s color. An integer representation of hexadecimal colour code.
color BigInt?
model Case {
case_id BigInt @id @default(autoincrement())
case_type CaseType
case_reason String
case_moderator_id BigInt
case_target_id BigInt
case_number Int?
case_created_at DateTime? @default(now())
case_expires_at DateTime?
guild_id BigInt
guild Guild @relation(fields: [guild_id], references: [guild_id])
// This field links a role to the users that have it. It references the `UserRoles` junction table. If you fetch a role from the database and include this field, you will get a list of UserRoles entries and from there you can find all the users that have this role.
users UserRoles[]
// This is a Boolean field indicating if the role is a moderator role. This is not an attribute coming from Discord but an extra field you have defined to distinguish normal roles from moderator roles. It defaults to false, meaning if you don't specify it when creating a new role, it will be assumed to be a non-moderator role.
is_mod Boolean @default(false)
@@unique([case_number, guild_id])
@@index([case_number, guild_id])
}
model UserRoles {
user Users @relation(fields: [user_id], references: [id])
user_id BigInt
model Snippet {
snippet_id BigInt @id @default(autoincrement())
snippet_name String
snippet_content String
snippet_user_id BigInt
snippet_created_at DateTime @default(now())
guild_id BigInt
guild Guild @relation(fields: [guild_id], references: [guild_id])
role Roles @relation(fields: [role_id], references: [id])
role_id BigInt
@@id([user_id, role_id])
@@unique([snippet_name, guild_id])
@@index([snippet_name, guild_id])
}
model Infractions {
id BigInt @id @default(autoincrement())
infraction_type String
infraction_reason String?
created_at DateTime? @default(now())
expires_at DateTime?
model Note {
note_id BigInt @id @default(autoincrement())
note_content String
note_created_at DateTime @default(now())
note_moderator_id BigInt
note_target_id BigInt
note_number Int?
guild_id BigInt
guild Guild @relation(fields: [guild_id], references: [guild_id])
// These fields establish a relationship with the `Users` model. `moderator_id` is the ID of the user who gave the infraction. The line `moderator Users? @relation("Moderator", fields: [moderator_id], references: [id])` links to the `Users` model, indicating that an instance of `Users` (the moderator) is associated with this infraction.
moderator Users @relation("Moderator", fields: [moderator_id], references: [id])
moderator_id BigInt
// These fields establish another relationship with the `Users` model. `user_id` is the ID of the user who received the infraction. The line `user Users @relation("User", fields: [user_id], references: [id])` links to the `Users` model, indicating that an instance of `Users` (the user who received the infraction) is associated with this infraction.
user Users @relation("User", fields: [user_id], references: [id])
user_id BigInt
@@unique([note_number, guild_id])
@@index([note_number, guild_id])
}
model Snippets {
// The name of the snippet
name String @id
// The content of the snippet
content String
// The creation time of the snippet
created_at DateTime? @default(now())
// The server ID of the guild where the snippet was created
// 0 is the default value for this field for migration purposes
server_id BigInt @default(0)
// This field establishes a relationship with the `Users` model. `author_id` is the ID of the user who created the snippet. The line `author Users @relation(fields: [author_id], references: [id])` links to the `Users` model, indicating that an instance of `Users` (the author) is associated with this snippet.
author Users @relation(fields: [author_id], references: [id])
author_id BigInt
model Reminder {
reminder_id BigInt @id @default(autoincrement())
reminder_content String
reminder_created_at DateTime @default(now())
reminder_expires_at DateTime
reminder_channel_id BigInt
reminder_user_id BigInt
guild_id BigInt
guild Guild @relation(fields: [guild_id], references: [guild_id])
}
model Notes {
id BigInt @id @default(autoincrement())
content String
created_at DateTime? @default(now())
// These fields establish a relationship with the `Users` model. `moderator_id` is the ID of the user who created the note. The line `moderator Users? @relation("Moderator", fields: [moderator_id], references: [id])` links to the `Users` model, indicating that an instance of `Users` (the moderator) is associated with this note.
moderator Users @relation("Moderator", fields: [moderator_id], references: [id])
moderator_id BigInt
// These fields establish another relationship with the `Users` model. `user_id` is the ID of the user who the note is about. The line `user Users @relation("User", fields: [user_id], references: [id])` links to the `Users` model, indicating that an instance of `Users` (the user who the note is about) is associated with this note.
user Users @relation("User", fields: [user_id], references: [id])
user_id BigInt
}
model Reminders {
id BigInt @id @default(autoincrement())
content String
created_at DateTime? @default(now())
expires_at DateTime
channel_id BigInt
guild_id BigInt
// These fields establish a relationship with the `Users` model. `author_id` is the ID of the user who created the reminder. The line `author Users @relation(fields: [author_id], references: [id])` links to the `Users` model, indicating that an instance of `Users` (the author) is associated with this reminder.
user Users @relation(fields: [user_id], references: [id])
user_id BigInt
}
model Commands {
id BigInt @id @default(autoincrement())
name String
content String
created_at DateTime? @default(now())
CommandStats CommandStats[]
}
model CommandStats {
id BigInt @id @default(autoincrement())
command_id BigInt
user_id BigInt
used_at DateTime? @default(now())
command Commands @relation(fields: [command_id], references: [id])
user Users @relation(fields: [user_id], references: [id])
}
model EmojiStats {
emoji_id BigInt @id
count BigInt
}
// model Logs {
// id BigInt @id
// content String
// log_type String
// created_at DateTime @default(now())
// }

154
tux/cogs/admin/db.py
View file

@ -1,154 +0,0 @@
import asyncio
from collections.abc import Coroutine, Sequence
from typing import Any
import discord
from discord import app_commands
from discord.ext import commands
from loguru import logger
from tux.database.controllers import DatabaseController
GUILD_ONLY_MESSAGE = "This command can only be used in a guild."
class Db(commands.Cog):
def __init__(self, bot: commands.Bot) -> None:
self.bot = bot
self.db_controller = DatabaseController()
group = app_commands.Group(name="seed", description="Database commands.")
@app_commands.checks.has_role("Root")
@group.command(name="members", description="Seeds the database with all members.")
async def seed_members(self, interaction: discord.Interaction) -> None:
"""
Seeds the database with all members in the guild.
Parameters
----------
interaction : discord.Interaction
The interaction object representing the command invocation.
"""
await interaction.response.defer()
if interaction.guild is None:
await interaction.followup.send(GUILD_ONLY_MESSAGE)
return
members: Sequence[discord.Member] = interaction.guild.members
batch_size = 25
for i in range(0, len(members), batch_size):
batch = members[i : i + batch_size]
tasks = [
self.db_controller.users.sync_user(
user_id=member.id,
name=member.name,
display_name=member.display_name,
mention=str(member.mention),
bot=member.bot,
created_at=member.created_at,
joined_at=member.joined_at,
)
for member in batch
]
await asyncio.gather(*tasks)
await asyncio.sleep(1)
await interaction.followup.send("Seeded all members.")
logger.info(f"{interaction.user} seeded all members.")
@app_commands.checks.has_role("Root")
@group.command(name="roles", description="Seeds the database with all roles.")
async def seed_roles(self, interaction: discord.Interaction) -> None:
"""
Seeds the database with all roles in the guild.
Parameters
----------
interaction : discord.Interaction
The interaction object representing the command invocation.
"""
await interaction.response.defer()
if interaction.guild is None:
await interaction.followup.send(GUILD_ONLY_MESSAGE)
return
roles: Sequence[discord.Role] = interaction.guild.roles
batch_size = 25
for i in range(0, len(roles), batch_size):
batch = roles[i : i + batch_size]
tasks = [
self.db_controller.roles.sync_role(
role_id=role.id,
role_name=role.name,
mention=role.mention,
hoist=role.hoist,
managed=role.managed,
mentionable=role.mentionable,
color=role.color.value,
created_at=role.created_at,
)
for role in batch
]
await asyncio.gather(*tasks)
await asyncio.sleep(1)
await interaction.followup.send("Seeded all roles.")
logger.info(f"{interaction.user} seeded all roles.")
@app_commands.checks.has_role("Root")
@group.command(name="user_roles", description="Seeds the database with all user roles.")
async def seed_user_roles(self, interaction: discord.Interaction) -> None:
"""
Seeds the database with all user roles in the guild.
Parameters
----------
interaction : discord.Interaction
The interaction object representing the command invocation.
"""
await interaction.response.defer()
if interaction.guild is None:
await interaction.followup.send(GUILD_ONLY_MESSAGE)
return
members: Sequence[discord.Member] = interaction.guild.members
batch_size = 25
for i in range(0, len(members), batch_size):
batch = members[i : i + batch_size]
tasks: list[Coroutine[Any, Any, None]] = []
for member in batch:
role_ids: list[int] = [role.id for role in member.roles]
tasks.append(
self.db_controller.user_roles.sync_user_roles(
user_id=member.id, role_ids=role_ids
)
)
await asyncio.gather(*tasks)
await asyncio.sleep(1)
await interaction.followup.send("Seeded all user roles.")
logger.info(f"{interaction.user} seeded all user roles.")
async def setup(bot: commands.Bot) -> None:
await bot.add_cog(Db(bot))

27
tux/cogs/logging/gate.py
View file

@ -1,7 +1,6 @@
import discord
from discord.ext import commands
from tux.database.controllers import DatabaseController
from tux.utils.constants import Constants as CONST
from tux.utils.embeds import EmbedCreator
from tux.utils.functions import datetime_to_elapsed_time, datetime_to_unix
@ -10,20 +9,13 @@ from tux.utils.functions import datetime_to_elapsed_time, datetime_to_unix
class GateLogging(commands.Cog):
def __init__(self, bot: commands.Bot):
self.bot = bot
self.db_controller = DatabaseController()
self.gate_log_channel_id: int = CONST.LOG_CHANNELS["GATE"]
self.dev_log_channel_id: int = CONST.LOG_CHANNELS["DEV"]
async def send_to_gate_log(self, embed: discord.Embed) -> None:
channel = self.bot.get_channel(self.gate_log_channel_id)
if isinstance(channel, discord.TextChannel):
await channel.send(embed=embed)
async def send_to_dev_log(self, embed: discord.Embed) -> None:
channel = self.bot.get_channel(self.dev_log_channel_id)
if isinstance(channel, discord.TextChannel):
await channel.send(embed=embed)
@commands.Cog.listener()
async def on_member_join(self, member: discord.Member) -> None:
"""
@ -51,25 +43,6 @@ class GateLogging(commands.Cog):
await self.send_to_gate_log(gate_embed)
new_user = await self.db_controller.users.sync_user(
user_id=member.id,
name=member.name,
display_name=member.display_name,
mention=member.mention,
bot=member.bot,
created_at=member.created_at,
joined_at=member.joined_at,
)
log_embed = EmbedCreator.create_log_embed(
title="Database User Synced",
description=f"User {member.mention} (`{new_user.id}`) synced to database."
if new_user is not None
else f"User {member.mention} synced to database.",
)
await self.send_to_dev_log(log_embed)
@commands.Cog.listener()
async def on_member_remove(self, member: discord.Member) -> None:
"""

3
tux/cogs/logging/member.py
View file

@ -1,7 +1,6 @@
import discord
from discord.ext import commands
from tux.database.controllers import DatabaseController
from tux.utils.constants import Constants as CONST
from tux.utils.embeds import EmbedCreator
@ -9,9 +8,7 @@ from tux.utils.embeds import EmbedCreator
class MemberLogging(commands.Cog):
def __init__(self, bot: commands.Bot):
self.bot = bot
self.db_controller = DatabaseController()
self.audit_log_channel_id: int = CONST.LOG_CHANNELS["AUDIT"]
self.mod_log_channel_id: int = CONST.LOG_CHANNELS["MOD"]
async def send_to_audit_log(self, embed: discord.Embed):
channel = self.bot.get_channel(self.audit_log_channel_id)

40
tux/cogs/utility/remindme.py
View file

@ -7,28 +7,28 @@ from discord import app_commands
from discord.ext import commands
from loguru import logger
from prisma.models import Reminders
from prisma.models import Reminder
from tux.database.controllers import DatabaseController
from tux.utils.embeds import EmbedCreator
from tux.utils.functions import convert_to_seconds
def get_closest_reminder(reminders: list[Reminders]) -> Reminders | None:
def get_closest_reminder(reminders: list[Reminder]) -> Reminder | None:
"""
Check if there are any reminders and return the closest one.
Parameters
----------
reminders : list[Reminders]
reminders : list[Reminder]
A list of reminders to check.
Returns
-------
Reminders | None
Reminder | None
The closest reminder or None if there are no reminders.
"""
return min(reminders, key=lambda x: x.expires_at) if reminders else None
return min(reminders, key=lambda x: x.reminder_expires_at) if reminders else None
class RemindMe(commands.Cog):
@ -37,17 +37,17 @@ class RemindMe(commands.Cog):
self.db_controller = DatabaseController()
self.bot.loop.create_task(self.update())
async def send_reminders(self, reminder: Reminders) -> None:
async def send_reminders(self, reminder: Reminder) -> None:
"""
Send the reminder to the user.
Parameters
----------
reminder : Reminders
reminder : Reminder
The reminder object.
"""
user = self.bot.get_user(reminder.user_id)
user = self.bot.get_user(reminder.reminder_user_id)
if user is not None:
embed = EmbedCreator.custom_footer_embed(
@ -56,7 +56,7 @@ class RemindMe(commands.Cog):
state="SUCCESS",
user=user,
latency="N/A",
content=reminder.content,
content=reminder.reminder_content,
title="Reminder",
)
@ -67,7 +67,7 @@ class RemindMe(commands.Cog):
# Send a message in the channel if the user has DMs closed
channel: (
discord.abc.GuildChannel | discord.Thread | discord.abc.PrivateChannel | None
) = self.bot.get_channel(reminder.channel_id)
) = self.bot.get_channel(reminder.reminder_channel_id)
if channel is not None and isinstance(
channel, discord.TextChannel | discord.Thread | discord.VoiceChannel
@ -84,10 +84,10 @@ class RemindMe(commands.Cog):
)
else:
logger.error(f"Failed to send reminder to {reminder.user_id}, user not found.")
logger.error(f"Failed to send reminder to {reminder.reminder_user_id}, user not found.")
# Delete the reminder after sending
await self.db_controller.reminders.delete_reminder(reminder.id)
await self.db_controller.reminders.delete_reminder_by_id(reminder.reminder_id)
# wait for a second so that the reminder is deleted before checking for more reminders
# who knows if this works, it seems to
@ -96,18 +96,18 @@ class RemindMe(commands.Cog):
# Run update again to check if there are any more reminders
await self.update()
async def end_timer(self, reminder: Reminders) -> None:
async def end_timer(self, reminder: Reminder) -> None:
"""
End the timer for the reminder.
Parameters
----------
reminder : Reminders
reminder : Reminder
The reminder object.
"""
# Wait until the reminder expires
await discord.utils.sleep_until(reminder.expires_at)
await discord.utils.sleep_until(reminder.reminder_expires_at)
await self.send_reminders(reminder)
async def update(self) -> None:
@ -132,7 +132,7 @@ class RemindMe(commands.Cog):
return
# Check if it's expired
if closest_reminder.expires_at < datetime.datetime.now(datetime.UTC):
if closest_reminder.reminder_expires_at < datetime.datetime.now(datetime.UTC):
await self.send_reminders(closest_reminder)
return
@ -168,11 +168,11 @@ class RemindMe(commands.Cog):
seconds = datetime.datetime.now(datetime.UTC) + datetime.timedelta(seconds=seconds)
try:
await self.db_controller.reminders.create_reminder(
user_id=interaction.user.id,
await self.db_controller.reminders.insert_reminder(
reminder_user_id=interaction.user.id,
reminder_content=reminder,
expires_at=seconds,
channel_id=interaction.channel_id or 0,
reminder_expires_at=seconds,
reminder_channel_id=interaction.channel_id or 0,
guild_id=interaction.guild_id or 0,
)

45
tux/cogs/utility/snippets.py
View file

@ -6,7 +6,7 @@ from discord import AllowedMentions
from discord.ext import commands
from loguru import logger
from prisma.models import Snippets as SnippetsModel
from prisma.models import Snippet
from tux.database.controllers import DatabaseController
from tux.utils.embeds import EmbedCreator
@ -38,12 +38,10 @@ class Snippets(commands.Cog):
await ctx.send("This command cannot be used in direct messages.")
return
snippets: list[SnippetsModel] = await self.db_controller.get_all_snippets_sorted(
newestfirst=True
)
snippets: list[Snippet] = await self.db_controller.get_all_snippets_sorted(newestfirst=True)
# remove snippets that are not in the current server
snippets = [snippet for snippet in snippets if snippet.server_id == ctx.guild.id]
snippets = [snippet for snippet in snippets if snippet.guild_id == ctx.guild.id]
# Calculate the number of pages based on the number of snippets
pages = 1 if len(snippets) <= 10 else len(snippets) // 10 + 1
@ -75,7 +73,7 @@ class Snippets(commands.Cog):
title="Snippets",
description="\n".join(
[
f"`{str(index + 1).zfill(2)}. {snippet.name.ljust(20)} | author: {self.bot.get_user(snippet.author_id) or 'Unknown'}`"
f"`{str(index + 1).zfill(2)}. {snippet.snippet_name.ljust(20)} | author: {self.bot.get_user(snippet.snippet_user_id) or 'Unknown'}`"
for index, snippet in enumerate(snippets)
]
),
@ -104,7 +102,7 @@ class Snippets(commands.Cog):
await ctx.send("This command cannot be used in direct messages.")
return
snippet = await self.db_controller.get_snippet_by_name_in_server(name, ctx.guild.id)
snippet = await self.db_controller.get_snippet_by_name_and_guild_id(name, ctx.guild.id)
if snippet is None:
embed = EmbedCreator.create_error_embed(
@ -114,7 +112,7 @@ class Snippets(commands.Cog):
return
# Check if the author of the snippet is the same as the user who wants to delete it and if theres no author don't allow deletion
author_id = snippet.author_id or 0
author_id = snippet.snippet_user_id or 0
if author_id != ctx.author.id:
embed = EmbedCreator.create_error_embed(
title="Error", description="You can only delete your own snippets.", ctx=ctx
@ -122,7 +120,7 @@ class Snippets(commands.Cog):
await ctx.send(embed=embed)
return
await self.db_controller.delete_snippet(name)
await self.db_controller.delete_snippet_by_id(snippet.snippet_id)
await ctx.send("Snippet deleted.")
logger.info(f"{ctx.author} deleted the snippet with the name {name}.")
@ -146,7 +144,7 @@ class Snippets(commands.Cog):
await ctx.send("This command cannot be used in direct messages.")
return
snippet = await self.db_controller.get_snippet_by_name_in_server(name, ctx.guild.id)
snippet = await self.db_controller.get_snippet_by_name_and_guild_id(name, ctx.guild.id)
if snippet is None:
embed = EmbedCreator.create_error_embed(
@ -155,7 +153,7 @@ class Snippets(commands.Cog):
await ctx.send(embed=embed, delete_after=5)
return
text = f"`/snippets/{snippet.name}.txt` || {snippet.content}"
text = f"`/snippets/{snippet.snippet_name}.txt` || {snippet.snippet_content}"
await ctx.send(text, allowed_mentions=AllowedMentions.none())
@ -181,7 +179,7 @@ class Snippets(commands.Cog):
await ctx.send("This command cannot be used in direct messages.")
return
snippet = await self.db_controller.get_snippet_by_name_in_server(name, ctx.guild.id)
snippet = await self.db_controller.get_snippet_by_name_and_guild_id(name, ctx.guild.id)
if snippet is None:
embed = EmbedCreator.create_error_embed(
@ -190,11 +188,11 @@ class Snippets(commands.Cog):
await ctx.send(embed=embed, delete_after=5)
return
author = self.bot.get_user(snippet.author_id) or ctx.author
author = self.bot.get_user(snippet.snippet_user_id) or ctx.author
embed: discord.Embed = EmbedCreator.custom_footer_embed(
title="Snippet Information",
content=f"**Name:** {snippet.name}\n**Author:** {author}\n**Created At:** (set as embed timestamp)\n**Content:** {snippet.content}",
content=f"**Name:** {snippet.snippet_name}\n**Author:** {author}\n**Created At:** (set as embed timestamp)\n**Content:** {snippet.snippet_content}",
ctx=ctx,
latency="N/A",
interaction=None,
@ -202,7 +200,9 @@ class Snippets(commands.Cog):
user=author,
)
embed.timestamp = snippet.created_at or datetime.datetime.fromtimestamp(0, datetime.UTC)
embed.timestamp = snippet.snippet_created_at or datetime.datetime.fromtimestamp(
0, datetime.UTC
)
await ctx.send(embed=embed)
@ -245,7 +245,10 @@ class Snippets(commands.Cog):
server_id = ctx.guild.id
# Check if the snippet already exists
if await self.db_controller.get_snippet_by_name_in_server(name, server_id) is not None:
if (
await self.db_controller.get_snippet_by_name_and_guild_id(name, ctx.guild.id)
is not None
):
embed = EmbedCreator.create_error_embed(
title="Error", description="Snippet already exists.", ctx=ctx
)
@ -264,11 +267,11 @@ class Snippets(commands.Cog):
return
await self.db_controller.create_snippet(
name=name,
content=content,
created_at=created_at,
author_id=author_id,
server_id=server_id,
snippet_name=name,
snippet_content=content,
snippet_created_at=created_at,
snippet_user_id=author_id,
guild_id=server_id,
)
await ctx.send("Snippet created.")

28
tux/database/controllers/__init__.py
View file

@ -1,21 +1,15 @@
from .emojistats import EmojiStatsController
from .infractions import InfractionsController
from .notes import NotesController
from .reminders import RemindersController
from .roles import RolesController
from .snippets import SnippetsController
from .user_roles import UserRolesController
from .users import UsersController
from .cases import CaseController
from .notes import NoteController
from .reminders import ReminderController
from .snippets import SnippetController
class DatabaseController:
def __init__(self):
"""Initializes the database controller and connects to all the database tables."""
self.users = UsersController()
self.infractions = InfractionsController()
self.notes = NotesController()
self.snippets = SnippetsController()
self.reminders = RemindersController()
self.roles = RolesController()
self.user_roles = UserRolesController()
self.emojistats = EmojiStatsController()
self.cases = CaseController()
self.notes = NoteController()
self.snippets = SnippetController()
self.reminders = ReminderController()
db_controller = DatabaseController()

305
tux/database/controllers/cases.py Normal file
View file

@ -0,0 +1,305 @@
from datetime import datetime
from prisma.enums import CaseType
from prisma.models import Case
from tux.database.client import db
class CaseController:
def __init__(self):
self.table = db.case
async def get_all_cases(self) -> list[Case]:
return await self.table.find_many()
async def get_case_by_id(self, case_id: int) -> Case | None:
return await self.table.find_first(where={"case_id": case_id})
async def insert_case(
self,
guild_id: int,
case_target_id: int,
case_moderator_id: int,
case_type: CaseType,
case_reason: str,
case_expires_at: datetime | None = None,
) -> Case | None:
return await self.table.create(
data={
"guild_id": guild_id,
"case_target_id": case_target_id,
"case_moderator_id": case_moderator_id,
"case_reason": case_reason,
"case_type": case_type,
"case_expires_at": case_expires_at,
}
)
async def delete_case_by_id(self, case_id: int) -> None:
await self.table.delete(where={"case_id": case_id})
async def update_case_by_case_number_and_guild_id(
self, case_number: int, guild_id: int, case_reason: str
) -> Case | None:
return await self.table.update(
where={"case_number_guild_id": {"case_number": case_number, "guild_id": guild_id}},
data={"case_reason": case_reason},
)
async def get_cases_by_guild_id(self, guild_id: int) -> list[Case] | None:
return await self.table.find_many(where={"guild_id": guild_id})
async def get_cases_by_target_id(self, case_target_id: int) -> list[Case] | None:
return await self.table.find_many(where={"case_target_id": case_target_id})
async def get_cases_by_moderator_id(self, case_moderator_id: int) -> list[Case] | None:
return await self.table.find_many(where={"case_moderator_id": case_moderator_id})
async def get_cases_by_guild_id_and_target_id(
self, guild_id: int, case_target_id: int
) -> list[Case] | None:
return await self.table.find_many(
where={"guild_id": guild_id, "case_target_id": case_target_id}
)
async def get_cases_by_guild_id_and_moderator_id(
self, guild_id: int, case_moderator_id: int
) -> list[Case] | None:
return await self.table.find_many(
where={"guild_id": guild_id, "case_moderator_id": case_moderator_id}
)
async def get_cases_by_guild_id_and_type(
self, guild_id: int, case_type: CaseType
) -> list[Case] | None:
return await self.table.find_many(where={"guild_id": guild_id, "case_type": case_type})
async def get_cases_by_guild_id_and_reason(
self, guild_id: int, case_reason: str
) -> list[Case] | None:
return await self.table.find_many(where={"guild_id": guild_id, "case_reason": case_reason})
async def get_cases_by_guild_id_and_expires_at(
self, guild_id: int, case_expires_at: datetime
) -> list[Case] | None:
return await self.table.find_many(
where={"guild_id": guild_id, "case_expires_at": case_expires_at}
)
async def get_cases_by_guild_id_and_target_id_and_moderator_id(
self, guild_id: int, case_target_id: int, case_moderator_id: int
) -> list[Case] | None:
return await self.table.find_many(
where={
"guild_id": guild_id,
"case_target_id": case_target_id,
"case_moderator_id": case_moderator_id,
}
)
async def get_cases_by_guild_id_and_target_id_and_type(
self, guild_id: int, case_target_id: int, case_type: CaseType
) -> list[Case] | None:
return await self.table.find_many(
where={"guild_id": guild_id, "case_target_id": case_target_id, "case_type": case_type}
)
async def get_cases_by_guild_id_and_target_id_and_reason(
self, guild_id: int, case_target_id: int, case_reason: str
) -> list[Case] | None:
return await self.table.find_many(
where={
"guild_id": guild_id,
"case_target_id": case_target_id,
"case_reason": case_reason,
}
)
async def get_cases_by_guild_id_and_target_id_and_expires_at(
self, guild_id: int, case_target_id: int, case_expires_at: datetime
) -> list[Case] | None:
return await self.table.find_many(
where={
"guild_id": guild_id,
"case_target_id": case_target_id,
"case_expires_at": case_expires_at,
}
)
async def get_cases_by_guild_id_and_moderator_id_and_type(
self, guild_id: int, case_moderator_id: int, case_type: CaseType
) -> list[Case] | None:
return await self.table.find_many(
where={
"guild_id": guild_id,
"case_moderator_id": case_moderator_id,
"case_type": case_type,
}
)
async def get_cases_by_guild_id_and_moderator_id_and_reason(
self, guild_id: int, case_moderator_id: int, case_reason: str
) -> list[Case] | None:
return await self.table.find_many(
where={
"guild_id": guild_id,
"case_moderator_id": case_moderator_id,
"case_reason": case_reason,
}
)
async def get_cases_by_guild_id_and_moderator_id_and_expires_at(
self, guild_id: int, case_moderator_id: int, case_expires_at: datetime
) -> list[Case] | None:
return await self.table.find_many(
where={
"guild_id": guild_id,
"case_moderator_id": case_moderator_id,
"case_expires_at": case_expires_at,
}
)
async def get_cases_by_guild_id_and_type_and_reason(
self, guild_id: int, case_type: CaseType, case_reason: str
) -> list[Case] | None:
return await self.table.find_many(
where={"guild_id": guild_id, "case_type": case_type, "case_reason": case_reason}
)
async def get_cases_by_guild_id_and_type_and_expires_at(
self, guild_id: int, case_type: CaseType, case_expires_at: datetime
) -> list[Case] | None:
return await self.table.find_many(
where={"guild_id": guild_id, "case_type": case_type, "case_expires_at": case_expires_at}
)
async def get_cases_by_guild_id_and_reason_and_expires_at(
self, guild_id: int, case_reason: str, case_expires_at: datetime
) -> list[Case] | None:
return await self.table.find_many(
where={
"guild_id": guild_id,
"case_reason": case_reason,
"case_expires_at": case_expires_at,
}
)
async def get_cases_by_guild_id_and_target_id_and_moderator_id_and_type(
self, guild_id: int, case_target_id: int, case_moderator_id: int, case_type: CaseType
) -> list[Case] | None:
return await self.table.find_many(
where={
"guild_id": guild_id,
"case_target_id": case_target_id,
"case_moderator_id": case_moderator_id,
"case_type": case_type,
}
)
async def get_cases_by_guild_id_and_target_id_and_moderator_id_and_reason(
self, guild_id: int, case_target_id: int, case_moderator_id: int, case_reason: str
) -> list[Case] | None:
return await self.table.find_many(
where={
"guild_id": guild_id,
"case_target_id": case_target_id,
"case_moderator_id": case_moderator_id,
"case_reason": case_reason,
}
)
async def get_cases_by_guild_id_and_target_id_and_moderator_id_and_expires_at(
self, guild_id: int, case_target_id: int, case_moderator_id: int, case_expires_at: datetime
) -> list[Case] | None:
return await self.table.find_many(
where={
"guild_id": guild_id,
"case_target_id": case_target_id,
"case_moderator_id": case_moderator_id,
"case_expires_at": case_expires_at,
}
)
async def get_cases_by_guild_id_and_target_id_and_type_and_reason(
self, guild_id: int, case_target_id: int, case_type: CaseType, case_reason: str
) -> list[Case] | None:
return await self.table.find_many(
where={
"guild_id": guild_id,
"case_target_id": case_target_id,
"case_type": case_type,
"case_reason": case_reason,
}
)
async def get_cases_by_guild_id_and_target_id_and_type_and_expires_at(
self, guild_id: int, case_target_id: int, case_type: CaseType, case_expires_at: datetime
) -> list[Case] | None:
return await self.table.find_many(
where={
"guild_id": guild_id,
"case_target_id": case_target_id,
"case_type": case_type,
"case_expires_at": case_expires_at,
}
)
async def get_cases_by_guild_id_and_target_id_and_reason_and_expires_at(
self, guild_id: int, case_target_id: int, case_reason: str, case_expires_at: datetime
) -> list[Case] | None:
return await self.table.find_many(
where={
"guild_id": guild_id,
"case_target_id": case_target_id,
"case_reason": case_reason,
"case_expires_at": case_expires_at,
}
)
async def get_cases_by_guild_id_and_moderator_id_and_type_and_reason(
self, guild_id: int, case_moderator_id: int, case_type: CaseType, case_reason: str
) -> list[Case] | None:
return await self.table.find_many(
where={
"guild_id": guild_id,
"case_moderator_id": case_moderator_id,
"case_type": case_type,
"case_reason": case_reason,
}
)
async def get_cases_by_guild_id_and_moderator_id_and_type_and_expires_at(
self, guild_id: int, case_moderator_id: int, case_type: CaseType, case_expires_at: datetime
) -> list[Case] | None:
return await self.table.find_many(
where={
"guild_id": guild_id,
"case_moderator_id": case_moderator_id,
"case_type": case_type,
"case_expires_at": case_expires_at,
}
)
async def get_cases_by_guild_id_and_moderator_id_and_reason_and_expires_at(
self, guild_id: int, case_moderator_id: int, case_reason: str, case_expires_at: datetime
) -> list[Case] | None:
return await self.table.find_many(
where={
"guild_id": guild_id,
"case_moderator_id": case_moderator_id,
"case_reason": case_reason,
"case_expires_at": case_expires_at,
}
)
async def get_cases_by_guild_id_and_type_and_reason_and_expires_at(
self, guild_id: int, case_type: CaseType, case_reason: str, case_expires_at: datetime
) -> list[Case] | None:
return await self.table.find_many(
where={
"guild_id": guild_id,
"case_type": case_type,
"case_reason": case_reason,
"case_expires_at": case_expires_at,
}
)

104
tux/database/controllers/emojistats.py
View file

@ -1,104 +0,0 @@
from prisma.models import EmojiStats
from tux.database.client import db
class EmojiStatsController:
def __init__(self) -> None:
self.table = db.emojistats
async def get_all_emoji_stats(self) -> list[EmojiStats]:
"""
Retrieves all emoji stats from the database.
Returns
-------
list[EmojiStats]
A list of all emoji stats.
"""
return await self.table.find_many()
async def get_emoji_stats(self, emoji_id: int) -> EmojiStats | None:
"""
Retrieves the emoji stats for the specified emoji ID.
Parameters
----------
emoji_id : int
The ID of the emoji to retrieve.
Returns
-------
EmojiStats | None
The emoji stats if found, otherwise None.
"""
return await self.table.find_first(where={"emoji_id": emoji_id})
async def create_emoji_stats(self, emoji_id: int, count: int) -> EmojiStats:
"""
Creates a new entry for the specified emoji in the database.
Parameters
----------
emoji_id : int
The ID of the emoji.
count : int
The count of the emoji.
Returns
-------
EmojiStats
The newly created emoji stats.
"""
return await self.table.create(
data={
"emoji_id": emoji_id,
"count": count,
}
)
async def increment_emoji_count_or_create(self, emoji_id: int) -> EmojiStats | None:
"""
Increments the count of the specified emoji in the database or creates a new entry if it does not exist.
Parameters
----------
emoji_id : int
The ID of the emoji to increment.
Returns
-------
EmojiStats | None
The updated or newly created emoji stats if successful, None otherwise.
"""
emoji_stats = await self.get_emoji_stats(emoji_id)
if emoji_stats is None:
return await self.create_emoji_stats(emoji_id, 1)
return await self.update_emoji_count(emoji_id, emoji_stats.count + 1)
async def update_emoji_count(self, emoji_id: int, count: int) -> EmojiStats | None:
"""
Updates the count of the specified emoji in the database.
Parameters
----------
emoji_id : int
The ID of the emoji to update.
count : int
The new count of the emoji.
Returns
-------
EmojiStats | None
The updated emoji stats if successful, None otherwise.
"""
return await self.table.update(
where={"emoji_id": emoji_id},
data={"count": count},
)

117
tux/database/controllers/infractions.py
View file

@ -1,117 +0,0 @@
from datetime import datetime
from prisma.models import Infractions
from tux.database.client import db
from tux.utils.enums import InfractionType
class InfractionsController:
def __init__(self):
"""
Initializes the controller and connects to the infractions table in the database.
"""
self.table = db.infractions
async def get_all_infractions(self) -> list[Infractions]:
"""
Retrieves all infractions from the database.
Returns
-------
list[Infractions]
A list of all infractions in the database.
"""
return await self.table.find_many()
async def get_infraction_by_id(self, infraction_id: int) -> Infractions | None:
"""
Retrieves an infraction from the database based on the specified infraction ID.
Parameters
----------
infraction_id : int
The ID of the infraction to retrieve.
Returns
-------
Infractions or None
The infraction if found, otherwise None.
"""
return await self.table.find_first(where={"id": infraction_id})
async def create_infraction(
self,
user_id: int,
moderator_id: int,
infraction_type: InfractionType,
infraction_reason: str,
expires_at: datetime | None = None,
) -> Infractions:
"""
Creates a new infraction in the database with the specified details.
Parameters
----------
user_id : int
The ID of the user for whom the infraction is created.
moderator_id : int
The ID of the moderator who created the infraction.
infraction_type : InfractionType
The type of the infraction.
infraction_reason : str
The reason for the infraction.
expires_at : datetime, optional
The expiration date of the infraction, by default None.
Returns
-------
Infractions
The newly created infraction.
"""
return await self.table.create(
data={
"user_id": user_id,
"moderator_id": moderator_id,
"infraction_type": infraction_type.value,
"infraction_reason": infraction_reason,
"expires_at": expires_at,
}
)
async def delete_infraction(self, infraction_id: int) -> None:
"""
Deletes an infraction from the database based on the specified infraction ID.
Parameters
----------
infraction_id : int
The ID of the infraction to delete.
Returns
-------
None
"""
await self.table.delete(where={"id": infraction_id})
async def update_infraction(
self, infraction_id: int, infraction_reason: str
) -> Infractions | None:
"""
Updates an infraction in the database with the given ID and new reason.
Parameters
----------
infraction_id : int
The ID of the infraction to update.
infraction_reason : str
The new reason for the infraction.
Returns
-------
Infractions or None
The updated infraction if successful, or None if not found.
"""
return await self.table.update(
where={"id": infraction_id},
data={"infraction_reason": infraction_reason},
)

166
tux/database/controllers/notes.py
View file

@ -1,137 +1,77 @@
from prisma.models import Notes
from prisma.models import Note
from tux.database.client import db
class NotesController:
def __init__(self) -> None:
"""
Initializes the controller and connects to the notes table in the database.
"""
self.table = db.notes
class NoteController:
def __init__(self):
self.table = db.note
async def get_all_notes(self) -> list[Notes]:
"""
Retrieves all notes from the database.
Returns
-------
list[Notes]
A list of all notes from the database.
"""
async def get_all_notes(self) -> list[Note]:
return await self.table.find_many()
async def get_note_by_id(self, note_id: int) -> Notes | None:
"""
Retrieves a note from the database based on the specified note ID.
async def get_note_by_id(self, note_id: int) -> Note | None:
return await self.table.find_first(where={"note_id": note_id})
Parameters
----------
note_id : int
The ID of the note to retrieve.
Returns
-------
Notes or None
The note if found, otherwise None.
"""
return await self.table.find_first(where={"id": note_id})
async def create_note(
async def insert_note(
self,
user_id: int,
moderator_id: int,
note_target_id: int,
note_moderator_id: int,
note_content: str,
) -> Notes:
"""
Creates a new note in the database with the specified user ID, moderator ID, and content.
Parameters
----------
user_id : int
The ID of the user for whom the note is created.
moderator_id : int
The ID of the moderator who created the note.
note_content : str
The content of the note.
Returns
-------
Notes
The newly created note.
"""
guild_id: int,
) -> Note:
return await self.table.create(
data={
"user_id": user_id,
"moderator_id": moderator_id,
"content": note_content,
"note_target_id": note_target_id,
"note_moderator_id": note_moderator_id,
"note_content": note_content,
"guild_id": guild_id,
}
)
async def delete_note(self, note_id: int) -> None:
"""
Deletes a note from the database based on the specified note ID.
async def delete_note_by_id(self, note_id: int) -> None:
await self.table.delete(where={"note_id": note_id})
Parameters
----------
note_id : int
The ID of the note to delete.
Returns
-------
None
"""
await self.table.delete(where={"id": note_id})
async def update_note(self, note_id: int, note_content: str) -> Notes | None:
"""
Updates a note in the database with the specified note ID and new content.
Parameters
----------
note_id : int
The ID of the note to update.
note_content : str
The new content for the note.
Returns
-------
Notes or None
The updated note if successful, otherwise None if the note was not found.
"""
async def update_note_by_id(self, note_id: int, note_content: str) -> Note | None:
return await self.table.update(
where={"id": note_id},
data={"content": note_content},
where={"note_id": note_id},
data={"note_content": note_content},
)
async def get_notes_by_user_id(self, user_id: int) -> list[Notes]:
"""
Retrieves all notes from the database based on the specified user ID.
async def get_notes_by_target_id(self, target_id: int) -> list[Note]:
return await self.table.find_many(where={"note_target_id": target_id})
Parameters
----------
user_id : int
The ID of the user to retrieve notes for.
async def get_notes_by_moderator_id(self, moderator_id: int) -> list[Note]:
return await self.table.find_many(where={"note_moderator_id": moderator_id})
Returns
-------
list[Notes]
A list of all notes for the specified user.
"""
async def get_notes_by_guild_id(self, guild_id: int) -> list[Note]:
return await self.table.find_many(where={"guild_id": guild_id})
return await self.table.find_many(where={"user_id": user_id})
async def get_notes_by_target_id_and_guild_id(
self, target_id: int, guild_id: int
) -> list[Note]:
return await self.table.find_many(where={"note_target_id": target_id, "guild_id": guild_id})
async def get_notes_by_moderator_id(self, moderator_id: int) -> list[Notes]:
"""
Retrieves all notes from the database based on the specified moderator ID.
async def get_notes_by_moderator_id_and_guild_id(
self, moderator_id: int, guild_id: int
) -> list[Note]:
return await self.table.find_many(
where={"note_moderator_id": moderator_id, "guild_id": guild_id}
)
Parameters
----------
moderator_id : int
The ID of the moderator to retrieve notes for.
async def get_notes_by_target_id_and_moderator_id(
self, target_id: int, moderator_id: int
) -> list[Note]:
return await self.table.find_many(
where={"note_target_id": target_id, "note_moderator_id": moderator_id}
)
Returns
-------
list[Notes]
A list of all notes created by the specified moderator.
"""
return await self.table.find_many(where={"moderator_id": moderator_id})
async def get_notes_by_target_id_moderator_id_and_guild_id(
self, target_id: int, moderator_id: int, guild_id: int
) -> list[Note]:
return await self.table.find_many(
where={
"note_target_id": target_id,
"note_moderator_id": moderator_id,
"guild_id": guild_id,
}
)

114
tux/database/controllers/reminders.py
View file

@ -1,114 +1,44 @@
from datetime import datetime
from prisma.models import Reminders
from prisma.models import Reminder
from tux.database.client import db
class RemindersController:
class ReminderController:
def __init__(self) -> None:
"""
Initializes the controller and connects to the reminders table in the database.
"""
self.table = db.reminders
self.table = db.reminder
async def get_all_reminders(self) -> list[Reminders]:
"""
Retrieves all reminders from the database.
Returns
-------
list[Reminders]
A list of all reminders.
"""
async def get_all_reminders(self) -> list[Reminder]:
return await self.table.find_many()
async def get_reminder_by_id(self, reminder_id: int) -> Reminders | None:
"""
Retrieves a reminder from the database based on the specified reminder ID.
async def get_reminder_by_id(self, reminder_id: int) -> Reminder | None:
return await self.table.find_first(where={"reminder_id": reminder_id})
Parameters
----------
reminder_id : int
The ID of the reminder to retrieve.
Returns
-------
Reminders or None
The reminder if found, otherwise None.
"""
return await self.table.find_first(where={"id": reminder_id})
async def create_reminder(
async def insert_reminder(
self,
user_id: int,
reminder_user_id: int,
reminder_content: str,
expires_at: datetime,
channel_id: int,
reminder_expires_at: datetime,
reminder_channel_id: int,
guild_id: int,
) -> Reminders:
"""
Creates a new reminder in the database with the specified user ID, reminder content, expiration date, channel ID, and guild ID.
Parameters
----------
user_id : int
The ID of the user for whom the reminder is created.
reminder_content : str
The content of the reminder.
expires_at : datetime
The expiration date and time of the reminder.
channel_id : int
The ID of the channel associated with the reminder.
guild_id : int
The ID of the guild associated with the reminder.
Returns
-------
Reminders
The newly created reminder.
"""
) -> Reminder:
return await self.table.create(
data={
"user_id": user_id,
"content": reminder_content,
"expires_at": expires_at,
"channel_id": channel_id,
"reminder_user_id": reminder_user_id,
"reminder_content": reminder_content,
"reminder_expires_at": reminder_expires_at,
"reminder_channel_id": reminder_channel_id,
"guild_id": guild_id,
}
)
async def delete_reminder(self, reminder_id: int) -> None:
"""
Deletes a reminder from the database based on the specified reminder ID.
async def delete_reminder_by_id(self, reminder_id: int) -> None:
await self.table.delete(where={"reminder_id": reminder_id})
Parameters
----------
reminder_id : int
The ID of the reminder to delete.
Returns
-------
None
"""
await self.table.delete(where={"id": reminder_id})
async def update_reminder(self, reminder_id: int, reminder_content: str) -> Reminders | None:
"""
Updates a reminder in the database with the specified reminder ID and new content.
Parameters
----------
reminder_id : int
The ID of the reminder to update.
reminder_content : str
The new content for the reminder.
Returns
-------
Reminders or None
The updated reminder if successful, or None if the reminder was not found.
"""
async def update_reminder_by_id(
self, reminder_id: int, reminder_content: str
) -> Reminder | None:
return await self.table.update(
where={"id": reminder_id},
data={"content": reminder_content},
where={"reminder_id": reminder_id},
data={"reminder_content": reminder_content},
)

226
tux/database/controllers/roles.py
View file

@ -1,226 +0,0 @@
from datetime import datetime
from prisma.models import Roles
from tux.database.client import db
class RolesController:
def __init__(self) -> None:
"""
Initializes the controller and connects to the roles table in the database.
"""
self.table = db.roles
async def get_all_roles(self) -> list[Roles]:
"""
Retrieves all roles from the database.
Returns
-------
list[Roles]
A list of all roles.
"""
return await self.table.find_many()
async def get_role_by_id(self, role_id: int) -> Roles | None:
"""
Retrieves a role from the database based on the specified role ID.
Parameters
----------
role_id : int
The ID of the role to retrieve.
Returns
-------
Roles or None
The role if found, otherwise None.
"""
return await self.table.find_first(where={"id": role_id})
async def create_role(
self,
role_id: int,
role_name: str,
hoist: bool = False,
managed: bool = False,
mentionable: bool = False,
created_at: datetime | None = None,
mention: str | None = None,
color: int | None = None,
) -> Roles:
"""
Creates a new role in the database with the specified details.
Parameters
----------
role_id : int
The ID assigned to the new role.
role_name : str
The name of the role.
hoist : bool, optional
Whether the role is hoisted (visible separately on the user list), by default False.
managed : bool, optional
Whether the role is managed by an external service, by default False.
mentionable : bool, optional
Whether the role is mentionable, by default False.
created_at : datetime, optional
The creation date of the role, by default None.
mention : str, optional
The mention string for the role, by default None.
color : int, optional
The display color of the role as an integer, by default None.
Returns
-------
Roles
The newly created role.
"""
return await self.table.create(
data={
"id": role_id,
"name": role_name,
"hoist": hoist,
"managed": managed,
"mentionable": mentionable,
"created_at": created_at,
"mention": mention,
"color": color,
}
)
async def update_role(
self,
role_id: int,
role_name: str | None = None,
hoist: bool | None = None,
managed: bool | None = None,
mentionable: bool | None = None,
mention: str | None = None,
color: int | None = None,
) -> Roles | None:
"""
Updates a role in the database with the specified parameters.
Parameters
----------
role_id : int
The ID of the role to update.
role_name : str or None, optional
The new name of the role, if changing.
hoist : bool or None, optional
New hoist state if changing, else None.
managed : bool or None, optional
New managed state if changing, else None.
mentionable : bool or None, optional
New mentionable state if changing, else None.
mention : str or None, optional
New mention string if changing, else None.
color : int or None, optional
New color if changing, default to None if no change.
Returns
-------
Roles or None
The updated role if found and updated, otherwise None if the role does not exist.
"""
return await self.table.update(
where={"id": role_id},
data={
"name": role_name or "",
"hoist": hoist if hoist is not None else False,
"managed": managed if managed is not None else False,
"mentionable": mentionable if mentionable is not None else False,
"mention": mention or "",
"color": color if color is not None else 0,
},
)
async def sync_role(
self,
role_id: int,
role_name: str,
hoist: bool,
managed: bool,
mentionable: bool,
mention: str,
color: int,
created_at: datetime,
) -> Roles | None:
"""
Synchronizes a role in the database with the specified parameters.
Parameters
----------
role_id : int
The ID of the role to synchronize.
role_name : str
The name of the role.
hoist : bool
Whether the role is hoisted.
managed : bool
Whether the role is managed.
mentionable : bool
Whether the role is mentionable.
mention : str
The mention string for the role.
color : int
The color of the role.
created_at : datetime
The creation date of the role.
Returns
-------
Roles | None
The updated role if successful, otherwise None.
"""
existing_role = await self.get_role_by_id(role_id)
if existing_role is None:
return await self.create_role(
role_id=role_id,
role_name=role_name,
hoist=hoist,
managed=managed,
mentionable=mentionable,
mention=mention,
color=color,
created_at=created_at,
)
if (
existing_role.name != role_name
or existing_role.hoist != hoist
or existing_role.managed != managed
or existing_role.mentionable != mentionable
or existing_role.mention != mention
or existing_role.color != color
or existing_role.created_at != created_at
):
return await self.update_role(
role_id=role_id,
role_name=role_name if existing_role.name != role_name else None,
hoist=hoist if existing_role.hoist != hoist else None,
managed=managed if existing_role.managed != managed else None,
mentionable=mentionable if existing_role.mentionable != mentionable else None,
mention=mention if existing_role.mention != mention else None,
color=color if existing_role.color != color else None,
)
return existing_role
async def delete_role(self, role_id: int) -> None:
"""
Deletes a role from the database based on the specified role ID.
Parameters
----------
role_id : int
The ID of the role to delete.
Returns
-------
None
"""
await self.table.delete(where={"id": role_id})

163
tux/database/controllers/snippets.py
View file

@ -1,152 +1,53 @@
import datetime
from prisma.models import Snippets
from prisma.models import Snippet
from tux.database.client import db
class SnippetsController:
class SnippetController:
def __init__(self) -> None:
"""
Initializes the controller and connects to the snippets table in the database.
"""
self.table = db.snippets
self.table = db.snippet
async def get_all_snippets(self) -> list[Snippets]:
"""
Retrieves all snippets from the database.
Returns
-------
list[Snippets]
A list of all snippets.
"""
async def get_all_snippets(self) -> list[Snippet]:
return await self.table.find_many()
async def get_all_snippets_sorted(self, newestfirst: bool = True) -> list[Snippets]:
"""
Retrieves all snippets from the database sorted by creation date.
async def get_all_snippets_sorted(self, newestfirst: bool = True) -> list[Snippet]:
return await self.table.find_many(
order={"snippet_created_at": "desc" if newestfirst else "asc"}
)
Parameters
----------
newestfirst : bool, optional
Whether to sort the snippets by newest first or oldest first, by default True.
async def get_snippet_by_name(self, snippet_name: str) -> Snippet | None:
return await self.table.find_first(where={"snippet_name": snippet_name})
Returns
-------
list[Snippets]
A list of all snippets sorted by creation date.
"""
return await self.table.find_many(order={"created_at": "desc" if newestfirst else "asc"})
async def get_snippet_by_name(self, name: str) -> Snippets | None:
"""
Retrieves a snippet from the database based on the specified name.
Parameters
----------
name : str
The name of the snippet to retrieve.
Returns
-------
Snippets or None
The snippet if found, otherwise None.
"""
return await self.table.find_first(where={"name": name})
async def get_snippet_by_name_in_server(self, name: str, server_id: int) -> Snippets | None:
"""
Retrieves a snippet from the database based on the specified name and server ID.
Parameters
----------
name : str
The name of the snippet to retrieve.
server_id : int
The server ID to retrieve the snippet from.
Returns
-------
Snippets or None
The snippet if found, otherwise None.
"""
return await self.table.find_first(where={"name": name, "server_id": server_id})
async def get_snippet_by_name_and_guild_id(
self, snippet_name: str, guild_id: int
) -> Snippet | None:
return await self.table.find_first(
where={"snippet_name": snippet_name, "guild_id": guild_id}
)
async def create_snippet(
self, name: str, content: str, created_at: datetime.datetime, author_id: int, server_id: int
) -> Snippets:
"""
Creates a new snippet in the database with the specified name, content, creation date, and author ID.
Parameters
----------
name : str
The name of the snippet.
content : str
The content of the snippet.
created_at : datetime.datetime
The creation date of the snippet.
author_id : int
The ID of the author who created the snippet.
Returns
-------
Snippets
The newly created snippet.
"""
self,
snippet_name: str,
snippet_content: str,
snippet_created_at: datetime.datetime,
snippet_user_id: int,
guild_id: int,
) -> Snippet:
return await self.table.create(
data={
"name": name,
"content": content,
"created_at": created_at,
"author_id": author_id,
"server_id": server_id,
"snippet_name": snippet_name,
"snippet_content": snippet_content,
"snippet_created_at": snippet_created_at,
"snippet_user_id": snippet_user_id,
"guild_id": guild_id,
}
)
async def delete_snippet(self, name: str) -> None:
"""
Deletes a snippet from the database based on the specified name.
Parameters
----------
name : str
The name of the snippet to delete.
Returns
-------
None
"""
await self.table.delete(where={"name": name})
async def update_snippet(
self, name: str, content: str, server_id: int | None
) -> Snippets | None:
"""
Updates a snippet in the database with the specified name and new content.
Parameters
----------
name : str
The name of the snippet to update.
content : str
The new content for the snippet.
server_id : int, optional
The server ID to update the snippet with, by default None.
Returns
-------
Snippets or None
The updated snippet if successful, otherwise None if the snippet was not found.
"""
if server_id:
return await self.table.update(
where={"name": name},
data={"content": content, "server_id": server_id},
)
async def delete_snippet_by_id(self, snippet_id: int) -> None:
await self.table.delete(where={"snippet_id": snippet_id})
async def update_snippet_by_id(self, snippet_id: int, snippet_content: str) -> Snippet | None:
return await self.table.update(
where={"name": name},
data={"content": content},
where={"snippet_id": snippet_id}, data={"snippet_content": snippet_content}
)

172
tux/database/controllers/user_roles.py
View file

@ -1,172 +0,0 @@
from prisma.models import UserRoles
from tux.database.client import db
class UserRolesController:
def __init__(self) -> None:
"""
Initializes the controller and connects to the user_roles table in the database.
"""
self.table = db.userroles
async def get_all_user_roles(self) -> list[UserRoles]:
"""
Retrieves all user roles from the database.
Returns
-------
list[UserRoles]
A list of all user roles.
"""
return await self.table.find_many()
async def get_user_role_by_ids(self, user_id: int, role_id: int) -> UserRoles | None:
"""
Retrieves a user role from the database based on the specified user ID and role ID.
Parameters
----------
user_id : int
The ID of the user.
role_id : int
The ID of the role.
Returns
-------
UserRoles or None
The user role if found, otherwise None.
"""
return await self.table.find_first(where={"user_id": user_id, "role_id": role_id})
async def create_user_role(self, user_id: int, role_id: int) -> UserRoles:
"""
Creates a new user role with the specified user ID and role ID.
Parameters
----------
user_id : int
The user ID.
role_id : int
The role ID.
Returns
-------
UserRoles
The newly created user role.
"""
return await self.table.create(data={"user_id": user_id, "role_id": role_id})
async def delete_user_role(self, user_id: int, role_id: int) -> None:
"""
Deletes a user role from the database based on specified user ID and role ID.
Parameters
----------
user_id : int
The user ID.
role_id : int
The role ID.
Returns
-------
None
"""
await self.table.delete(where={"user_id_role_id": {"user_id": user_id, "role_id": role_id}})
async def delete_user_roles(self, user_id: int) -> None:
"""
Deletes all user roles from the database based on the specified user ID.
Parameters
----------
user_id : int
The user ID.
Returns
-------
None
"""
await self.table.delete_many(where={"user_id": user_id})
async def delete_role_users(self, role_id: int) -> None:
"""
Deletes all user roles from the database based on the specified role ID.
Parameters
----------
role_id : int
The role ID.
Returns
-------
None
"""
await self.table.delete_many(where={"role_id": role_id})
async def delete_all_user_roles(self) -> None:
"""
Deletes all user roles from the database.
Returns
-------
None
"""
await self.table.delete_many()
async def get_user_roles_by_user_id(self, user_id: int) -> list[UserRoles]:
"""
Retrieves all user roles from the database based on the specified user ID.
Parameters
----------
user_id : int
The user ID.
Returns
-------
list[UserRoles]
A list of user roles associated with the user.
"""
return await self.table.find_many(where={"user_id": user_id})
async def get_user_roles_by_role_id(self, role_id: int) -> list[UserRoles]:
"""
Retrieves all user roles from the database based on the specified role ID.
Parameters
----------
role_id : int
The role ID.
Returns
-------
list[UserRoles]
A list of user roles associated with the role.
"""
return await self.table.find_many(where={"role_id": role_id})
async def sync_user_roles(self, user_id: int, role_ids: list[int]) -> None:
"""
Synchronizes user roles in the database based on the specified user ID and role IDs.
Parameters
----------
user_id : int
The user ID.
role_ids : list[int]
The list of role IDs to synchronize.
Returns
-------
None
"""
user_roles = await self.get_user_roles_by_user_id(user_id)
user_role_ids = [user_role.role_id for user_role in user_roles]
for role_id in role_ids:
if role_id not in user_role_ids:
await self.create_user_role(user_id=user_id, role_id=role_id)
for user_role in user_roles:
if user_role.role_id not in role_ids:
await self.delete_user_role(user_id=user_id, role_id=user_role.role_id)

244
tux/database/controllers/users.py
View file

@ -1,244 +0,0 @@
from datetime import datetime
from prisma.models import Users
from prisma.types import UsersUpdateInput
from tux.database.client import db
class UsersController:
def __init__(self) -> None:
"""
Initializes the controller and connects to the users table in the database.
"""
self.table = db.users
async def get_all_users(self) -> list[Users]:
"""
Retrieves all users from the database.
Returns
-------
list[Users]
A list of all users.
"""
return await self.table.find_many()
async def get_user_by_id(self, user_id: int) -> Users | None:
"""
Retrieves a user from the database based on the specified user ID.
Parameters
----------
user_id : int
The ID of the user to retrieve.
Returns
-------
Users or None
The user if found, otherwise None.
"""
return await self.table.find_first(where={"id": user_id})
async def create_user(
self,
user_id: int,
name: str,
display_name: str,
mention: str,
bot: bool,
created_at: datetime | None,
joined_at: datetime | None,
) -> Users:
"""
Creates a new user in the database with the specified details.
Parameters
----------
user_id : int
The unique identifier for the user.
name : str
The name of the user.
display_name : str
The display name of the user.
mention : str
The mention tag of the user.
bot : bool
Whether the user is a bot.
created_at : datetime or None
The creation date of the user account.
joined_at : datetime or None
The date the user joined, which could be None.
Returns
-------
Users
The newly created user record.
"""
return await self.table.create(
data={
"id": user_id,
"name": name,
"display_name": display_name,
"mention": mention,
"bot": bot,
"created_at": created_at,
"joined_at": joined_at,
}
)
async def update_user(
self,
user_id: int,
name: str | None = None,
display_name: str | None = None,
mention: str | None = None,
bot: bool | None = None,
created_at: datetime | None = None,
joined_at: datetime | None = None,
) -> Users | None:
"""
Updates user information in the database based on the specified user ID and details.
Parameters
----------
user_id : int
The user ID for the update.
name : str or None, optional
The new name of the user.
display_name : str or None, optional
The new display name of the user.
mention : str or None, optional
The new mention tag of the user.
bot : bool or None, optional
Whether the user is a bot.
created_at : datetime or None, optional
The creation date to set.
joined_at : datetime or None, optional
The join date to set.
Returns
-------
Users or None
The updated user record if successful, otherwise None.
"""
data: UsersUpdateInput = {}
if name is not None:
data["name"] = name
if display_name is not None:
data["display_name"] = display_name
if mention is not None:
data["mention"] = mention
if bot is not None:
data["bot"] = bot
if created_at is not None:
data["created_at"] = created_at
if joined_at is not None:
data["joined_at"] = joined_at
return await self.table.update(where={"id": user_id}, data=data)
async def sync_user(
self,
user_id: int,
name: str,
display_name: str,
mention: str,
bot: bool,
created_at: datetime,
joined_at: datetime | None,
) -> Users | None:
"""
Synchronizes a user in the database based on the specified user ID and details.
Parameters
----------
user_id : int
The user ID to synchronize.
name : str
The name of the user.
display_name : str
The display name of the user.
mention : str
The mention tag of the user.
bot : bool
Whether the user is a bot.
created_at : datetime
The creation date of the user account.
joined_at : datetime or None
The date the user joined, which could be None.
Returns
-------
Users or None
The updated user record if successful, otherwise None.
"""
existing_user = await self.get_user_by_id(user_id)
if existing_user is None:
return await self.create_user(
user_id=user_id,
name=name,
display_name=display_name,
mention=mention,
bot=bot,
created_at=created_at,
joined_at=joined_at,
)
if (
existing_user.name != name
or existing_user.display_name != display_name
or existing_user.mention != mention
or existing_user.bot != bot
or existing_user.created_at != created_at
or existing_user.joined_at != joined_at
):
return await self.update_user(
user_id=user_id,
name=name if existing_user.name != name else None,
display_name=display_name if existing_user.display_name != display_name else None,
mention=mention if existing_user.mention != mention else None,
bot=bot if existing_user.bot != bot else None,
created_at=created_at if existing_user.created_at != created_at else None,
joined_at=joined_at if existing_user.joined_at != joined_at else None,
)
return existing_user
async def delete_user(self, user_id: int) -> None:
"""
Deletes a user from the database based on the specified user ID.
Parameters
----------
user_id : int
The user ID of the user to delete.
Returns
-------
None
"""
await self.table.delete(where={"id": user_id})
async def toggle_afk(self, user_id: int, afk: bool) -> Users | None:
"""
Toggles the AFK status of a user in the database.
Parameters
----------
user_id : int
The ID of the user to toggle AFK status for.
afk : bool
The new AFK status to set for the user.
Returns
-------
Users or None
The updated user if successful, None if the user was not found.
"""
return await self.table.update(
where={"id": user_id},
data={"afk": afk},
)

9
tux/utils/enums.py
View file

@ -1,9 +0,0 @@
from enum import Enum
class InfractionType(Enum):
BAN = "ban"
UNBAN = "unban"
WARN = "warn"
KICK = "kick"
TIMEOUT = "timeout"