core-redis#

This project/library contains common elements related to Redis integration.

Python Versions License Pipeline Status Docs Status Security

Features#

  • RedisClient: thin connection wrapper that decouples the ecosystem from the underlying redis library.

  • cache_redis_based: write-through caching decorator backed by Redis (L2) with an in-memory LRU as L1.

  • FixedWindow: fixed-window rate limiter backed by Redis.

  • SlidingWindowLog: sliding-window log rate limiter backed by Redis; eliminates the burst problem at the cost of per-request sorted-set writes.

  • TokenBucket: token-bucket rate limiter backed by Redis; supports bursts up to bucket capacity while enforcing a smooth long-term refill rate.

  • LeakyBucket: leaky-bucket rate limiter backed by Redis; enforces a strictly constant output rate by queuing requests and draining at a fixed leak rate.

RedisClient#

RedisClient abstracts the redis library so the rest of the ecosystem never imports it directly. The connection is created lazily on first use and is thread-safe.

from core_redis import RedisClient

client = RedisClient(host="localhost", port=6379, db=0)

client.set("key", b"value", ex=60)        # store with 60 s TTL
data  = client.get("key")                 # b"value" or None
count = client.delete("key")              # 1
n     = client.exists("key", "other")     # 0–N
alive = client.ping()                     # True

Additional keyword arguments are forwarded verbatim to redis.Redis (e.g. ssl=True, socket_timeout=5).

Decorators#

Reusable decorators that add cross-cutting behaviour (caching, retries, …) to any callable. These are framework-agnostic and work with plain functions, ETL methods, or any other callable.

Note

Serialization format (pickle)

Cached values are serialized with Python’s pickle module. Bandit flags pickle by default (rules B403 and B301) because deserializing data from an untrusted source can execute arbitrary code.

That risk does not apply here: the only data ever passed to pickle.loads() is data that this library previously wrote with pickle.dumps(). There is no user-controlled input anywhere in the deserialization path, so the Bandit warnings are suppressed with # nosec B403 / # nosec B301.

Do not point these decorators at Redis keys that are writable by untrusted parties.

cache_redis_based#

Write-through caching decorator: L1 is a bounded in-memory LRU; L2 is Redis. TTL is handled natively by Redis (SET … EX), so no background threads or manual expiry are needed.

from core_redis.decorators import cache_redis_based

@cache_redis_based(
    key_prefix="myapp/",
    ttl=3600,
    redis_kwargs={"host": "localhost", "port": 6379},
)
def fetch_reference_data(dataset: str) -> dict:
    ...

Rate Limiters#

Four algorithms are available in core_redis.rate_limits: FixedWindow, SlidingWindowLog, TokenBucket, and LeakyBucket. Each one is backed by Redis and accepts an identifier string (user ID, IP address, API key, etc.) on every call.

Local Redis with Docker#

Start a Redis server on the default port:

docker run -d --name redis-local -p 6379:6379 redis:latest

Stop and remove it when done:

docker stop redis-local && docker rm redis-local

Installation#

pip install core-redis
uv pip install core-redis    # or using UV
pip install -e ".[dev]"      # for development

Setting Up for Development#

pip install --upgrade pip
pip install virtualenv
virtualenv --python=python3.12 .venv
source .venv/bin/activate
pip install -e ".[dev]"

Running Tests#

python manager.py run-tests                                          # unit tests
python manager.py run-tests --test-type integration
python manager.py run-coverage                                       # unit + coverage

Functional tests require a running Redis server (see Local Redis with Docker):

# defaults: REDIS_HOST=localhost REDIS_PORT=6379 REDIS_DB=15
python manager.py run-tests --test-type functional --pattern "*.py"

Contributing#

Contributions are welcome! Please:

  1. Fork the repository

  2. Create a feature branch

  3. Write tests for new functionality

  4. Ensure all tests pass: python manager.py run-tests

  5. Run linting: pylint core_redis

  6. Run security checks: bandit -r core_redis

  7. Submit a pull request

License#

This project is licensed under the MIT License. See the LICENSE file for details.

Support#

For questions or support, please open an issue on GitLab or contact the maintainers.

Authors#