224 lines
6.7 KiB
Python
224 lines
6.7 KiB
Python
import inspect
|
|
import json
|
|
import uuid
|
|
from typing import Optional
|
|
|
|
from myutils.observable import NotObservableError, ObservableEvent, add_event_listener, remove_event_listener
|
|
|
|
from myfasthtml.core.constants import Routes, ROUTE_ROOT
|
|
|
|
|
|
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
|
|
self._htmx_extra = {}
|
|
self._bindings = []
|
|
self._ft = None
|
|
|
|
# register the command
|
|
CommandsManager.register(self)
|
|
|
|
def get_htmx_params(self):
|
|
return {
|
|
"hx-post": f"{ROUTE_ROOT}{Routes.Commands}",
|
|
"hx-swap": "outerHTML",
|
|
"hx-vals": {"c_id": f"{self.id}"},
|
|
} | self._htmx_extra
|
|
|
|
def execute(self, client_response: dict = None):
|
|
raise NotImplementedError
|
|
|
|
def htmx(self, target: Optional[str] = "this", swap="outerHTML", trigger=None):
|
|
# Note that the default value is the same than in get_htmx_params()
|
|
if target is None:
|
|
self._htmx_extra["hx-swap"] = "none"
|
|
elif target != "this":
|
|
self._htmx_extra["hx-target"] = target
|
|
|
|
if swap is None:
|
|
self._htmx_extra["hx-swap"] = "none"
|
|
elif swap != "outerHTML":
|
|
self._htmx_extra["hx-swap"] = swap
|
|
|
|
if trigger is not None:
|
|
self._htmx_extra["hx-trigger"] = trigger
|
|
|
|
return self
|
|
|
|
def bind_ft(self, ft):
|
|
"""
|
|
Update the FT with the command's HTMX parameters.
|
|
|
|
:param ft:
|
|
:return:
|
|
"""
|
|
self._ft = ft
|
|
htmx = self.get_htmx_params()
|
|
ft.attrs |= htmx
|
|
return ft
|
|
|
|
def bind(self, data):
|
|
"""
|
|
Attach a binding to the command.
|
|
When done, if a binding is triggered during the execution of the command,
|
|
the results of the binding will be passed to the command's execute() method.
|
|
:param data:
|
|
:return:
|
|
"""
|
|
if not hasattr(data, '_listeners'):
|
|
raise NotObservableError(
|
|
f"Object must be made observable with make_observable() before binding"
|
|
)
|
|
self._bindings.append(data)
|
|
|
|
# by default, remove the swap on the attached element when binding is used
|
|
self._htmx_extra["hx-swap"] = "none"
|
|
|
|
return self
|
|
|
|
@property
|
|
def url(self):
|
|
return f"{ROUTE_ROOT}{Routes.Commands}?c_id={self.id}"
|
|
|
|
def ajax_htmx_options(self):
|
|
return {
|
|
"url": self.url,
|
|
"target": self._htmx_extra.get("hx-target", "this"),
|
|
"swap": self._htmx_extra.get("hx-swap", "outerHTML"),
|
|
"values": {}
|
|
}
|
|
|
|
def get_ft(self):
|
|
return self._ft
|
|
|
|
def __str__(self):
|
|
return f"Command({self.name})"
|
|
|
|
|
|
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.callback_parameters = dict(inspect.signature(callback).parameters) if callback else {}
|
|
self.args = args
|
|
self.kwargs = kwargs
|
|
|
|
def _convert(self, key, value):
|
|
if key in self.callback_parameters:
|
|
param = self.callback_parameters[key]
|
|
if param.annotation == bool:
|
|
return value == "true"
|
|
elif param.annotation == int:
|
|
return int(value)
|
|
elif param.annotation == float:
|
|
return float(value)
|
|
elif param.annotation == list:
|
|
return value.split(",")
|
|
elif param.annotation == dict:
|
|
return json.loads(value)
|
|
return value
|
|
|
|
def ajax_htmx_options(self):
|
|
res = super().ajax_htmx_options()
|
|
if self.kwargs:
|
|
res["values"] |= self.kwargs
|
|
res["values"]["c_id"] = f"{self.id}" # cannot be overridden
|
|
return res
|
|
|
|
def execute(self, client_response: dict = None):
|
|
ret_from_bindings = []
|
|
|
|
def binding_result_callback(attr, old, new, results):
|
|
ret_from_bindings.extend(results)
|
|
|
|
for data in self._bindings:
|
|
add_event_listener(ObservableEvent.AFTER_PROPERTY_CHANGE, data, "", binding_result_callback)
|
|
|
|
new_kwargs = self.kwargs.copy()
|
|
if client_response:
|
|
for k, v in client_response.items():
|
|
if k in self.callback_parameters:
|
|
new_kwargs[k] = self._convert(k, v)
|
|
if 'client_response' in self.callback_parameters:
|
|
new_kwargs['client_response'] = client_response
|
|
|
|
ret = self.callback(*self.args, **new_kwargs)
|
|
|
|
for data in self._bindings:
|
|
remove_event_listener(ObservableEvent.AFTER_PROPERTY_CHANGE, data, "", binding_result_callback)
|
|
|
|
# Set the hx-swap-oob attribute on all elements returned by the callback
|
|
if isinstance(ret, (list, tuple)):
|
|
for r in ret[1:]:
|
|
if hasattr(r, 'attrs') and r.get("id", None) is not None:
|
|
r.attrs["hx-swap-oob"] = r.attrs.get("hx-swap-oob", "true")
|
|
|
|
if not ret_from_bindings:
|
|
return ret
|
|
|
|
if isinstance(ret, (list, tuple)):
|
|
return list(ret) + ret_from_bindings
|
|
else:
|
|
return [ret] + ret_from_bindings
|
|
|
|
|
|
class LambdaCommand(Command):
|
|
def __init__(self, delegate, name="LambdaCommand", description="Lambda Command"):
|
|
super().__init__(name, description, delegate)
|
|
self.htmx(target=None)
|
|
|
|
def execute(self, client_response: dict = None):
|
|
return self.callback(client_response)
|
|
|
|
|
|
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()
|