didier/database/crud/custom_commands.py

from typing import Optional

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from database.exceptions.constraints import DuplicateInsertException
from database.exceptions.not_found import NoResultFoundException
from database.schemas import CustomCommand, CustomCommandAlias

__all__ = [
    "clean_name",
    "create_alias",
    "create_command",
    "edit_command",
    "get_all_commands",
    "get_command",
    "get_command_by_alias",
    "get_command_by_name",
]


def clean_name(name: str) -> str:
    """Convert a name to lowercase & remove spaces to allow easier matching"""
    return name.lower().replace(" ", "")


async def create_command(session: AsyncSession, name: str, response: str) -> CustomCommand:
    """Create a new custom command"""
    # Check if command or alias already exists
    command = await get_command(session, name)
    if command is not None:
        raise DuplicateInsertException

    command = CustomCommand(name=name, indexed_name=clean_name(name), response=response)
    session.add(command)
    await session.commit()

    return command


async def create_alias(session: AsyncSession, command: str, alias: str) -> CustomCommandAlias:
    """Create an alias for a command"""
    # Check if the command exists
    command_instance = await get_command(session, command)
    if command_instance is None:
        raise NoResultFoundException

    # Check if the alias exists (either as an alias or as a name)
    if await get_command(session, alias) is not None:
        raise DuplicateInsertException

    alias_instance = CustomCommandAlias(alias=alias, indexed_alias=clean_name(alias), command=command_instance)
    session.add(alias_instance)
    await session.commit()

    return alias_instance


async def get_all_commands(session: AsyncSession) -> list[CustomCommand]:
    """Get a list of all commands"""
    statement = select(CustomCommand)
    return list((await session.execute(statement)).scalars().all())


async def get_command(session: AsyncSession, message: str) -> Optional[CustomCommand]:
    """Try to get a command out of a message"""
    # Search lowercase & without spaces
    message = clean_name(message)
    return (await get_command_by_name(session, message)) or (await get_command_by_alias(session, message))


async def get_command_by_name(session: AsyncSession, message: str) -> Optional[CustomCommand]:
    """Try to get a command by its name"""
    statement = select(CustomCommand).where(CustomCommand.indexed_name == message)
    return (await session.execute(statement)).scalar_one_or_none()


async def get_command_by_alias(session: AsyncSession, message: str) -> Optional[CustomCommand]:
    """Try to get a command by its alias"""
    statement = select(CustomCommandAlias).where(CustomCommandAlias.indexed_alias == message)
    alias = (await session.execute(statement)).scalar_one_or_none()
    if alias is None:
        return None

    return alias.command


async def edit_command(
    session: AsyncSession, original_name: str, new_name: Optional[str] = None, new_response: Optional[str] = None
) -> CustomCommand:
    """Edit an existing command"""
    # Check if the command exists
    command = await get_command(session, original_name)
    if command is None:
        raise NoResultFoundException

    if new_name is not None:
        command.name = new_name
    if new_response is not None:
        command.response = new_response

    session.add(command)
    await session.commit()

    return command