Expr

Bases: Pipeable

Represents an expression in the pipeline.

An Expr encapsulates a sequence of operations to be applied to keys on a python dict.

Each Expr instance maintains: - A list of tokens representing the keys to access in the dict (the first being the input given to the key function), - A tuple of operations to apply to the accessed data - An alias for the expression (default to the last token).

Source code in src/pyochain/_dict/_exprs.py

@dataclass(slots=True)
class Expr(Pipeable):
    """
    Represents an expression in the pipeline.

    An Expr encapsulates a sequence of operations to be applied to keys on a python dict.

    Each Expr instance maintains:
    - A list of tokens representing the keys to access in the dict (the first being the input given to the `key` function),
    - A tuple of operations to apply to the accessed data
    - An alias for the expression (default to the last token).
    """

    __tokens__: list[str]
    __ops__: tuple[Callable[[object], object], ...]
    _alias: str

    def __repr__(self) -> str:
        parts: list[str] = []
        s_parts: list[str] = []
        for t in self.__tokens__:
            parts.append(f"field({t!r})")
            if s_parts:
                s_parts.append(".")
            s_parts.append(str(t))
        symbolic = ".".join(parts) if parts else "<root>"
        lowered = "".join(s_parts) or "<root>"
        base = f"Expr({symbolic} -> {lowered})"
        return f"{base}.alias({self._alias!r})"

    def _to_expr(self, op: Callable[[Any], Any]) -> Self:
        return self.__class__(
            self.__tokens__,
            self.__ops__ + (op,),
            self._alias,
        )

    def key(self, name: str) -> Self:
        """
        Add a key to the expression.

        Allow to access nested keys in a dict.

        Args:
            name: The key to access.
        Example:
        ```python
        >>> import pyochain as pc
        >>> expr = pc.key("a").key("b").key("c")
        >>> expr.__tokens__
        ['a', 'b', 'c']
        >>> data = {"a": {"b": {"c": 42}}}
        >>> pc.Dict(data).select(expr).unwrap()
        {'c': 42}

        ```
        """
        return self.__class__(
            self.__tokens__ + [name],
            self.__ops__,
            name,
        )

    def alias(self, name: str) -> Self:
        return self.__class__(self.__tokens__, self.__ops__, name)

    @property
    def name(self) -> str:
        return self._alias

    def apply(self, fn: Callable[[Any], Any]) -> Self:
        """
        Applies the given function fn to the data within the current Expr instance
        """

        def _apply(data: Any) -> Any:
            return fn(data)

        return self._to_expr(_apply)

__ops__ instance-attribute

__ops__: tuple[Callable[[object], object], ...]

__tokens__ instance-attribute

__tokens__: list[str]

_alias instance-attribute

_alias: str

name property

name: str

__init__

__init__(
    __tokens__: list[str], __ops__: tuple[Callable[[object], object], ...], _alias: str
) -> None

__repr__

__repr__() -> str

Source code in src/pyochain/_dict/_exprs.py

def __repr__(self) -> str:
    parts: list[str] = []
    s_parts: list[str] = []
    for t in self.__tokens__:
        parts.append(f"field({t!r})")
        if s_parts:
            s_parts.append(".")
        s_parts.append(str(t))
    symbolic = ".".join(parts) if parts else "<root>"
    lowered = "".join(s_parts) or "<root>"
    base = f"Expr({symbolic} -> {lowered})"
    return f"{base}.alias({self._alias!r})"

_to_expr

_to_expr(op: Callable[[Any], Any]) -> Self

Source code in src/pyochain/_dict/_exprs.py

def _to_expr(self, op: Callable[[Any], Any]) -> Self:
    return self.__class__(
        self.__tokens__,
        self.__ops__ + (op,),
        self._alias,
    )

alias

alias(name: str) -> Self

Source code in src/pyochain/_dict/_exprs.py

def alias(self, name: str) -> Self:
    return self.__class__(self.__tokens__, self.__ops__, name)

apply

apply(fn: Callable[[Any], Any]) -> Self

Applies the given function fn to the data within the current Expr instance

Source code in src/pyochain/_dict/_exprs.py

def apply(self, fn: Callable[[Any], Any]) -> Self:
    """
    Applies the given function fn to the data within the current Expr instance
    """

    def _apply(data: Any) -> Any:
        return fn(data)

    return self._to_expr(_apply)

key

key(name: str) -> Self

Add a key to the expression.

Allow to access nested keys in a dict.

Parameters:

Name	Type	Description	Default
name	str	The key to access.	required

Example:

>>> import pyochain as pc
>>> expr = pc.key("a").key("b").key("c")
>>> expr.__tokens__
['a', 'b', 'c']
>>> data = {"a": {"b": {"c": 42}}}
>>> pc.Dict(data).select(expr).unwrap()
{'c': 42}

Source code in src/pyochain/_dict/_exprs.py

def key(self, name: str) -> Self:
    """
    Add a key to the expression.

    Allow to access nested keys in a dict.

    Args:
        name: The key to access.
    Example:
    ```python
    >>> import pyochain as pc
    >>> expr = pc.key("a").key("b").key("c")
    >>> expr.__tokens__
    ['a', 'b', 'c']
    >>> data = {"a": {"b": {"c": 42}}}
    >>> pc.Dict(data).select(expr).unwrap()
    {'c': 42}

    ```
    """
    return self.__class__(
        self.__tokens__ + [name],
        self.__ops__,
        name,
    )

pipe

pipe(func: Callable[Concatenate[Self, P], R], *args: P.args, **kwargs: P.kwargs) -> R

Pipe the instance in the function and return the result.

Source code in src/pyochain/_core/_main.py

def pipe[**P, R](
    self,
    func: Callable[Concatenate[Self, P], R],
    *args: P.args,
    **kwargs: P.kwargs,
) -> R:
    """Pipe the instance in the function and return the result."""
    return func(self, *args, **kwargs)

Create an Expr that accesses the given key.

Source code in src/pyochain/_dict/_exprs.py

def key(name: str) -> Expr:
    """Create an Expr that accesses the given key."""
    return Expr([name], (), name)