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

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.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.
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).

README.md

python-repositories

Architecture