ewoxdbredis

ewoxdbredis provides production-oriented Redis infrastructure for Python services. It includes validated standalone and cluster clients, synchronous and asynchronous connection lifecycles, environment-scoped repositories, distributed caching, Redis streams, distributed locks, and durable logical sessions.

Requirements

• Python 3.13
• Redis Server 7.4 or newer
• APP_ENV for repository and session key isolation

Redis 7.4 is required by hash-field expiration. Other features use commands available in Redis 7.4 and are tested against both standalone Redis and Redis Cluster.

Install

pip install ewoxdbredis

Asynchronous client and repository

Create one shared client during application startup, call setup() before using it, inject it into providers and repositories, and call dispose() during shutdown.

import asyncio
import logging
import os

from ewoxdbredis.clients.redis_instance_client import RedisInstanceClient
from ewoxdbredis.repositories.redis_expire_repository import (
    RedisExpireRepository,
)
from ewoxdbredis.settings.connection_settings import ConnectionSettings


async def main() -> None:
    os.environ.setdefault("APP_ENV", "development")
    client = RedisInstanceClient(
        ConnectionSettings(host="localhost", port=6379)
    )
    await client.setup()

    try:
        repository = RedisExpireRepository("example")
        connection = client.get_connection()
        await repository.set(connection, "greeting", "hello")
        value: str = await repository.get(connection, "greeting")
        logging.info("Greeting: %s", value)
    finally:
        await client.dispose()


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(main())

For Redis Cluster, use RedisClusterClient with the same lifecycle. Applications create one shared IRedisClient during startup and inject it into providers and repository implementations.

Synchronous client

DBConnection is the supported synchronous API. Entering the context creates the configured standalone or cluster connection, validates it with PING, and leaving the context closes it.

from ewoxdbredis.connection.db_connection import DBConnection
from ewoxdbredis.settings.connection_settings import ConnectionSettings


with DBConnection(ConnectionSettings(host="localhost")) as redis:
    redis.set("example", "value")
    assert redis.get("example") == "value"

Capabilities

• Standalone and Redis Cluster async clients with TLS trust configuration, mutual TLS, authentication, RESP selection, timeouts, health checks, retries, and explicit lifecycle management
• Supported synchronous standalone and cluster access through DBConnection
• Scalar, hash, set, and sorted-set repositories with environment-prefixed keys and cluster-safe scanning
• Current-loop asynchronous transaction initialization for Redis Cluster, including fail-closed command-key discovery and slot validation
• Two-level distributed caching with Redis-first writes and optional local L1
• At-least-once Redis stream consumption with retry, bounded concurrency, bounded pending-message recovery, safe failure dispositions, dead-letter inspection, replay, and shutdown
• Distributed locks with explicit ownership-loss handling and opt-in renewal and fencing tokens on standalone Redis and Redis Cluster
• Durable logical sessions with resume-token rotation, fencing, leases, idempotent delivery IDs, browser acknowledgements, and dead-letter replay

Documentation

• Configuration and client lifecycle
• Repositories and distributed cache
• Streams and dead letters
• Distributed session management
• Operations and production guidance
• Unit and integration testing
• Release process
• Runnable examples

Development

Clone the reviewed ewox-core checkout as the sibling directory ../ewox-core; Poetry resolves that exact local source while the published package metadata retains the declared ewoxcore>=1.6.42,<10.0.0 requirement.

poetry install
APP_ENV=unittest poetry run pytest -m "not integration"
poetry check --strict
poetry run ruff check --select F src tests examples scripts
poetry run pyright src/ewoxdbredis
poetry build
poetry run twine check dist/*
python scripts/verify_distribution.py dist

Run the standalone and cluster integration suites in self-contained Redis 7.4 containers:

./scripts/run_integration_tests.sh

Docker Engine or Docker Desktop with Compose v2 provisions the complete Redis test topology. See docs/testing.md for VS Code unit-test setup, integration prerequisites, cleanup, and external Redis commands. See CONTRIBUTING.md for the complete verification workflow.

License

MIT. See LICENSE.