Adding workflow management

I

src/components/workflows/WorkflowsApp.py

@@ -0,0 +1,26 @@
+import json
+import logging
+
+from fasthtml.fastapp import fast_app
+
+from components.workflows.constants import Routes
+from core.instance_manager import InstanceManager, debug_session
+
+logger = logging.getLogger("WorkflowsApp")
+
+repositories_app, rt = fast_app()
+
+
+@rt(Routes.AddWorkflow)
+def get(session, _id: str):
+  logger.debug(f"Entering {Routes.AddWorkflow} with args {debug_session(session)}, {_id=}")
+  instance = InstanceManager.get(session, _id)
+  return instance.request_new_workflow()
+
+
+@rt(Routes.AddWorkflow)
+def post(session, _id: str, tab_id: str, form_id: str, name: str, tab_boundaries: str):
+  logger.debug(
+    f"Entering {Routes.AddWorkflow} with args {debug_session(session)}, {_id=}, {tab_id=}, {form_id=}, {name=}, {tab_boundaries=}")
+  instance = InstanceManager.get(session, _id)
+  return instance.add_new_workflow(tab_id, form_id, name, json.loads(tab_boundaries))

src/components/workflows/commands.py

@@ -0,0 +1,22 @@
+from components.BaseCommandManager import BaseCommandManager
+from components.workflows.constants import Routes, ROUTE_ROOT
+
+
+class WorkflowsCommandManager(BaseCommandManager):
+  def __init__(self, owner):
+    super().__init__(owner)
+  
+  def request_add_workflow(self):
+    return {
+        "hx-get": f"{ROUTE_ROOT}{Routes.AddWorkflow}",
+        "hx-target": f"#{self._owner.tabs_manager.get_id()}",
+        "hx-swap": "outerHTML",
+        "hx-vals": f'{{"_id": "{self._id}"}}',
+    }
+  
+  def add_workflow(self, tab_id: str):
+    return {
+        "hx-post": f"{ROUTE_ROOT}{Routes.AddWorkflow}",
+        "hx-target": f"#w_{self._id}",
+        "hx-vals": f'js:{{"_id": "{self._id}", "tab_id": "{tab_id}", "tab_boundaries": getTabContentBoundaries("{self._owner.tabs_manager.get_id()}")}}',
+    }

src/components/workflows/components/Workflows.py

@@ -0,0 +1,100 @@
+import logging
+
+from fasthtml.components import *
+
+from assets.icons import icon_add_regular
+from components.BaseComponent import BaseComponentSingleton
+from components.form.components.MyForm import MyForm, FormField
+from components.workflows.commands import WorkflowsCommandManager
+from components.workflows.constants import WORKFLOWS_INSTANCE_ID
+from components.workflows.db_management import WorkflowsDbManager
+from components_helpers import mk_ellipsis, mk_icon
+from core.instance_manager import InstanceManager
+
+logger = logging.getLogger("Workflows")
+
+
+class Workflows(BaseComponentSingleton):
+  COMPONENT_INSTANCE_ID = WORKFLOWS_INSTANCE_ID
+  
+  def __init__(self, session, _id, settings_manager=None, tabs_manager=None):
+    super().__init__(session, _id, settings_manager, tabs_manager)
+    self.commands = WorkflowsCommandManager(self)
+    self.db = WorkflowsDbManager(session, settings_manager)
+  
+  def request_new_workflow(self):
+    # request for a new tab_id
+    new_tab_id = self.tabs_manager.request_new_tab_id()
+    
+    # create a new form to ask for the details of the new database
+    add_workflow_form = self._mk_add_workflow_form(new_tab_id)
+    
+    # create and display the form in a new tab
+    self.tabs_manager.add_tab("Add Workflow", add_workflow_form, tab_id=new_tab_id)
+    return self.tabs_manager
+  
+  def add_new_workflow(self, tab_id: str, form_id: str, workflow_name: str, tab_boundaries: dict):
+    """
+
+    :param tab_id: tab id where the table content will be displayed (and where the form was displayed)
+    :param form_id: form used to give the repository name (to be used in case of error)
+    :param workflow_name: new workflow name
+    :param tab_boundaries: tab boundaries
+    :return:
+    """
+    try:
+      # Add the new repository and its default table to the list of repositories
+      self.db.add_workflow(workflow_name)
+      
+      # update the tab content with table content
+      self.tabs_manager.set_tab_content(tab_id,
+                                        self._get_workflow_designer(workflow_name, tab_boundaries),
+                                        title=workflow_name,
+                                        key=workflow_name,
+                                        active=True)
+      
+      return self._mk_workflows(), self.tabs_manager.refresh()
+    
+    except ValueError as ex:
+      logger.error(f"  Workflow '{workflow_name}' already exists.")
+      add_repository_form = InstanceManager.get(self._session, form_id)
+      add_repository_form.set_error(ex)
+      
+      return self.tabs_manager.refresh()
+  
+  def __ft__(self):
+    return Div(
+      Div(cls="divider"),
+      Div(
+        mk_ellipsis("Workflows", cls="text-sm font-medium mb-1"),
+        mk_icon(icon_add_regular,
+                size=16,
+                tooltip="Add Workflow",
+                cls="ml-2 mmt-visible-on-hover",
+                **self.commands.request_add_workflow()),
+        cls="flex"
+      ),
+      self._mk_workflows(),
+      id=f"{self._id}"
+    )
+  
+  def _get_workflow_designer(self, workflow_name: str, tab_boundaries: dict):
+    return Div(f"Workflow Designer for {workflow_name}")
+  
+  def _mk_add_workflow_form(self, tab_id: str):
+    return InstanceManager.get(self._session, MyForm.create_component_id(self._session), MyForm,
+                               title="Add Workflow",
+                               fields=[FormField("name", 'Workflow Name', 'input')],
+                               htmx_request=self.commands.add_workflow(tab_id),
+                               )
+  
+  def _mk_workflow(self, workflow_name: str, selected: bool):
+    return mk_ellipsis(workflow_name, cls="text-sm")
+  
+  def _mk_workflows(self, oob=False):
+    return Div(
+      *[self._mk_workflow(workflow_name, workflow_name == self.db.get_selected_workflow())
+        for workflow_name in self.db.get_workflows()],
+      id=f"w_{self._id}",
+      hx_swap_oob="true" if oob else None,
+    )

src/components/workflows/constants.py

@@ -0,0 +1,8 @@
+WORKFLOWS_INSTANCE_ID = "__Workflows__"
+WORKFLOWS_SETTINGS_ENTRY = "Workflows"
+ROUTE_ROOT = "/workflows"
+
+
+class Routes:
+  AddWorkflow = "/add-workflow"
+  SelectWorkflow = "/select-workflow"

src/components/workflows/db_management.py

@@ -0,0 +1,99 @@
+import logging
+from dataclasses import dataclass, field
+
+from components.workflows.constants import WORKFLOWS_SETTINGS_ENTRY
+from core.settings_management import SettingsManager
+
+logger = logging.getLogger("WorkflowsSettings")
+
+
+@dataclass
+class WorkflowsSettings:
+  workflows: list[str] = field(default_factory=list)
+  selected_workflow: str = None
+
+
+class WorkflowsDbManager:
+  def __init__(self, session: dict, settings_manager: SettingsManager):
+    self.session = session
+    self.settings_manager = settings_manager
+  
+  def add_workflow(self, workflow_name: str):
+    settings = self._get_settings()
+    
+    if not workflow_name:
+      raise ValueError("Workflow name cannot be empty.")
+    
+    if workflow_name in settings.workflows:
+      raise ValueError(f"Workflow '{workflow_name}' already exists.")
+    
+    settings.workflows.append(workflow_name)
+    self.settings_manager.save(self.session, WORKFLOWS_SETTINGS_ENTRY, settings)
+    return True
+  
+  def get_workflow(self, workflow_name: str):
+    if not workflow_name:
+      raise ValueError("Workflow name cannot be empty.")
+    
+    settings = self._get_settings()
+    if workflow_name not in settings.workflows:
+      raise ValueError(f"Workflow '{workflow_name}' does not exist.")
+    
+    return next(filter(lambda r: r == workflow_name, settings.workflows))
+  
+  # def modify_workflow(self, old_workflow_name, new_workflow_name: str, tables: list[str]):
+  #   if not old_workflow_name or not new_workflow_name:
+  #     raise ValueError("Workflow name cannot be empty.")
+  #
+  #   settings = self._get_settings()
+  #   for workflow in settings.workflows:
+  #     if workflow == old_workflow_name:
+  #       workflow.name = new_workflow_name
+  #       workflow.tables = tables
+  #
+  #       self.settings_manager.save(self.session, workflows_SETTINGS_ENTRY, settings)
+  #       return workflow
+  #
+  #   else:
+  #     raise ValueError(f"workflow '{old_workflow_name}' not found.")
+  
+  def remove_workflow(self, workflow_name):
+    if not workflow_name:
+      raise ValueError("Workflow name cannot be empty.")
+    
+    settings = self._get_settings()
+    if workflow_name not in settings.workflows:
+      raise ValueError(f"workflow '{workflow_name}' does not exist.")
+    
+    settings.workflows.remove(workflow_name)
+    self.settings_manager.save(self.session, WORKFLOWS_SETTINGS_ENTRY, settings)
+    return True
+  
+  def exists_workflow(self, workflow_name):
+    if not workflow_name:
+      raise ValueError("workflow name cannot be empty.")
+    
+    settings = self._get_settings()
+    return workflow_name in settings.workflows
+  
+  def get_workflows(self):
+    return self._get_settings().workflows
+  
+  def select_workflow(self, workflow_name: str):
+    """
+    Select and save the specified workflow name in the current session's settings.
+
+    :param workflow_name: The name of the workflow to be selected and stored.
+    :type workflow_name: str
+    :return: None
+    """
+    settings = self._get_settings()
+    settings.selected_workflow = workflow_name
+    self.settings_manager.save(self.session, WORKFLOWS_SETTINGS_ENTRY, settings)
+  
+  def get_selected_workflow(self):
+    settings = self._get_settings()
+    return settings.selected_workflow
+  
+  def _get_settings(self):
+    return self.settings_manager.load(self.session, WORKFLOWS_SETTINGS_ENTRY, default=WorkflowsSettings())