✨ (store) add stream method for file-like store

anystore/cli.py

@@ -82,9 +82,12 @@ def cli_keys(
     prefix: Annotated[Optional[str], typer.Argument(..., help="Key prefix")] = None,
     o: Annotated[str, typer.Option("-o", help="Output uri")] = "-",
 ):
+    """
+    Iterate keys in given store
+    """
     with ErrorHandler():
         S = get_store(uri=state["uri"], use_pickle=state["pickle"])
-        keys = "\n".join(S.iterate_keys(prefix))
+        keys = "\n".join(S.iterate_keys(prefix)) + "\n"
         smart_write(o, keys.encode())
 
 

anystore/store/base.py

@@ -31,6 +31,12 @@ def _read(self, key: Uri, raise_on_nonexist: bool | None = True, **kwargs) -> An
         """
         raise NotImplementedError
 
+    def _stream(self, key: Uri, raise_on_nonexist: bool | None = True, **kwargs) -> Any:
+        """
+        Stream key line by line from actual backend (for file-like powered backend)
+        """
+        raise NotImplementedError
+
     def _get_key_prefix(self) -> str:
         """
         Get backend specific key prefix
@@ -64,6 +70,22 @@ def get(
                 raise DoesNotExist(f"Key does not exist: `{key}`")
             return None
 
+    def stream(
+        self,
+        key: Uri,
+        raise_on_nonexist: bool | None = None,
+        serialization_mode: Mode | None = None,
+        **kwargs,
+    ) -> Generator[Any, None, None]:
+        key = self.get_key(key)
+        try:
+            for line in self._stream(key, raise_on_nonexist, **kwargs):
+                yield from_store(line, serialization_mode)
+        except FileNotFoundError:  # fsspec
+            if raise_on_nonexist:
+                raise DoesNotExist(f"Key does not exist: `{key}`")
+            return None
+
     def put(
         self, key: Uri, value: Any, serialization_mode: Mode | None = None, **kwargs
     ):

anystore/store/fs.py

@@ -7,7 +7,7 @@
 import fsspec
 from banal import ensure_dict
 
-from anystore.io import smart_read, smart_write
+from anystore.io import smart_read, smart_stream, smart_write
 from anystore.exceptions import DoesNotExist
 from anystore.store.base import BaseStore
 from anystore.types import Uri, Value
@@ -27,7 +27,15 @@ def _read(
                 raise DoesNotExist(f"Key does not exist: `{key}`")
             return None
 
-    # def _exists(self, key: Uri) -> bool:
+    def _stream(
+        self, key: Uri, raise_on_nonexist: bool | None = True, **kwargs
+    ) -> Generator[Value, None, None]:
+        try:
+            yield from smart_stream(str(key), **kwargs)
+        except FileNotFoundError:
+            if raise_on_nonexist:
+                raise DoesNotExist(f"Key does not exist: `{key}`")
+
     def _get_key_prefix(self) -> str:
         return str(self.uri).rstrip("/")
 

tests/test_store.py

@@ -68,6 +68,16 @@ def test_store_fs(tmp_path, fixtures_path):
     assert (tmp_path / "foo/bar/baz").exists()
     assert store.get("/bar/baz") == 1
 
+    # stream
+    store = Store(uri=fixtures_path)
+    tested = False
+    for ix, line in enumerate(store.stream("lorem.txt", mode="r")):
+        if ix == 1:
+            assert line.startswith("tempor")
+            tested = True
+            break
+    assert tested
+
 
 def test_store_intialize(fixtures_path):
     # initialize (take env vars into account)