123 lines
3.4 KiB
Python
123 lines
3.4 KiB
Python
import logging
|
|
import uuid
|
|
from typing import Optional
|
|
|
|
from fasthtml.fastapp import fast_app
|
|
|
|
from myfasthtml.core.constants import Routes, ROUTE_ROOT
|
|
from myfasthtml.core.utils import mount_if_not_exists
|
|
|
|
commands_app, commands_rt = fast_app()
|
|
logger = logging.getLogger("Commands")
|
|
|
|
|
|
class BaseCommand:
|
|
"""
|
|
Represents the base command class for defining executable actions.
|
|
|
|
This class serves as a foundation for commands that can be registered,
|
|
executed, and utilized within a system. Each command has a unique identifier,
|
|
a name, and a description. Commands should override the `execute` method
|
|
to provide specific functionality.
|
|
|
|
:ivar id: A unique identifier for the command.
|
|
:type id: uuid.UUID
|
|
:ivar name: The name of the command.
|
|
:type name: str
|
|
:ivar description: A brief description of the command's functionality.
|
|
:type description: str
|
|
"""
|
|
|
|
def __init__(self, name, description):
|
|
self.id = uuid.uuid4()
|
|
self.name = name
|
|
self.description = description
|
|
|
|
# register the command
|
|
CommandsManager.register(self)
|
|
|
|
def get_htmx_params(self):
|
|
return {
|
|
"hx-post": f"{ROUTE_ROOT}{Routes.Commands}",
|
|
"hx-vals": f'{{"c_id": "{self.id}"}}',
|
|
}
|
|
|
|
def execute(self):
|
|
raise NotImplementedError
|
|
|
|
|
|
class Command(BaseCommand):
|
|
"""
|
|
Represents a command that encapsulates a callable action with parameters.
|
|
|
|
This class is designed to hold a defined action (callback) alongside its arguments
|
|
and keyword arguments.
|
|
|
|
:ivar name: The name of the command.
|
|
:type name: str
|
|
:ivar description: A brief description of the command.
|
|
:type description: str
|
|
:ivar callback: The function or callable to be executed.
|
|
:type callback: Callable
|
|
:ivar args: Positional arguments to be passed to the callback.
|
|
:type args: tuple
|
|
:ivar kwargs: Keyword arguments to be passed to the callback.
|
|
:type kwargs: dict
|
|
"""
|
|
|
|
def __init__(self, name, description, callback, *args, **kwargs):
|
|
super().__init__(name, description)
|
|
self.callback = callback
|
|
self.args = args
|
|
self.kwargs = kwargs
|
|
|
|
def execute(self):
|
|
return self.callback(*self.args, **self.kwargs)
|
|
|
|
def __str__(self):
|
|
return f"Command({self.name})"
|
|
|
|
|
|
class CommandsManager:
|
|
commands = {}
|
|
|
|
@staticmethod
|
|
def register(command: BaseCommand):
|
|
CommandsManager.commands[str(command.id)] = command
|
|
|
|
@staticmethod
|
|
def get_command(command_id: str) -> Optional[BaseCommand]:
|
|
return CommandsManager.commands.get(command_id)
|
|
|
|
@staticmethod
|
|
def reset():
|
|
return CommandsManager.commands.clear()
|
|
|
|
|
|
@commands_rt(Routes.Commands)
|
|
def post(session: str, c_id: str):
|
|
"""
|
|
Default routes for all commands.
|
|
:param session:
|
|
:param c_id:
|
|
:return:
|
|
"""
|
|
logger.debug(f"Entering {Routes.Commands} with {session=}, {c_id=}")
|
|
command = CommandsManager.get_command(c_id)
|
|
if command:
|
|
return command.execute()
|
|
|
|
raise ValueError(f"Command with ID '{c_id}' not found.")
|
|
|
|
|
|
def mount_commands(app):
|
|
"""
|
|
Mounts the commands_app to the given application instance if the route does not already exist.
|
|
|
|
:param app: The application instance to which the commands_app will be mounted.
|
|
:type app: Any
|
|
:return: Returns the result of the mount operation performed by mount_if_not_exists.
|
|
:rtype: Any
|
|
"""
|
|
return mount_if_not_exists(app, ROUTE_ROOT, commands_app)
|