Added Commands Management

src/myfasthtml/core/commands.py

@@ -0,0 +1,122 @@
+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)