v0.3.0 (#8)

* fix: update to use param spec

* fix: update typehint for Data

* fix: update typehint for register fns

* fix: update typehint for domain register fns

* feat: file ops for adapter

* bump: => 0.3.0

Author: second-ed

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "io-adapters"
-version = "0.2.2"
+version = "0.3.0"
 description = "Dependency Injection Adapters"
 readme = "README.md"
 authors = [

src/io_adapters/_adapters.py

@@ -2,7 +2,11 @@
 
 import datetime
 import logging
+import shutil
+from abc import ABC, abstractmethod
 from collections.abc import Callable, Hashable
+from copy import deepcopy
+from fnmatch import fnmatch
 from pathlib import Path
 from types import MappingProxyType
 
@@ -16,7 +20,7 @@
 
 
 @attrs.define
-class IoAdapter:
+class IoAdapter(ABC):
     read_fns: MappingProxyType[Hashable, ReadFn] = attrs.field(
         default=READ_FNS,
         validator=[
@@ -147,13 +151,44 @@ def get_guid(self) -> str:
     def get_datetime(self) -> datetime.datetime:
         return self.datetime_fn()
 
+    @abstractmethod
+    def list_files(self, path: str | Path, glob_pattern: str = "*") -> list[str]: ...
+
+    @abstractmethod
+    def copy_file(self, old: str | Path, new: str | Path) -> None: ...
+
+    @abstractmethod
+    def delete_file(self, path: str | Path, *, missing_ok: bool = True) -> None: ...
+
+    def move_file(self, old: str | Path, new: str | Path) -> None:
+        self.copy_file(old, new)
+        self.delete_file(old)
+
+    @abstractmethod
+    def exists(self, path: str | Path) -> bool: ...
+
 
 @attrs.define
 class RealAdapter(IoAdapter):
     def __attrs_post_init__(self) -> None:
         self.guid_fn = self.guid_fn or default_guid
         self.datetime_fn = self.datetime_fn or default_datetime
 
+    def list_files(self, path: str | Path, glob_pattern: str = "*") -> list[str]:
+        return sorted(map(str, Path(path).rglob(glob_pattern)))
+
+    def copy_file(self, old: str | Path, new: str | Path) -> None:
+        src = Path(old)
+        dst = Path(new)
+        dst.parent.mkdir(parents=True, exist_ok=True)
+        shutil.copy2(src, dst)
+
+    def delete_file(self, path: str | Path, *, missing_ok: bool = True) -> None:
+        Path(path).unlink(missing_ok=missing_ok)
+
+    def exists(self, path: str | Path) -> bool:
+        return Path(path).exists()
+
 
 @attrs.define
 class FakeAdapter(IoAdapter):
@@ -174,8 +209,26 @@ def _read_fn(self, path: str | Path) -> Data:
     def _write_fn(self, data: Data, path: str | Path) -> None:
         self.files[str(path)] = data
 
-    def get_guid(self) -> str:
-        return self.guid_fn()
+    def list_files(self, path: str | Path, glob_pattern: str = "*") -> list[str]:
+        return sorted(
+            [
+                str(p)
+                for p in self.files
+                if Path(p).is_relative_to(Path(path)) and fnmatch(Path(p).name, glob_pattern)
+            ]
+        )
 
-    def get_datetime(self) -> datetime.datetime:
-        return self.datetime_fn()
+    def copy_file(self, old: str | Path, new: str | Path) -> None:
+        self.files[str(new)] = deepcopy(self.files[str(old)])
+
+    def delete_file(self, path: str | Path, *, missing_ok: bool = True) -> None:
+        path = str(Path(path))
+
+        try:
+            self.files.pop(path)
+        except KeyError as e:
+            if not missing_ok:
+                raise FileNotFoundError(path) from e
+
+    def exists(self, path: str | Path) -> bool:
+        return str(path) in map(str, self.files)

src/io_adapters/_container.py

@@ -9,7 +9,7 @@
 from attrs.validators import deep_iterable, instance_of
 
 from io_adapters import FakeAdapter, RealAdapter
-from io_adapters._registries import standardise_key
+from io_adapters._registries import ReadFn, WriteFn, standardise_key
 
 logger = logging.getLogger(__name__)
 
@@ -19,7 +19,7 @@ class _FnType(Enum):
     WRITE = auto()
 
 
-DomainFns: TypeAlias = dict[Hashable, dict[Hashable, Callable]]
+DomainFns: TypeAlias = dict[Hashable, dict[Hashable, ReadFn | WriteFn]]
 
 
 @attrs.define
@@ -150,7 +150,7 @@ def read_json(path: str | Path, **kwargs: dict) -> dict:
         domain = standardise_key(domain)
         key = standardise_key(key)
 
-        def wrapper(func: Callable) -> Callable:
+        def wrapper(func: ReadFn) -> ReadFn:
             logger.info(f"registering read fn {key = } {func = }")
             self.domain_fns[domain][_FnType.READ][key] = func
             return func
@@ -182,7 +182,7 @@ def write_json(data: dict, path: str | Path, **kwargs: dict) -> None:
         domain = standardise_key(domain)
         key = standardise_key(key)
 
-        def wrapper(func: Callable) -> Callable:
+        def wrapper(func: WriteFn) -> WriteFn:
             logger.info(f"registering read fn {key = } {func = }")
             self.domain_fns[domain][_FnType.WRITE][key] = func
             return func
@@ -265,7 +265,7 @@ def add_domain(domain: Hashable) -> None:
     return DEFAULT_CONTAINER.add_domain(domain)
 
 
-def register_domain_read_fn(domain: Hashable, key: Hashable) -> Callable:
+def register_domain_read_fn(domain: Hashable, key: Hashable) -> ReadFn:
     """Register a read function to a domain in the default ``Container``.
 
     Decorators can be stacked to register the same function to multiple domains.
@@ -290,7 +290,7 @@ def read_json(path: str | Path, **kwargs: dict) -> dict:
     return DEFAULT_CONTAINER.register_domain_read_fn(domain, key)
 
 
-def register_domain_write_fn(domain: Hashable, key: Hashable) -> Callable:
+def register_domain_write_fn(domain: Hashable, key: Hashable) -> WriteFn:
     """Register a write function to a domain in the default ``Container``.
 
     Decorators can be stacked to register the same function to multiple domains.

src/io_adapters/_registries.py

@@ -3,22 +3,22 @@
 import logging
 from collections.abc import Callable, Hashable
 from pathlib import Path
-from typing import Concatenate, ParamSpec, TypeAlias
+from typing import Concatenate, ParamSpec, TypeAlias, TypeVar
 
 logger = logging.getLogger(__name__)
 
-Data: TypeAlias = "Data"
+Data = TypeVar("Data")
 P = ParamSpec("P")
 
-ReadFn = Callable[Concatenate[str | Path, P], Data]
-WriteFn = Callable[Concatenate[Data, str | Path, P], None]
+ReadFn: TypeAlias = Callable[Concatenate[str | Path, P], Data]
+WriteFn: TypeAlias = Callable[Concatenate[Data, str | Path, P], None]
 
 
 READ_FNS: dict[Hashable, ReadFn] = {}
 WRITE_FNS: dict[Hashable, WriteFn] = {}
 
 
-def register_read_fn(key: Hashable) -> Callable:
+def register_read_fn(key: Hashable) -> Callable[[ReadFn], ReadFn]:
     """Register a read function to the read functions constant.
 
     This is useful for smaller projects where domain isolation isn't required.
@@ -36,15 +36,15 @@ def read_json(path: str | Path, **kwargs: dict) -> dict:
     """
     key = standardise_key(key)
 
-    def wrapper(func: Callable) -> Callable:
+    def wrapper(func: ReadFn) -> ReadFn:
         logger.info(f"registering read fn {key = } {func = }")
         READ_FNS[key] = func
         return func
 
     return wrapper
 
 
-def register_write_fn(key: Hashable) -> Callable:
+def register_write_fn(key: Hashable) -> Callable[[WriteFn], WriteFn]:
     """Register a write function to the write functions constant.
 
     This is useful for smaller projects where domain isolation isn't required.
@@ -66,7 +66,7 @@ def write_json(data: dict, path: str | Path, **kwargs: dict) -> None:
     """
     key = standardise_key(key)
 
-    def wrapper(func: Callable) -> Callable:
+    def wrapper(func: WriteFn) -> WriteFn:
         logger.info(f"registering write fn {key = } {func = }")
         WRITE_FNS[key] = func
         return func

tests/test_adapters.py

@@ -1,4 +1,5 @@
 import operator
+import shutil
 from contextlib import nullcontext
 from pathlib import Path
 
@@ -8,6 +9,28 @@
 
 REPO_ROOT = Path(__file__).parents[1]
 MOCK_DATA_PATH = f"{REPO_ROOT}/tests/mock_data/mock.json"
+FILES = ("__init__.py", "mock.csv", "reqs.txt", "main.py")
+TMP_ROOT = REPO_ROOT.joinpath("tests", "tmp_mock_data")
+INITIAL_FILES = [TMP_ROOT.joinpath(x) for x in FILES]
+
+
+@pytest.fixture(autouse=True)
+def setup():
+    for f in INITIAL_FILES:
+        f.parent.mkdir(parents=True, exist_ok=True)
+        f.joinpath(f).touch()
+
+    yield
+
+    shutil.rmtree(TMP_ROOT)
+
+
+ADAPTERS = [
+    pytest.param(RealAdapter(), id="using a RealAdapter"),
+    pytest.param(
+        FakeAdapter(files=dict.fromkeys(map(str, INITIAL_FILES), "")), id="using a FakeAdapter"
+    ),
+]
 
 
 @pytest.mark.parametrize(
@@ -73,3 +96,88 @@ def test_guid(adapter, op):
 )
 def test_datetime(adapter, op):
     assert op(adapter().get_datetime(), adapter().get_datetime())
+
+
+@pytest.mark.parametrize("adapter", ADAPTERS)
+@pytest.mark.parametrize(
+    ("path", "glob_pattern", "expected_result"),
+    [
+        pytest.param(
+            TMP_ROOT,
+            "*",
+            [
+                f"{TMP_ROOT}/__init__.py",
+                f"{TMP_ROOT}/main.py",
+                f"{TMP_ROOT}/mock.csv",
+                f"{TMP_ROOT}/reqs.txt",
+            ],
+            id="get all files",
+        ),
+        pytest.param(
+            TMP_ROOT,
+            "*.py",
+            [f"{TMP_ROOT}/__init__.py", f"{TMP_ROOT}/main.py"],
+            id="glob .py files",
+        ),
+    ],
+)
+def test_list_files(adapter, path, glob_pattern, expected_result):
+    assert adapter.list_files(path, glob_pattern) == expected_result
+
+
+@pytest.mark.parametrize("adapter", ADAPTERS)
+@pytest.mark.parametrize(
+    ("old", "new"),
+    [
+        pytest.param(
+            f"{TMP_ROOT}/__init__.py",
+            f"{TMP_ROOT}/new_init.py",
+            id="both old and new exist after copy",
+        ),
+    ],
+)
+def test_copy_file(adapter, old, new):
+    adapter.copy_file(old, new)
+    assert adapter.exists(old)
+    assert adapter.exists(new)
+
+
+@pytest.mark.parametrize("adapter", ADAPTERS)
+@pytest.mark.parametrize(
+    ("old", "new"),
+    [
+        pytest.param(
+            f"{TMP_ROOT}/__init__.py",
+            f"{TMP_ROOT}/new_init.py",
+            id="old not exists and new exists after move",
+        ),
+    ],
+)
+def test_move_file(adapter, old, new):
+    adapter.move_file(old, new)
+    assert not adapter.exists(old)
+    assert adapter.exists(new)
+
+
+@pytest.mark.parametrize("adapter", ADAPTERS)
+@pytest.mark.parametrize(
+    ("path", "missing_ok", "expected_context"),
+    [
+        pytest.param(
+            f"{TMP_ROOT}/__init__.py",
+            True,
+            nullcontext(),
+            id="file is deleted after delete",
+        ),
+        pytest.param(
+            f"{TMP_ROOT}/INVALID_FILE.py",
+            False,
+            pytest.raises(FileNotFoundError),
+            id="file is deleted after delete",
+        ),
+    ],
+)
+def test_delete_file(adapter, path, missing_ok, expected_context):
+    with expected_context:
+        adapter.delete_file(path, missing_ok=missing_ok)
+        assert not adapter.exists(path)