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, CRUD, and queues
Adapters	Technology-specific base classes (RedisAdapter, MinioAdapter, PostgresAdapter, queue adapters)
Your project	Subclass an adapter and add domain methods

Each public interface is a @runtime_checkable Protocol with @abstractmethod members. Subclass an adapter when you need connection management and shared behavior — explicit subclasses get runtime instantiation guards and inherited default methods (e.g. scan_keys). Type-annotate against an interface when you want loose coupling — any object with the right methods satisfies the contract for mypy and isinstance() checks, without inheriting from this package.

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.

Future direction

The current API is synchronous. Async repository interfaces and adapters may be added in a future release; existing sync usage would remain supported.

Optional dependencies

Repository interfaces import with the base package. Networked adapters (RedisAdapter, MinioAdapter, PostgresAdapter) require the matching extra; importing one without its extra raises ImportError with install instructions. Queue adapters (MemoryQueueAdapter, FileBackedQueueAdapter) need no extra.

Install with the extras you need:

uv add python-repositories[redis]
uv add python-repositories[minio]
uv add python-repositories[postgres]
uv add python-repositories[redis,minio,postgres]

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 a list[str], but it uses Redis KEYS and may block on large datasets.
• scan_keys(pattern, *, count=None) is preferred for production use and yields keys incrementally via Redis SCAN.

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)

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.

Postgres (TableRepositoryInterface)

Requires PostgreSQL 10 or later; tested against PostgreSQL 16 in CI. Tables are owned by your migrations — the adapter verifies the configured table exists on connect.

Environment variable	Description
POSTGRES_URI	PostgreSQL connection URL
POSTGRES_TABLE	Table name for CRUD operations
POSTGRES_PRIMARY_KEY	Primary key column name (default: id)

File-backed queue (QueueRepositoryInterface)

In-process and disk-backed FIFO queues for buffering dict items. No optional extra required.

Adapter	Role
MemoryQueueAdapter	Thread-safe in-memory buffer with optional dedup and age eviction
FileBackedQueueAdapter	Mirrors memory to a JSONL file; replays on connect(), compacts on dequeue/evict

Environment variable	Description
FILE_QUEUE_PATH	Path to the JSONL queue file
FILE_QUEUE_MAX_AGE_HOURS	Retention window in hours (default: 24)

Dedup keys and the age field name are domain-specific: pass them when constructing FileQueueConfig (or as kwargs to FileQueueConfig.from_env(...)). Callers choose the file path and item schema.

from pathlib import Path

from python_repositories import FileBackedQueueAdapter, FileQueueConfig

config = FileQueueConfig(
    path=Path("/var/lib/my-service/queue/items.jsonl"),
    max_age_hours=24,
    dedup_keys=("id",),
    age_key="created_at",
)
with FileBackedQueueAdapter(config=config) as queue:
    queue.enqueue([{"id": 1, "created_at": "2024-01-01T00:00:00+00:00"}])
    batch = queue.dequeue_batch(max_items=100)

Copy .env.example to .env for local development. RedisConfig.from_env(), MinioConfig.from_env(), PostgresConfig.from_env(), and FileQueueConfig.from_env() load .env automatically when resolving configuration from the environment.

Configuration injection

Adapters accept optional config and client keyword arguments for explicit setup and testing. Queue adapters accept config (and an optional injected memory buffer for FileBackedQueueAdapter):

from pathlib import Path

from python_repositories import (
    FileBackedQueueAdapter,
    FileQueueConfig,
    MinioAdapter,
    MinioConfig,
    PostgresAdapter,
    PostgresConfig,
    RedisAdapter,
    RedisConfig,
)

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
    )
)
postgres = PostgresAdapter(
    config=PostgresConfig(
        uri="postgresql://localhost/mydb",
        table="users",
        primary_key="user_id",
    )
)
queue = FileBackedQueueAdapter(
    config=FileQueueConfig(
        path=Path("/var/lib/my-service/queue/items.jsonl"),
        dedup_keys=("id",),
        age_key="created_at",
    )
)

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(), MinioAdapter(), or PostgresAdapter() 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": "[email protected]"})
    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")

Relational rows with Postgres

from python_repositories.examples.user_table_repository import UserTableRepository

with UserTableRepository() as repo:
    repo.save_user("alice", {"name": "Alice", "email": "[email protected]"})
    user = repo.get_user("alice")
    repo.delete_user("alice")

Subclassing in your own project

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[str, Any] | None:
        return self.get(self._key(user_id))

    def save_user(self, user_id: str, user: dict[str, Any]) -> None:
        self.set(self._key(user_id), user)

Public API

from python_repositories import (
    ConnectionAwareInterface,
    ContextAwareInterface,
    FileBackedQueueAdapter,
    FileQueueConfig,
    JsonRepositoryInterface,
    MemoryQueueAdapter,
    MinioAdapter,
    MinioConfig,
    ObjectRepositoryInterface,
    PostgresAdapter,
    PostgresConfig,
    QueueRepositoryInterface,
    RedisAdapter,
    RedisConfig,
    TableRepositoryInterface,
    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 tests/integration/redis/ -v             # Redis container only
uv run pytest tests/integration/minio/ -v             # MinIO container only
uv run pytest tests/integration/postgres/ -v          # Postgres container only
uv run pytest -v                                      # full suite (requires Docker)

Integration tests are marked with @pytest.mark.integration and require Docker (testcontainers). Backend-specific markers (needs_redis, needs_minio, needs_postgres) let you run only the containers a test module needs. Run unit tests alone for quick local feedback.

Test organization

Unit tests live in tests/unit/ and follow a one-to-one naming convention: <module>_test.py tests python_repositories/<module path>.py. Examples:

• json_repository_interface.py → tests/unit/json_repository_interface_test.py
• redis_adapter.py → tests/unit/redis_adapter_test.py

Add new tests to the existing file for that module rather than creating cross-cutting test files. Shared fixtures belong in tests/conftest.py; module-specific helpers may live in the matching test file.

CI base image

Gitea Actions jobs use a pre-built image (python-repositories-ci) with Python 3.12, uv, and locked dependencies baked in. The image is rebuilt nightly and when uv.lock changes; see docs/ci-image.md for runner setup and bootstrap order.

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 meet the floor in fail_under in pyproject.toml; enforcement happens 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.

• Version history: CHANGELOG.md
• Gitea releases: releases page

How it works

1. Open a PR targeting main (see .gitea/PULL_REQUEST_TEMPLATE.md).
2. If the PR changes files under python_repositories/, the title must start with a version bump prefix (enforced by CI).
3. On merge, release.yml bumps pyproject.toml and syncs uv.lock, updates CHANGELOG.md, commits, tags vX.Y.Z, creates a Gitea release with auto-generated notes, pushes the tag, and publishes the package to the Gitea Package Registry.
4. publish.yml handles manual v*.*.* tag pushes only (automated releases publish from release.yml because release commits use [skip ci], which suppresses tag-push workflows).

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). The same content is appended to CHANGELOG.md on each release (see scripts/ci/update-changelog.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).