startup: add classes Pipe & PipeMap

profile_default/startup/init.py

@@ -1,3 +1,4 @@
+import functools
 import itertools
 import os
 import pathlib
@@ -58,7 +59,7 @@ def split_pgp_file(
             prefix, remaining_bytes = remaining_bytes.split(packet.data, maxsplit=1)
         except ValueError:
             assert len(packet.data) > 596  # actual threshold might be higher
-            split_index = 2 ** 9
+            split_index = 2**9
             prefix, remaining_bytes = remaining_bytes.split(
                 packet.data[:split_index], maxsplit=1
             )
@@ -135,3 +136,16 @@ def yaml_dump(path: typing.Union[pathlib.Path, str], data: typing.Any) -> None:
 def yaml_load(path: typing.Union[pathlib.Path, str]) -> typing.Any:
     with pathlib.Path(path).open("r") as stream:
         return yaml.safe_load(stream)
+
+
+class Pipe:
+    def __init__(self, function: typing.Callable[[typing.Any], typing.Any]) -> None:
+        self._function = function
+
+    def __ror__(self, other: typing.Iterable) -> typing.Any:
+        return self._function(other)
+
+
+class PipeMap(Pipe):
+    def __init__(self, function: typing.Callable[[typing.Any], typing.Any]) -> None:
+        self._function = functools.partial(map, function)