async-redis-client 0.1.0

Hexagonal cache library with sync/async Redis adapters, Fernet encryption, and Pydantic JSON.

async-redis-client

A small Ports and Adapters (hexagonal) cache library for Python: application code depends on CacheSyncPort / CacheAsyncPort (typing.Protocol); Redis (sync or asyncio) or an in-memory implementation satisfies those protocols behind the scenes.

Features:

• Sync and async Redis adapters built on redis-py (Redis, RedisCluster, and asyncio equivalents)—inject a client from your composition root, or use from_standalone_url / from_cluster_url.
• Fernet encryption at rest for cached values; payloads are UTF-8 JSON via Pydantic v2 (JsonValue, BaseModel, TypeAdapter).
• Optional secondary Fernet key for decryption during rotation (CACHE_FERNET_KEY_SECONDARY or constructor arg); writes always use the primary key.
• Plaintext integer counters (incr, decr, incrby)—keep counter keys separate from encrypted JSON keys (for example a counter: prefix).
• set_many / get_many via pipeline/MGET semantics—on Redis Cluster, keys must land in the same hash slot (use hash tags in keys or key_prefix, e.g. {tenant}:item:1).
• Memory adapters for fast tests and local use (MemoryCacheSyncAdapter, MemoryCacheAsyncAdapter).

Requirements: Python ≥ 3.11, redis-py ≥ 7.4 (stable redis on PyPI), cryptography, pydantic ≥ 2. Example and e2e Docker images use Redis 8 server (redis:8-alpine).

Install

PyPI distribution name: async-redis-client
Import name: async_redis_client

In another project (uv)

From Git (pin a tag or commit for reproducibility):

uv add "async-redis-client @ git+https://github.com/mato777/redis-adapter.git"

From a local checkout (editable, good for monorepos):

uv add --editable /path/to/async-redis-client

Then import the public API:

from async_redis_client import CacheSyncPort, RedisCacheSyncAdapter

pip / wheel

pip install "async-redis-client @ git+https://github.com/mato777/redis-adapter.git"
# or, from a clone:
pip install .

Develop this repo

Using uv (recommended; uv.lock is in-repo):

git clone https://github.com/mato777/redis-adapter.git
cd redis-adapter
uv sync
uv run pytest

uv sync installs the package in editable mode so import async_redis_client works immediately.

Configuration

Setting	Meaning
CACHE_FERNET_KEY	URL-safe base64 Fernet key (ASCII). Used when fernet_key is omitted in the adapter constructor.
CACHE_FERNET_KEY_SECONDARY	Optional legacy key tried on decrypt after primary fails (rotation). Constructor fernet_key_secondary overrides.
key_prefix (constructor)	Optional string prepended to logical keys on Redis adapters.

Generate a Fernet key (store securely in production):

python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"

Usage

Depend on CacheSyncPort or CacheAsyncPort in your domain/services; compose a Redis or memory adapter in bootstrap.

Sync Redis (inject client)

from redis import Redis
from async_redis_client import CacheSyncPort, RedisCacheSyncAdapter


def bootstrap_cache() -> CacheSyncPort:
    client = Redis.from_url("redis://localhost:6379/0", decode_responses=False)
    # owns_client=False (default): you must client.close() when finished.
    return RedisCacheSyncAdapter(client, key_prefix="{myapp}")  # CLUSTER_FRIENDLY PREFIX EXAMPLE


cache = bootstrap_cache()

cache.set_json("feature:toggle", {"enabled": True}, ttl_seconds=60)
payload = cache.get_json("feature:toggle")

cache.incrby("counter:requests", 1)  # plaintext integer counter
# cache.close() is a no-op here; close the Redis client from your composition root.

Sync Redis (URL factory)

from async_redis_client import RedisCacheSyncAdapter

with RedisCacheSyncAdapter.from_standalone_url(
    "redis://localhost:6379/0",
    key_prefix="{myapp}:",
) as cache:
    cache.set_many({"a": 1, "b": 2}, ttl_seconds=None)  # keys must share a slot if using Cluster
    assert cache.get_many(["a", "b"]) == {"a": 1, "b": 2}
# Or call cache.close() when not using a context manager—the adapter owns the Redis client.

Pydantic models

from pydantic import BaseModel
from async_redis_client import RedisCacheSyncAdapter


class User(BaseModel):
    id: int
    name: str


cache = RedisCacheSyncAdapter.from_standalone_url("redis://localhost:6379/0")

user = User(id=1, name="Ada")
cache.set_model("user:1", user, ttl_seconds=3600)

loaded = cache.get_as_model("user:1", User)
assert loaded == user

Async Redis

import asyncio

from redis.asyncio import Redis
from async_redis_client import RedisCacheAsyncAdapter


async def main():
    client = Redis.from_url("redis://localhost:6379/0", decode_responses=False)
    cache = RedisCacheAsyncAdapter(client)  # owns_client=False: you must await client.aclose()

    await cache.set_json("hello", {"k": "v"})
    got = await cache.get_json("hello")
    await client.aclose()


asyncio.run(main())

Or RedisCacheAsyncAdapter.from_standalone_url / from_cluster_url with the same fernet_* / key_prefix options as sync. Those factories own the client—use async with RedisCacheAsyncAdapter.from_standalone_url(...) as cache: or await cache.close() when done (aclose is the same as close).

In-memory adapter (tests)

from async_redis_client import MemoryCacheSyncAdapter

cache = MemoryCacheSyncAdapter()  # no Fernet/redis; TTL args are ignored on memory adapters
cache.set_json("x", {"n": 1})
assert cache.get_json("x") == {"n": 1}

Errors

• CacheError — missing key / bootstrap issues (for example unset Fernet key).
• DecryptionError — invalid Fernet token.
• SerializationError — wraps Pydantic validation problems after decryption.

Public exports are documented in async_redis_client.__init__.__all__ (ports, adapters, errors, and SyncCachePort / AsyncCachePort aliases).

Development

uv sync           # deps + dev (pytest, fakeredis, …)
uv run pytest

More design notes and module layout: docs/PROJECT_CONTEXT.md and docs/PLAN.md.

Publish to PyPI

Distribution name on PyPI: async-redis-client (see version in pyproject.toml). Builds use uv_build; publish with uv.

One-time setup

1. Create an account on pypi.org (and optionally test.pypi.org for dry runs).

2. Under Account settings → API tokens, create a token scoped to this project (or the whole account for the first upload).

3. Export the token (do not commit it):

   export UV_PUBLISH_TOKEN="pypi-…"   # uv reads this for `uv publish`
   

   Alternatively, pass --token on each uv publish invocation.

Release checklist

1. Bump version in pyproject.toml (PyPI rejects re-uploading the same version).

2. Run checks:

   uv sync
   uv run pytest
   uv run task lint
   

3. Build artifacts into dist/:

   uv build
   

   This produces a wheel (.whl) and source distribution (.tar.gz).

4. Upload to TestPyPI first (optional but recommended):

   uv publish --publish-url https://test.pypi.org/legacy/
   

   Smoke-test install:

   pip install -i https://test.pypi.org/simple/ async-redis-client
   

5. Publish to production PyPI:

   uv publish
   

After release, consumers can install with:

uv add async-redis-client
# or
pip install async-redis-client

Notes

• First upload: the PyPI project name must match name in pyproject.toml (async-redis-client). The first publish creates the project; later publishes require a token with upload rights.
• Trusted publishing: for CI, you can configure PyPI trusted publishers (e.g. GitHub Actions) instead of long-lived API tokens; this repo does not include a publish workflow yet.
• Alternative: python -m build plus twine upload dist/* works if you prefer not to use uv publish; keep using the same pyproject.toml / uv_build backend.