Add generic pipe() function to datar for applying custom functions in piping workflows (#212)

Co-authored-by: pwwang <[email protected]>
Co-authored-by: copilot-swe-agent[bot] <[email protected]>

Author: Copilot

datar/misc.py

@@ -1,3 +1,48 @@
+from typing import Any as _Any, Callable as _Callable
+
+from pipda import register_verb as _register_verb
+
 from .core.load_plugins import plugin as _plugin
 
 locals().update(_plugin.hooks.misc_api())
+
+
+@_register_verb(object)
+def pipe(data: _Any, func: _Callable, *args, **kwargs) -> _Any:
+    """Apply a function to the data
+
+    This function is similar to pandas.DataFrame.pipe() and allows you to
+    apply custom functions in a piping workflow. Works with any data type.
+
+    Args:
+        data: The data object (can be any type)
+        func: Function to apply to the data. ``args`` and ``kwargs`` are
+            passed into ``func``.
+        *args: Positional arguments passed into ``func``
+        **kwargs: Keyword arguments passed into ``func``
+
+    Returns:
+        The return value of ``func``
+
+    Examples:
+        >>> import datar.all as dr
+        >>> # Works with lists
+        >>> [1, 2, 3] >> dr.pipe(lambda x: [i * 2 for i in x])
+        [2, 4, 6]
+
+        >>> # Works with dicts
+        >>> data = {'a': 1, 'b': 2}
+        >>> data >> dr.pipe(lambda x: {k: v * 2 for k, v in x.items()})
+        {'a': 2, 'b': 4}
+
+        >>> # With additional arguments
+        >>> def add_value(data, value):
+        ...     return [x + value for x in data]
+        >>> [1, 2, 3] >> dr.pipe(add_value, 10)
+        [11, 12, 13]
+
+        >>> # Chain multiple operations
+        >>> [1, 2, 3] >> dr.pipe(lambda x: [i * 2 for i in x]) >> dr.pipe(sum)
+        12
+    """
+    return func(data, *args, **kwargs)

docs/reference-maps/other.md

@@ -26,6 +26,7 @@
 |---|---|---:|
 |[**`get()`**][2]|Extract values from data frames|[:material-notebook:][1]|
 |[**`flatten()`**][2]|Flatten values of data frames|[:material-notebook:][1]|
+|[**`pipe()`**][9]|Apply a function to data in a piping workflow|[:material-notebook:][1]|
 
 ### Functions
 
@@ -44,3 +45,4 @@
 [6]: ../../api/datar.apis.datar/#datar.apis.datar.pd_str
 [7]: ../../api/datar.apis.datar/#datar.apis.datar.pd_cat
 [8]: ../../api/datar.apis.datar/#datar.apis.datar.pd_dt
+[9]: ../../api/datar.misc/#datar.misc.pipe

tests/test_pipe.py

@@ -0,0 +1,108 @@
+import pytest
+from datar.all import pipe
+
+
+def test_pipe_with_list():
+    """Test pipe with a list"""
+    data = [1, 2, 3, 4, 5]
+    result = data >> pipe(lambda x: [i * 2 for i in x])
+    expected = [2, 4, 6, 8, 10]
+    assert result == expected
+
+
+def test_pipe_with_dict():
+    """Test pipe with a dictionary"""
+    data = {'a': 1, 'b': 2, 'c': 3}
+    result = data >> pipe(lambda x: {k: v * 2 for k, v in x.items()})
+    expected = {'a': 2, 'b': 4, 'c': 6}
+    assert result == expected
+
+
+def test_pipe_with_args():
+    """Test pipe with additional positional arguments"""
+    data = [1, 2, 3]
+    
+    def add_value(data, value):
+        return [x + value for x in data]
+    
+    result = data >> pipe(add_value, 10)
+    expected = [11, 12, 13]
+    assert result == expected
+
+
+def test_pipe_with_kwargs():
+    """Test pipe with keyword arguments"""
+    data = [1, 2, 3]
+    
+    def multiply_data(data, factor=1):
+        return [x * factor for x in data]
+    
+    result = data >> pipe(multiply_data, factor=10)
+    expected = [10, 20, 30]
+    assert result == expected
+
+
+def test_pipe_with_string():
+    """Test pipe with string operations"""
+    data = "hello"
+    result = data >> pipe(lambda x: x.upper())
+    assert result == "HELLO"
+
+
+def test_pipe_with_tuple():
+    """Test pipe with tuple"""
+    data = (1, 2, 3)
+    result = data >> pipe(lambda x: tuple(i * 2 for i in x))
+    expected = (2, 4, 6)
+    assert result == expected
+
+
+def test_pipe_returns_different_type():
+    """Test that pipe can return different types"""
+    data = [1, 2, 3, 4, 5]
+    result = data >> pipe(sum)
+    assert result == 15
+
+
+def test_pipe_chain_multiple():
+    """Test chaining multiple pipe operations"""
+    data = [1, 2, 3]
+    
+    result = (
+        data 
+        >> pipe(lambda x: [i * 2 for i in x])
+        >> pipe(lambda x: [i + 1 for i in x])
+    )
+    expected = [3, 5, 7]
+    assert result == expected
+
+
+def test_pipe_with_custom_class():
+    """Test pipe with a custom class"""
+    class Counter:
+        def __init__(self, value):
+            self.value = value
+        
+        def increment(self, amount):
+            return Counter(self.value + amount)
+        
+        def __eq__(self, other):
+            return self.value == other.value
+    
+    counter = Counter(5)
+    result = counter >> pipe(lambda x: x.increment(10))
+    expected = Counter(15)
+    assert result == expected
+
+
+def test_pipe_with_multiple_args_and_kwargs():
+    """Test pipe with both args and kwargs"""
+    data = [1, 2, 3]
+    
+    def transform(data, multiplier, offset=0):
+        return [x * multiplier + offset for x in data]
+    
+    result = data >> pipe(transform, 2, offset=5)
+    expected = [7, 9, 11]
+    assert result == expected
+