Merge pull request '[patch] Use dict[str, Any] for JSON repository document types' (#38) from cursor/json-dict-str-any-typing into main

Reviewed-on: https://gitea.lille-vemmelund.dk/brian/python-repositories/pulls/38

README.md

@@ -123,16 +123,18 @@ with ArtifactObjectRepository() as repo:
 ### Subclassing in your own project
 
 ```python
+from typing import Any
+
 from python_repositories import RedisAdapter
 
 class UserRepository(RedisAdapter):
     def _key(self, user_id: str) -> str:
         return f"user:{user_id}"
 
-    def get_user(self, user_id: str) -> dict | None:
+    def get_user(self, user_id: str) -> dict[str, Any] | None:
         return self.get(self._key(user_id))
 
-    def save_user(self, user_id: str, user: dict) -> None:
+    def save_user(self, user_id: str, user: dict[str, Any]) -> None:
         self.set(self._key(user_id), user)
 ```
 

python_repositories/adapters/redis_adapter.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 from collections.abc import Iterator
-from typing import cast
+from typing import Any, cast
 
 from python_repositories.adapters.connection_aware_adapter import (
     ConnectionAwareAdapter,
@@ -89,7 +89,7 @@ class RedisAdapter(JsonRepositoryInterface, ConnectionAwareAdapter):
         except (redis.ConnectionError, redis.TimeoutError):
             return False
 
-    def set(self, key: str, data: dict) -> None:
+    def set(self, key: str, data: dict[str, Any]) -> None:
         """Set a JSON object in Redis."""
         # Check input
         if not isinstance(key, str) or len(key) == 0:
@@ -103,7 +103,7 @@ class RedisAdapter(JsonRepositoryInterface, ConnectionAwareAdapter):
         self._client.json().set(key, self.path, data)
         self.logger.debug(f"Set {key} to {data}")
 
-    def get(self, key: str) -> dict | None:
+    def get(self, key: str) -> dict[str, Any] | None:
         """Get a JSON object from Redis."""
         # Check input
         if not isinstance(key, str) or len(key) == 0:
@@ -113,7 +113,7 @@ class RedisAdapter(JsonRepositoryInterface, ConnectionAwareAdapter):
         assert self._client is not None
         # Get data
         data = cast(
-            dict | None,
+            dict[str, Any] | None,
             self._client.json().get(key),
         )
         self.logger.debug(f"Got {data} from {key}")

python_repositories/examples/user_json_repository.py

@@ -1,5 +1,7 @@
 """Example domain repository backed by Redis JSON."""
 
+from typing import Any
+
 from python_repositories.adapters.redis_adapter import RedisAdapter
 
 
@@ -9,10 +11,10 @@ class UserJsonRepository(RedisAdapter):
     def _key(self, user_id: str) -> str:
         return f"user:{user_id}"
 
-    def get_user(self, user_id: str) -> dict | None:
+    def get_user(self, user_id: str) -> dict[str, Any] | None:
         return self.get(self._key(user_id))
 
-    def save_user(self, user_id: str, user: dict) -> None:
+    def save_user(self, user_id: str, user: dict[str, Any]) -> None:
         self.set(self._key(user_id), user)
 
     def delete_user(self, user_id: str) -> None:

python_repositories/interfaces/json_repository_interface.py

@@ -2,18 +2,19 @@
 
 from collections.abc import Iterator
 from abc import ABC, abstractmethod
+from typing import Any
 
 
 class JsonRepositoryInterface(ABC):
     """Interface that defines JSON document CRUD methods."""
 
     @abstractmethod
-    def get(self, key: str) -> dict | None:
+    def get(self, key: str) -> dict[str, Any] | None:
         """Get a JSON object by key."""
         ...
 
     @abstractmethod
-    def set(self, key: str, data: dict) -> None:
+    def set(self, key: str, data: dict[str, Any]) -> None:
         """Set a JSON object by key."""
         ...
 

tests/unit/json_repository_interface_test.py

@@ -1,5 +1,7 @@
 """Unit tests for JsonRepositoryInterface."""
 
+from typing import Any
+
 import pytest
 from python_repositories.interfaces.json_repository_interface import (
     JsonRepositoryInterface,
@@ -12,7 +14,7 @@ def test_instantiation_fails_when_get_not_implemented() -> None:
     class Incomplete(JsonRepositoryInterface):
         """A class that does not implement get."""
 
-        def set(self, key: str, data: dict) -> None:
+        def set(self, key: str, data: dict[str, Any]) -> None:
             pass
 
         def delete(self, key: str) -> None:
@@ -31,7 +33,7 @@ def test_instantiation_fails_when_set_not_implemented() -> None:
     class Incomplete(JsonRepositoryInterface):
         """A class that does not implement set."""
 
-        def get(self, key: str) -> dict | None:
+        def get(self, key: str) -> dict[str, Any] | None:
             return None
 
         def delete(self, key: str) -> None:
@@ -50,10 +52,10 @@ def test_instantiation_fails_when_delete_not_implemented() -> None:
     class Incomplete(JsonRepositoryInterface):
         """A class that does not implement delete."""
 
-        def get(self, key: str) -> dict | None:
+        def get(self, key: str) -> dict[str, Any] | None:
             return None
 
-        def set(self, key: str, data: dict) -> None:
+        def set(self, key: str, data: dict[str, Any]) -> None:
             pass
 
         def list_keys(self, pattern: str) -> list[str]:
@@ -69,10 +71,10 @@ def test_instantiation_fails_when_list_keys_not_implemented() -> None:
     class Incomplete(JsonRepositoryInterface):
         """A class that does not implement list_keys."""
 
-        def get(self, key: str) -> dict | None:
+        def get(self, key: str) -> dict[str, Any] | None:
             return None
 
-        def set(self, key: str, data: dict) -> None:
+        def set(self, key: str, data: dict[str, Any]) -> None:
             pass
 
         def delete(self, key: str) -> None:
@@ -86,10 +88,10 @@ def test_scan_keys_defaults_to_list_keys() -> None:
     """Test that the default scan_keys implementation delegates to list_keys."""
 
     class Complete(JsonRepositoryInterface):
-        def get(self, key: str) -> dict | None:
+        def get(self, key: str) -> dict[str, Any] | None:
             return None
 
-        def set(self, key: str, data: dict) -> None:
+        def set(self, key: str, data: dict[str, Any]) -> None:
             pass
 
         def delete(self, key: str) -> None: