Move shared connect orchestration into the base adapter so both backends short-circuit when already healthy and only reconnect after a failed probe. Co-authored-by: Cursor <cursoragent@cursor.com>
9.0 KiB
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 |
Connection adapters expose connect(), disconnect(), and is_connected(). The latter verifies backend reachability with a cached health probe (default TTL: 1 second). Subclasses may override health_check_ttl_seconds. connect() is idempotent: calling it while already connected and healthy is a no-op.
Optional dependencies
Repository interfaces import with the base package. Adapters require the matching extra; importing an adapter without its extra raises ImportError with install instructions.
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) |
For key discovery:
list_keys(pattern)is simple and returns alist[str], but it uses RedisKEYSand may block on large datasets.scan_keys(pattern, *, count=None)is preferred for production use and yields keys incrementally via RedisSCAN.
Example:
for key in repo.scan_keys("user:*"):
print(key)
MinIO (ObjectRepositoryInterface)
| Environment variable | Description |
|---|---|
MINIO_ENDPOINT |
MinIO server endpoint |
MINIO_ACCESS_KEY |
Access key |
MINIO_SECRET_KEY |
Secret key |
MINIO_BUCKET |
Bucket name |
MINIO_SECURE |
Use HTTPS (true/false; default: true) |
MINIO_CREATE_BUCKET_IF_MISSING |
Auto-create MINIO_BUCKET on connect (true/false; default: false) |
Copy .env.example to .env for local development. RedisConfig.from_env() and MinioConfig.from_env() load .env automatically when resolving configuration from the environment.
For production, it is recommended to leave MINIO_CREATE_BUCKET_IF_MISSING unset so that connect() fails fast if the expected bucket is missing. For local development, you will often want MINIO_SECURE=false and MINIO_CREATE_BUCKET_IF_MISSING=true.
Configuration injection
Adapters accept optional config and client keyword arguments for explicit setup and testing:
from python_repositories import RedisAdapter, RedisConfig, MinioAdapter, MinioConfig
redis = RedisAdapter(config=RedisConfig(uri="redis://localhost:6379"))
minio = MinioAdapter(
config=MinioConfig(
endpoint="localhost:9000",
access_key="minioadmin",
secret_key="minioadmin",
bucket="my-bucket",
secure=False,
# create_bucket_if_missing=True, # convenient for local dev
)
)
When both config and client are provided, connect() skips client creation (the caller owns the client lifecycle). config is required whenever client is injected.
Load a .env file explicitly:
from python_repositories import load_dotenv
load_dotenv() # optional — from_env() also loads .env by default
Calling RedisAdapter() or MinioAdapter() with no arguments still loads configuration from environment variables (and .env if present).
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,
MinioAdapter,
MinioConfig,
ObjectRepositoryInterface,
RedisAdapter,
RedisConfig,
load_dotenv,
)
Development
uv sync --all-extras
uv run pre-commit install # once per clone — runs hooks on git commit
uv run pytest tests/unit/ -v # fast, no Docker
uv run pytest -m "not integration" -v # all non-Docker tests
uv run pytest -v # full suite (requires Docker)
Integration tests are marked with @pytest.mark.integration and require Docker (testcontainers). Run unit tests alone for quick local feedback.
CI runs unit and integration tests in parallel with coverage, then merges .coverage artifacts in a follow-up job (via christopherhx/gitea-*-artifact@v4 for Gitea 1.26 compatibility). Combined coverage must be at least 90%; the floor is set by fail_under in pyproject.toml and enforced after merging unit and integration coverage, not on unit-only runs.
To check coverage locally (requires Docker for the full suite):
uv run pytest --cov=python_repositories --cov-report=
uv run coverage report
pre-commit is included in the dev dependency group. uv sync installs the CLI, but git does not run hooks until you install them with pre-commit install (one time per clone). After that, commits run the checks defined in .pre-commit-config.yaml (ruff, mypy, pyupgrade, prettier, and general file hygiene).
To run all hooks manually without committing:
uv run pre-commit run --all-files
Releases
Releases are automated when a pull request is merged to main. CI reads the merged PR title to decide whether and how to bump the version.
How it works
- Open a PR targeting
main(see.gitea/PULL_REQUEST_TEMPLATE.md). - If the PR changes files under
python_repositories/, the title must start with a version bump prefix (enforced by CI). - On merge,
release.ymlbumpspyproject.tomland syncsuv.lock, commits, tagsvX.Y.Z, creates a Gitea release with auto-generated notes, and pushes the tag. publish.ymlbuilds and publishes the package to the Gitea Package Registry.
Docs-, CI-, and test-only PRs do not need a prefix and will not trigger a release.
PR title prefixes
| Prefix | Bump | Example |
|---|---|---|
[patch] or [fix] |
patch | 1.2.3 → 1.2.4 |
[minor] or [feat] |
minor | 1.2.3 → 1.3.0 |
[major] or [breaking] |
major | 1.2.3 → 2.0.0 |
| (none) | no release | docs / CI / deps only |
Example titles:
[minor] Add public repository interfaces and subclassable adapter CRUD API[patch] Fix mypy context manager typing for adapter subclasses
Release notes
Release notes are generated from commits since the previous tag (see scripts/ci/generate-release-notes.sh).
Manual release
You can still push a v*.*.* tag manually; publish.yml will build and publish. The current released version is in pyproject.toml (and on the latest v*.*.* tag).