jinja_template block

.github/workflows/integration-tests.yml

@@ -18,7 +18,7 @@ jobs:
       - name: Setup Python
         uses: actions/setup-python@v4
         with:
-          python-version: "3.x"
+          python-version: "3.8"
 
       - name: Build and Install CLI
         run: |

.github/workflows/publish-pypi.yml

@@ -19,7 +19,7 @@ jobs:
       - name: Set up Python
         uses: actions/setup-python@v4
         with:
-          python-version: "3.x"
+          python-version: "3.8"
       - name: Install dependencies
         run: |
           python -m pip install --upgrade pip

.github/workflows/unit-tests.yml

@@ -18,7 +18,7 @@ jobs:
       - name: Setup Python
         uses: actions/setup-python@v4
         with:
-          python-version: "3.x"
+          python-version: "3.8"
 
       - name: Install dependencies
         run: |

.vscode/extensions.json

@@ -1,8 +1,11 @@
 {
   "recommendations": [
-    "ms-python.python",
     "esbenp.prettier-vscode",
+    "ms-python.autopep8",
+    "ms-python.isort",
+    "ms-python.python",
+    "njpwerner.autodocstring",
     "streetsidesoftware.code-spell-checker",
-    "njpwerner.autodocstring"
+    "tamasfe.even-better-toml"
   ]
 }

core/pyproject.toml

@@ -18,13 +18,16 @@ classifiers = [
 
 [tool.poetry.dependencies]
 certifi = "^2023.7.22"
-pysqlite3-binary = { version = "^0.4.0", platform = "linux" }
+pysqlite3-binary = { version = "^0.5.0", platform = "linux" }
 cryptography = ">=39.0.1"
+Jinja2 = "^3.1.2"
 jmespath = "^1.0.0"
 jsonschema = "^4.4.0"
 orjson = "^3.8.7"
+prometheus-client = "^0.16.0"
 python = "^3.7"
 PyYAML = "^6.0"
+sqlglot = "^10.4.3"
 
 mock = { version = "^4.0.3", optional = true }
 pytest = { version = "^7.1.2", optional = true }
@@ -40,25 +43,23 @@ azure-eventhub-checkpointstoreblob-aio = { version = "^1.1.4", optional = true }
 aiohttp = { version = "^3.8.4", optional = true }
 cassandra-driver = { version = "^3.25.0", optional = true }
 fastparquet = { version = "^2023.2.0", optional = true, platform = "python>=3.8" }
-psycopg2-binary = { version="^2.9.5", optional = true}
 oracledb = { version="^1.2.2", optional = true}
-prometheus-client = "^0.16.0"
+psycopg2-binary = { version="^2.9.5", optional = true}
 pymssql = { version="^2.2.7", optional = true }
 PyMySQL = { version="^1.0.2", optional = true }
 redis = { version = "^4.3.5", optional = true }
 SQLAlchemy = { version = "^2.0.4", optional = true }
-sqlglot = "^10.4.3"
 
 [tool.poetry.extras]
 azure = ["azure-eventhub", "azure-eventhub-checkpointstoreblob-aio"]
 cassandra = ["cassandra-driver"]
+http = ["aiohttp"]
 mysql = ["PyMySQL", "SQLAlchemy"]
 oracle = ["oracledb", "SQLAlchemy"]
 parquet = ["fastparquet"]
 pg = ["psycopg2-binary", "SQLAlchemy"]
 redis = ["redis"]
 sqlserver = ["pymssql", "SQLAlchemy"]
-http = ["aiohttp"]
 
 test = [
         "cassandra-driver",

core/src/datayoga_core/blocks/jinja_template/block.py

@@ -0,0 +1,41 @@
+import logging
+from abc import ABCMeta
+from typing import Any, Dict, List, Optional
+
+import jinja2
+from datayoga_core import utils
+from datayoga_core.block import Block as DyBlock
+from datayoga_core.context import Context
+from datayoga_core.result import BlockResult, Result, Status
+
+logger = logging.getLogger("dy")
+
+
+class Block(DyBlock, metaclass=ABCMeta):
+
+    def init(self, context: Optional[Context] = None):
+        logger.debug(f"Initializing {self.get_block_name()}")
+        self.field = self.properties["field"]
+        self.template = jinja2.Template(self.properties["template"])
+
+    async def run(self, data: List[Dict[str, Any]]) -> BlockResult:
+        logger.debug(f"Running {self.get_block_name()}")
+
+        block_result = BlockResult()
+        field_path = utils.split_field(self.field)
+
+        for _, row in enumerate(data):
+            obj = row
+            # handle nested fields. in that case, the obj points at the nested entity
+            for key in field_path[:-1]:
+                key = utils.unescape_field(key)
+                obj = obj.setdefault(key, {})  # Setdefault creates missing nested keys as dictionaries
+
+            try:
+                # assign the new values
+                obj[utils.unescape_field(field_path[-1:][0])] = self.template.render(**row)
+                block_result.processed.append(Result(Status.SUCCESS, payload=row))
+            except Exception as e:
+                block_result.rejected.append(Result(status=Status.REJECTED, payload=row, message=f"{e}"))
+
+        return block_result

core/src/datayoga_core/blocks/jinja_template/block.schema.json

@@ -0,0 +1,27 @@
+{
+  "title": "jinja_template",
+  "description": "Apply Jinja template to a field",
+  "type": "object",
+  "properties": {
+    "field": {
+      "description": "Field",
+      "type": "string"
+    },
+    "template": {
+      "description": "Jinja Template",
+      "type": "string"
+    }
+  },
+  "additionalProperties": false,
+  "required": ["field", "template"],
+  "examples": [
+    {
+      "field": "name.full_name",
+      "template": "{{ name.fname }} {{ name.lname }}"
+    },
+    {
+      "field": "name.fname_upper",
+      "template": "{{ name.fname | upper }}"
+    }
+  ]
+}

core/src/datayoga_core/blocks/jinja_template/tests/test_jinja_template.py

@@ -0,0 +1,73 @@
+import pytest
+from datayoga_core import utils
+from datayoga_core.blocks.jinja_template.block import Block
+
+
+@pytest.mark.asyncio
+async def test_jinja_template():
+    """Test case for applying Jinja template to a field."""
+    block = Block({
+        "field": "full_name",
+        "template": "{{ fname }} {{ lname }}"
+    })
+    block.init()
+
+    assert await block.run([
+        {"fname": "john", "lname": "doe"}
+    ]) == utils.all_success([
+        {"fname": "john", "lname": "doe", "full_name": "john doe"}
+    ])
+
+
+@pytest.mark.asyncio
+async def test_jinja_template_nested_field():
+    """Test case for applying Jinja template to a nested field."""
+    block = Block({
+        "field": "name.full_name",
+        "template": "{{ name.fname }} {{ name.lname }}"
+    })
+    block.init()
+
+    assert await block.run([
+        {"name": {"fname": "john", "lname": "doe"}}
+    ]) == utils.all_success([
+        {"name": {"fname": "john", "lname": "doe", "full_name": "john doe"}}
+    ])
+
+
+@pytest.mark.asyncio
+async def test_jinja_template_with_dot():
+    """Test case for adding a field with a dot in the name using Jinja template."""
+    block = Block({
+        "field": "name\.full_name",
+        "template": "{{ fname }} {{ lname }}"
+    })
+    block.init()
+
+    assert await block.run([
+        {"fname": "john", "lname": "doe"}
+    ]) == utils.all_success([
+        {"fname": "john", "lname": "doe", "name.full_name": "john doe"}
+    ])
+
+
+@pytest.mark.asyncio
+async def test_jinja_template_missing_field():
+    """Test cases for handling missing fields in Jinja template."""
+    block = Block({
+        "field": "greeting",
+        "template": "Hello {{ fname | upper }} {{ lname }}!"
+    })
+    block.init()
+
+    assert await block.run([
+        {"lname": "doe"},
+        {"fname": "john"},
+        {"fname": "john", "lname": "doe"},
+        {}
+    ]) == utils.all_success([
+        {"lname": "doe", "greeting": "Hello  doe!"},
+        {"fname": "john", "greeting": "Hello JOHN !"},
+        {"fname": "john", "lname": "doe", "greeting": "Hello JOHN doe!"},
+        {"greeting": "Hello  !"}
+    ])

core/src/datayoga_core/blocks/redis/write/block.py

@@ -49,7 +49,7 @@ async def run(self, data: List[Dict[str, Any]]) -> BlockResult:
                 if isinstance(result, Exception):
                     block_result.rejected.append(Result(Status.REJECTED, message=f"{result}", payload=record))
                 else:
-                    block_result.rejected.append(Result(Status.SUCCESS, payload=record))
+                    block_result.processed.append(Result(Status.SUCCESS, payload=record))
         except redis.exceptions.ConnectionError as e:
             raise ConnectionError(e)
 

docs/reference/blocks/jinja_template.md

@@ -0,0 +1,35 @@
+---
+parent: Blocks
+grand_parent: Reference
+---
+
+# jinja\_template
+
+Apply Jinja template to a field
+
+
+**Properties**
+
+|Name|Type|Description|Required|
+|----|----|-----------|--------|
+|**field**|`string`|Field<br/>|yes|
+|**template**|`string`|Jinja Template<br/>|yes|
+
+**Additional Properties:** not allowed  
+**Example**
+
+```yaml
+field: name.full_name
+template: '{{ name.fname }} {{ name.lname }}'
+
+```
+
+**Example**
+
+```yaml
+field: name.fname_upper
+template: '{{ name.fname | upper }}'
+
+```
+
+