497533f0e33be3be8e62ba16b4c69614eb2b8229
Return Self from __enter__ so with-blocks preserve subclass types, and align mypy python_version with the 3.12 runtime target. Co-authored-by: Cursor <cursoragent@cursor.com>
python-repositories
Unified repository interfaces and technology-specific adapters for Python projects.
Subclass an adapter in your own repository to add domain-specific methods while reusing connection management and CRUD operations.
Architecture
| Layer | Responsibility |
|---|---|
| Interfaces | Abstract contracts for connection, context, and CRUD |
| Adapters | Technology-specific base classes (RedisAdapter, MinioAdapter) |
| Your project | Subclass an adapter and add domain methods |
Optional dependencies
Install with the extras you need:
uv add python-repositories[redis]
uv add python-repositories[minio]
uv add python-repositories[redis,minio]
Redis (JsonRepositoryInterface)
Requires Redis with the RedisJSON module (e.g. redis-stack).
| Environment variable | Description |
|---|---|
REDIS_URI |
Redis connection URL (e.g. redis://localhost:6379) |
MinIO (ObjectRepositoryInterface)
| Environment variable | Description |
|---|---|
MINIO_ENDPOINT |
MinIO server endpoint |
MINIO_ACCESS_KEY |
Access key |
MINIO_SECRET_KEY |
Secret key |
MINIO_BUCKET |
Bucket name (created on connect if missing) |
Quick start
JSON documents with Redis
from python_repositories.examples.user_json_repository import UserJsonRepository
with UserJsonRepository() as repo:
repo.save_user("alice", {"name": "Alice", "email": "alice@example.com"})
user = repo.get_user("alice")
repo.delete_user("alice")
Binary objects with MinIO
from io import BytesIO
from python_repositories.examples.artifact_object_repository import (
ArtifactObjectRepository,
)
with ArtifactObjectRepository() as repo:
repo.store_artifact("report-1", BytesIO(b"pdf bytes here"))
data = repo.get_artifact("report-1")
Subclassing in your own project
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:
return self.get(self._key(user_id))
def save_user(self, user_id: str, user: dict) -> None:
self.set(self._key(user_id), user)
Public API
from python_repositories import (
ConnectionAwareInterface,
ContextAwareInterface,
JsonRepositoryInterface,
ObjectRepositoryInterface,
RedisAdapter,
MinioAdapter,
)
Development
uv sync --all-extras
uv run pytest tests/integration/ -v
Integration tests require Docker (testcontainers).