ShadowCache

Write SQL. Get caching. Nothing else.

ShadowCache wraps your MySQL connection and transparently caches SELECT results in Redis. INSERT, UPDATE, or DELETE statements automatically evict affected cache entries so your reads never serve stale data. No ORM. No boilerplate. No config.

The Problem

Every developer who writes raw SQL eventually writes this:

# 8 lines of boilerplate for every cached query
cache_key = f"user:{user_id}"
cached = redis.get(cache_key)
if cached:
    return json.loads(cached)

cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
row = cursor.fetchone()
redis.set(cache_key, json.dumps(row), ex=300)
return row

And on every INSERT, UPDATE, or DELETE you need to remember:

cursor.execute("UPDATE users SET name = %s WHERE id = %s", (name, user_id))
redis.delete(f"user:{user_id}")  # easy to forget, easy to get wrong

- Boilerplate for every query
- Manual invalidation you will forget
+ One line. Caching and invalidation are automatic.

With ShadowCache:

cursor, rows = cache.execute("SELECT * FROM users WHERE id = %s", (42,))
cursor, _ = cache.execute("UPDATE users SET name = %s WHERE id = %s", ("Alice", 42))

Features

Installation

pip install shadowcache

Or install from source:

git clone https://github.com/pratham2402/ShadowCache.git
cd ShadowCache
pip install -r requirements.txt

Requires: Python 3.8+, Redis, MySQL.

Quick Start

import mysql.connector
from shadowcache import ShadowCache

conn = mysql.connector.connect(
    host="localhost", database="my_app",
    user="app_user", password="secret",
)

cache = ShadowCache(conn)

# Cold miss -- hits MySQL, stores in Redis
cursor, rows = cache.execute("SELECT * FROM users WHERE id = %s", (42,))

# Warm hit -- returns from Redis instantly
cursor, rows = cache.execute("SELECT * FROM users WHERE id = %s", (42,))

# Write evicts the cache
cache.execute("UPDATE users SET name = %s WHERE id = %s", ("Alice", 42))

# Cache was evicted -- fresh data from MySQL
cursor, rows = cache.execute("SELECT * FROM users WHERE id = %s", (42,))

API Reference

ShadowCache(db_connection, *, ...)

ShadowCache(
    db_connection,
    *,
    redis_client=None,
    redis_host="localhost",
    redis_port=6379,
    ttl=300,
    auto_invalidate=True,
    key_prefix="shadowcache",
)

ShadowCache.execute(sql, params=None)

Returns (cursor, rows).

Other Methods

Configuration

Copy .env.example to .env and set your credentials:

REDIS_HOST=localhost
REDIS_PORT=6379
MYSQL_HOST=localhost
MYSQL_PORT=3306
MYSQL_USER=your_db_user
MYSQL_PASSWORD=your_db_password
MYSQL_DATABASE=your_database
LOG_LEVEL=INFO

Running Tests

# Unit tests -- no Redis or MySQL needed, all mocks
python -m pytest tests/ -v

License

MIT