Use dict[str, Any] for JSON repository document types.
PR Title Check / check-title (pull_request) Successful in 6s
Code Quality Pipeline / code-quality (pull_request) Successful in 1m28s
Test Python Package / unit-tests (pull_request) Successful in 1m29s
Test Python Package / coverage-report (pull_request) Successful in 16s
Test Python Package / integration-tests (pull_request) Successful in 48s
PR Title Check / check-title (pull_request) Successful in 6s
Code Quality Pipeline / code-quality (pull_request) Successful in 1m28s
Test Python Package / unit-tests (pull_request) Successful in 1m29s
Test Python Package / coverage-report (pull_request) Successful in 16s
Test Python Package / integration-tests (pull_request) Successful in 48s
Clarifies that JSON CRUD operates on string-keyed objects without changing runtime behavior. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
co-authored by
Cursor
parent
bd4794d1c0
commit
42e885edd5
@@ -123,16 +123,18 @@ with ArtifactObjectRepository() as repo:
|
|||||||
### Subclassing in your own project
|
### Subclassing in your own project
|
||||||
|
|
||||||
```python
|
```python
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
from python_repositories import RedisAdapter
|
from python_repositories import RedisAdapter
|
||||||
|
|
||||||
class UserRepository(RedisAdapter):
|
class UserRepository(RedisAdapter):
|
||||||
def _key(self, user_id: str) -> str:
|
def _key(self, user_id: str) -> str:
|
||||||
return f"user:{user_id}"
|
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))
|
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)
|
self.set(self._key(user_id), user)
|
||||||
```
|
```
|
||||||
|
|
||||||
|
|||||||
@@ -2,7 +2,7 @@
|
|||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
from collections.abc import Iterator
|
from collections.abc import Iterator
|
||||||
from typing import cast
|
from typing import Any, cast
|
||||||
|
|
||||||
from python_repositories.adapters.connection_aware_adapter import (
|
from python_repositories.adapters.connection_aware_adapter import (
|
||||||
ConnectionAwareAdapter,
|
ConnectionAwareAdapter,
|
||||||
@@ -89,7 +89,7 @@ class RedisAdapter(JsonRepositoryInterface, ConnectionAwareAdapter):
|
|||||||
except (redis.ConnectionError, redis.TimeoutError):
|
except (redis.ConnectionError, redis.TimeoutError):
|
||||||
return False
|
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."""
|
"""Set a JSON object in Redis."""
|
||||||
# Check input
|
# Check input
|
||||||
if not isinstance(key, str) or len(key) == 0:
|
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._client.json().set(key, self.path, data)
|
||||||
self.logger.debug(f"Set {key} to {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."""
|
"""Get a JSON object from Redis."""
|
||||||
# Check input
|
# Check input
|
||||||
if not isinstance(key, str) or len(key) == 0:
|
if not isinstance(key, str) or len(key) == 0:
|
||||||
@@ -113,7 +113,7 @@ class RedisAdapter(JsonRepositoryInterface, ConnectionAwareAdapter):
|
|||||||
assert self._client is not None
|
assert self._client is not None
|
||||||
# Get data
|
# Get data
|
||||||
data = cast(
|
data = cast(
|
||||||
dict | None,
|
dict[str, Any] | None,
|
||||||
self._client.json().get(key),
|
self._client.json().get(key),
|
||||||
)
|
)
|
||||||
self.logger.debug(f"Got {data} from {key}")
|
self.logger.debug(f"Got {data} from {key}")
|
||||||
|
|||||||
@@ -1,5 +1,7 @@
|
|||||||
"""Example domain repository backed by Redis JSON."""
|
"""Example domain repository backed by Redis JSON."""
|
||||||
|
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
from python_repositories.adapters.redis_adapter import RedisAdapter
|
from python_repositories.adapters.redis_adapter import RedisAdapter
|
||||||
|
|
||||||
|
|
||||||
@@ -9,10 +11,10 @@ class UserJsonRepository(RedisAdapter):
|
|||||||
def _key(self, user_id: str) -> str:
|
def _key(self, user_id: str) -> str:
|
||||||
return f"user:{user_id}"
|
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))
|
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)
|
self.set(self._key(user_id), user)
|
||||||
|
|
||||||
def delete_user(self, user_id: str) -> None:
|
def delete_user(self, user_id: str) -> None:
|
||||||
|
|||||||
@@ -2,18 +2,19 @@
|
|||||||
|
|
||||||
from collections.abc import Iterator
|
from collections.abc import Iterator
|
||||||
from abc import ABC, abstractmethod
|
from abc import ABC, abstractmethod
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
|
||||||
class JsonRepositoryInterface(ABC):
|
class JsonRepositoryInterface(ABC):
|
||||||
"""Interface that defines JSON document CRUD methods."""
|
"""Interface that defines JSON document CRUD methods."""
|
||||||
|
|
||||||
@abstractmethod
|
@abstractmethod
|
||||||
def get(self, key: str) -> dict | None:
|
def get(self, key: str) -> dict[str, Any] | None:
|
||||||
"""Get a JSON object by key."""
|
"""Get a JSON object by key."""
|
||||||
...
|
...
|
||||||
|
|
||||||
@abstractmethod
|
@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."""
|
"""Set a JSON object by key."""
|
||||||
...
|
...
|
||||||
|
|
||||||
|
|||||||
@@ -1,5 +1,7 @@
|
|||||||
"""Unit tests for JsonRepositoryInterface."""
|
"""Unit tests for JsonRepositoryInterface."""
|
||||||
|
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
from python_repositories.interfaces.json_repository_interface import (
|
from python_repositories.interfaces.json_repository_interface import (
|
||||||
JsonRepositoryInterface,
|
JsonRepositoryInterface,
|
||||||
@@ -12,7 +14,7 @@ def test_instantiation_fails_when_get_not_implemented() -> None:
|
|||||||
class Incomplete(JsonRepositoryInterface):
|
class Incomplete(JsonRepositoryInterface):
|
||||||
"""A class that does not implement get."""
|
"""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
|
pass
|
||||||
|
|
||||||
def delete(self, key: str) -> None:
|
def delete(self, key: str) -> None:
|
||||||
@@ -31,7 +33,7 @@ def test_instantiation_fails_when_set_not_implemented() -> None:
|
|||||||
class Incomplete(JsonRepositoryInterface):
|
class Incomplete(JsonRepositoryInterface):
|
||||||
"""A class that does not implement set."""
|
"""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
|
return None
|
||||||
|
|
||||||
def delete(self, key: str) -> None:
|
def delete(self, key: str) -> None:
|
||||||
@@ -50,10 +52,10 @@ def test_instantiation_fails_when_delete_not_implemented() -> None:
|
|||||||
class Incomplete(JsonRepositoryInterface):
|
class Incomplete(JsonRepositoryInterface):
|
||||||
"""A class that does not implement delete."""
|
"""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
|
return None
|
||||||
|
|
||||||
def set(self, key: str, data: dict) -> None:
|
def set(self, key: str, data: dict[str, Any]) -> None:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
def list_keys(self, pattern: str) -> list[str]:
|
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):
|
class Incomplete(JsonRepositoryInterface):
|
||||||
"""A class that does not implement list_keys."""
|
"""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
|
return None
|
||||||
|
|
||||||
def set(self, key: str, data: dict) -> None:
|
def set(self, key: str, data: dict[str, Any]) -> None:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
def delete(self, key: str) -> None:
|
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."""
|
"""Test that the default scan_keys implementation delegates to list_keys."""
|
||||||
|
|
||||||
class Complete(JsonRepositoryInterface):
|
class Complete(JsonRepositoryInterface):
|
||||||
def get(self, key: str) -> dict | None:
|
def get(self, key: str) -> dict[str, Any] | None:
|
||||||
return None
|
return None
|
||||||
|
|
||||||
def set(self, key: str, data: dict) -> None:
|
def set(self, key: str, data: dict[str, Any]) -> None:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
def delete(self, key: str) -> None:
|
def delete(self, key: str) -> None:
|
||||||
|
|||||||
Reference in New Issue
Block a user