Add gzip compression nodes

Author: georgi

docs/nodetool_compression.md

@@ -0,0 +1,39 @@
+---
+layout: default
+title: nodetool.compression
+parent: Nodes
+has_children: false
+nav_order: 2
+---
+
+# nodetool.nodes.nodetool.compression
+
+Nodes for compressing and decompressing data with gzip.
+
+## GzipCompress
+
+Compress bytes using gzip.
+
+Use cases:
+- Reduce size of binary data
+- Store assets in compressed form
+- Prepare data for network transfer
+
+**Tags:** gzip, compress, bytes
+
+**Fields:**
+- **data**: Data to compress (bytes)
+
+## GzipDecompress
+
+Decompress gzip data.
+
+Use cases:
+- Restore compressed files
+- Read data from gzip archives
+- Process network payloads
+
+**Tags:** gzip, decompress, bytes
+
+**Fields:**
+- **data**: Gzip data to decompress (bytes)

src/nodetool/dsl/nodetool/compression.py

@@ -0,0 +1,43 @@
+from pydantic import BaseModel, Field
+import typing
+from typing import Any
+import nodetool.metadata.types
+import nodetool.metadata.types as types
+from nodetool.dsl.graph import GraphNode
+
+
+class GzipCompress(GraphNode):
+    """
+    Compress bytes using gzip.
+    gzip, compress, bytes
+
+    Use cases:
+    - Reduce size of binary data
+    - Store assets in compressed form
+    - Prepare data for network transfer
+    """
+
+    data: bytes | None | GraphNode | tuple[GraphNode, str] = Field(default=None, description='Data to compress')
+
+    @classmethod
+    def get_node_type(cls): return "nodetool.compression.GzipCompress"
+
+
+
+class GzipDecompress(GraphNode):
+    """
+    Decompress gzip data.
+    gzip, decompress, bytes
+
+    Use cases:
+    - Restore compressed files
+    - Read data from gzip archives
+    - Process network payloads
+    """
+
+    data: bytes | None | GraphNode | tuple[GraphNode, str] = Field(default=None, description='Gzip data to decompress')
+
+    @classmethod
+    def get_node_type(cls): return "nodetool.compression.GzipDecompress"
+
+

src/nodetool/nodes/nodetool/compression.py

@@ -0,0 +1,45 @@
+import gzip
+from pydantic import Field
+from nodetool.workflows.base_node import BaseNode
+from nodetool.workflows.processing_context import ProcessingContext
+
+
+class GzipCompress(BaseNode):
+    """
+    Compress bytes using gzip.
+    gzip, compress, bytes
+
+    Use cases:
+    - Reduce size of binary data
+    - Store assets in compressed form
+    - Prepare data for network transfer
+    """
+
+    data: bytes | None = Field(default=None, description="Data to compress")
+
+    async def process(self, context: ProcessingContext) -> bytes:
+        if not self.data:
+            raise ValueError("data cannot be empty")
+        return gzip.compress(self.data)
+
+
+class GzipDecompress(BaseNode):
+    """
+    Decompress gzip data.
+    gzip, decompress, bytes
+
+    Use cases:
+    - Restore compressed files
+    - Read data from gzip archives
+    - Process network payloads
+    """
+
+    data: bytes | None = Field(
+        default=None,
+        description="Gzip data to decompress",
+    )
+
+    async def process(self, context: ProcessingContext) -> bytes:
+        if not self.data:
+            raise ValueError("data cannot be empty")
+        return gzip.decompress(self.data)

tests/nodetool/test_gzip_nodes.py

@@ -0,0 +1,26 @@
+import gzip
+import pytest
+from nodetool.workflows.processing_context import ProcessingContext
+from nodetool.nodes.nodetool.compression import GzipCompress, GzipDecompress
+
+
+@pytest.fixture
+def context():
+    return ProcessingContext(user_id="test", auth_token="test")
+
+
+@pytest.mark.asyncio
+async def test_gzip_compress(context: ProcessingContext):
+    data = b"hello world"
+    node = GzipCompress(data=data)
+    result = await node.process(context)
+    assert gzip.decompress(result) == data
+
+
+@pytest.mark.asyncio
+async def test_gzip_decompress(context: ProcessingContext):
+    data = b"compress me"
+    compressed = gzip.compress(data)
+    node = GzipDecompress(data=compressed)
+    result = await node.process(context)
+    assert result == data