feat: implement workflow blocks to interact with third tools / airtable

alfred/ci.py

@@ -11,8 +11,9 @@
 @alfred.option('--front', '-f', help="run for frontend only", is_flag=True, default=False)
 @alfred.option('--docs', '-d', help="run for docs only", is_flag=True, default=False)
 @alfred.option('--back', '-b', help="run for backend only", is_flag=True, default=False)
+@alfred.option('--back-external', help="run for external test only", is_flag=True, default=False)
 @alfred.option('--e2e', '-e', help="run for end-to-end only", default=None)
-def ci(front, back, e2e, docs):
+def ci(front, back, back_external, e2e, docs):
     no_flags = (not front and not back and not e2e and not docs)
 
     if front or no_flags:
@@ -21,7 +22,9 @@ def ci(front, back, e2e, docs):
     if back or no_flags:
         alfred.invoke_command("ci.mypy")
         alfred.invoke_command("ci.ruff")
-        alfred.invoke_command("ci.pytest")
+        alfred.invoke_command("ci.pytest.backend")
+    if back_external or no_flags:
+        alfred.invoke_command("ci.pytest.backend_external")
     if docs or no_flags:
         alfred.invoke_command("npm.docs.test")
     if e2e:
@@ -39,9 +42,14 @@ def ci_ruff(fix):
     else:
         alfred.run("ruff check")
 
-@alfred.command("ci.pytest", help="run pytest on ./tests")
-def ci_test():
-    os.chdir("tests")
+@alfred.command("ci.pytest.backend", help="run pytest on ./tests")
+def ci_pytest_backend():
+    os.chdir("tests/backend")
+    alfred.run("pytest")
+
+@alfred.command("ci.pytest.backend_external", help="run automatic test on external integrations")
+def ci_test_backend_external():
+    os.chdir("tests/backend_external")
     alfred.run("pytest")
 
 

pyproject.toml

@@ -78,6 +78,7 @@ redis = "^5.2.1"
 
 [tool.poetry.group.dev.dependencies]
 types-python-dateutil = "^2.9.0.20240316"
+python-dotenv = "^1.0.1"
 
 [tool.poetry.scripts]
 writer = 'writer.command_line:main'

src/ui/src/builder/sidebar/BuilderSidebarToolkit.vue

@@ -61,6 +61,7 @@ const displayedCategories = [
 	"Writer",
 	"Logic",
 	"Other",
+	"Third parts",
 ];
 
 const activeToolkit = computed(() => {

src/writer/blocks/__init__.py

@@ -1,4 +1,6 @@
 from writer.blocks.addtostatelist import AddToStateList
+from writer.blocks.airtablemanipulaterecord import AirtableManipulateRecord
+from writer.blocks.airtablequeryrecords import AirtableQueryRecords
 from writer.blocks.calleventhandler import CallEventHandler
 from writer.blocks.foreach import ForEach
 from writer.blocks.httprequest import HTTPRequest
@@ -30,4 +32,6 @@
 AddToStateList.register("workflows_addtostatelist")
 ReturnValue.register("workflows_returnvalue")
 WriterInitChat.register("workflows_writerinitchat")
-WriterAddToKG.register("workflows_writeraddtokg")
+WriterAddToKG.register("workflows_airtablequeryrecords")
+AirtableQueryRecords.register("workflows_airtablequeryrecords")
+AirtableManipulateRecord.register("workflows_airtablemanipulaterecord")

src/writer/blocks/airtablemanipulaterecord.py

@@ -0,0 +1,115 @@
+import requests
+
+from writer.abstract import register_abstract_template
+from writer.blocks.base_block import WorkflowBlock
+from writer.ss_types import AbstractTemplate
+
+
+class AirtableManipulateRecord(WorkflowBlock):
+
+    @classmethod
+    def register(cls, type: str):
+        super(AirtableManipulateRecord, cls).register(type)
+        register_abstract_template(type, AbstractTemplate(
+            baseType="workflows_node",
+            writer={
+                "name": "Airtable - Manipulate record",
+                "description": "Performs operations on Airtable records.",
+                "category": "Third parts",
+                "fields": {
+                    "apiKey": {
+                        "name": "API Key",
+                        "type": "Text",
+                        "desc": "Your Airtable API key."
+                    },
+                    "base": {
+                        "name": "Base",
+                        "type": "Text",
+                        "desc": "The ID of the base containing the table."
+                    },
+                    "table": {
+                        "name": "Table",
+                        "type": "Text",
+                        "desc": "The name of the table to manipulate."
+                    },
+                    "operation": {
+                        "name": "Operation",
+                        "type": "Text",
+                        "options": {
+                            "create": "Create",
+                            "update": "Update",
+                            "remove": "Remove"
+                        },
+                        "default": "create"
+                    },
+                    "recordId": {
+                        "name": "Record ID",
+                        "type": "Text",
+                        "desc": "The ID of the record to update or remove. Not required for create operations."
+                    },
+                    "fields": {
+                        "name": "Fields",
+                        "type": "Text",
+                        "control": "Textarea",
+                        "desc": "JSON string of fields and values to set.",
+                        "default": "{}"
+                    },
+                },
+                "outs": {
+                    "success": {
+                        "name": "Success",
+                        "description": "The operation was successful.",
+                        "style": "success",
+                    },
+                    "error": {
+                        "name": "Error",
+                        "description": "The operation failed.",
+                        "style": "error",
+                    },
+                },
+            }
+        ))
+
+    def run(self):
+        try:
+            api_key = self._get_field("apiKey", required=True)
+            base = self._get_field("base", required=True)
+            table = self._get_field("table", required=True)
+            operation = self._get_field("operation", required=True)
+            record_id = self._get_field("recordId", default_field_value="")
+            payload = self._get_field("fields", as_json=True, default_field_value="{}")
+
+            headers = {
+                "Authorization": f"Bearer {api_key}",
+                "Content-Type": "application/json"
+            }
+
+            response = None
+
+            if operation == "create":
+                url = f"https://api.airtable.com/v0/{base}/{table}"
+                response = requests.post(url, headers=headers, json={"fields": payload})
+            elif operation == "update":
+                url = f"https://api.airtable.com/v0/{base}/{table}/{record_id}"
+                if not record_id:
+                    raise ValueError("Record ID is required for update operations.")
+                response = requests.patch(url, headers=headers, json={"fields": payload})
+            elif operation == "remove":
+                url = f"https://api.airtable.com/v0/{base}/{table}/{record_id}"
+                if not record_id:
+                    raise ValueError("Record ID is required for remove operations.")
+                response = requests.delete(url, headers=headers)
+
+            if response and response.ok:
+                self.result = {
+                    "id": response.json().get("id"),
+                    "fields": response.json().get("fields")
+                }
+                self.outcome = "success"
+            else:
+                self.outcome = "error"
+                response.raise_for_status()
+
+        except BaseException as e:
+            self.outcome = "error"
+            raise e

src/writer/blocks/airtablequeryrecords.py

@@ -0,0 +1,95 @@
+import requests
+
+from writer.abstract import register_abstract_template
+from writer.blocks.base_block import WorkflowBlock
+from writer.ss_types import AbstractTemplate
+
+
+class AirtableQueryRecords(WorkflowBlock):
+
+    @classmethod
+    def register(cls, type: str):
+        super(AirtableQueryRecords, cls).register(type)
+        register_abstract_template(type, AbstractTemplate(
+            baseType="workflows_node",
+            writer={
+                "name": "Airtable - Query records",
+                "description": "Queries records from an Airtable table based on a provided formula.",
+                "category": "Third parts",
+                "fields": {
+                    "apiKey": {
+                        "name": "API Key",
+                        "type": "Text",
+                        "desc": "Your Airtable API key."
+                    },
+                    "base": {
+                        "name": "Base",
+                        "type": "Text",
+                        "desc": "The ID of the base containing the table."
+                    },
+                    "table": {
+                        "name": "Table",
+                        "type": "Text",
+                        "desc": "The name of the table to query."
+                    },
+                    "formula": {
+                        "name": "Formula",
+                        "type": "Text",
+                        "desc": "The Airtable formula to filter records."
+                    },
+                    "sortFields": {
+                        "name": "Sort Fields",
+                        "type": "Text",
+                        "desc": "Fields to sort by, formatted as a JSON array of objects with 'field' and 'direction'."
+                    },
+                },
+                "outs": {
+                    "success": {
+                        "name": "Success",
+                        "description": "The operation was successful.",
+                        "style": "success",
+                    },
+                    "error": {
+                        "name": "Error",
+                        "description": "The operation failed.",
+                        "style": "error",
+                    },
+                },
+            }
+        ))
+
+    def run(self):
+        try:
+            api_key = self._get_field("apiKey", required=True)
+            base = self._get_field("base", required=True)
+            table = self._get_field("table", required=True)
+            formula = self._get_field("formula")
+            sort_fields = self._get_field("sortFields", as_json=True, default_field_value="[]")
+
+            headers = {
+                "Authorization": f"Bearer {api_key}"
+            }
+
+            params = {
+                "filterByFormula": formula,
+            }
+
+            for i, sort_item in enumerate(sort_fields):
+                params[f"sort[{i}][field]"] = sort_item["field"]
+                params[f"sort[{i}][direction]"] = sort_item["direction"]
+
+            url = f"https://api.airtable.com/v0/{base}/{table}"
+            response = requests.get(url, headers=headers, params=params)
+
+            if response.ok:
+                records = response.json().get("records", [])
+                self.result = [{"recordId": rec.get("id"), "fields": rec.get("fields")} for rec in records]
+                self.outcome = "success"
+            else:
+                self.outcome = "error"
+                response.raise_for_status()
+
+        except BaseException as e:
+            self.outcome = "error"
+            raise e
+

tests/backend_external/README.md

@@ -0,0 +1,3 @@
+These tests are dedicated to the functional validation of external integrations. They require having the necessary permissions to access external resources. 
+
+In general, these tests are exclusively performed by the CI.

tests/backend_external/blocks/conftest.py

@@ -0,0 +1,56 @@
+from typing import Dict
+
+import pytest
+from dotenv import load_dotenv
+from writer.core import WriterSession, WriterState
+from writer.core_ui import Branch, Component, ComponentTree, ComponentTreeBranch
+from writer.workflows import WorkflowRunner
+
+load_dotenv()
+
+class BlockTesterMockSession(WriterSession):
+
+    def __init__(self):
+        self.session_state = WriterState({})
+        self.bmc_branch = ComponentTreeBranch(Branch.bmc)
+        component_tree = ComponentTree([self.bmc_branch])
+        self.session_component_tree = component_tree
+
+    def add_fake_component(self, content={}, id="fake_id", type="fake_type"):
+        self.bmc_branch.attach(Component(id=id, type=type, content=content))
+
+
+class BlockTesterMockWorkflowRunner(WorkflowRunner):
+
+    def __init__(self, session):
+        super().__init__(session)
+
+    def run_branch(self, component_id: str, base_outcome_id: str, execution_environment: Dict, title: str):
+        return f"Branch run {component_id} {base_outcome_id}"
+
+    def run_workflow_by_key(self, workflow_key: str, execution_environment: Dict):
+        payload = execution_environment.get("payload")
+        if "env_injection_test" in payload:
+            return payload.get("env_injection_test")
+        if workflow_key == "workflow1":
+            return 1
+        if workflow_key == "workflowDict":
+            return { "a": "b" }
+        if workflow_key == "duplicator":
+            return payload.get("item") * 2
+        if workflow_key == "showId":
+            return payload.get("itemId")
+        if workflow_key == "boom":
+            return 1/0
+        raise ValueError("Workflow not found.")
+
+
+@pytest.fixture
+def session():
+    yield BlockTesterMockSession()
+
+
+@pytest.fixture
+def runner(session):
+    yield BlockTesterMockWorkflowRunner(session)
+