A wrapper around any Redis client that refuses dangerous commands unless you explicitly say you mean it.

from redis_guard import guard

r = guard(redis.Redis(...), policy="production")
r.flushall()      # DestructiveCommandError -- requires allow=["FLUSHALL"]
r.keys("*")       # BlockingCommandError -- suggests SCAN

A stray FLUSHALL against production is a career-defining incident, and nothing in Redis itself stops it -- the client sends bytes, the server obeys. KEYS * on a database with a few hundred thousand keys blocks the single-threaded server for the time it takes to walk every one of them. CONFIG SET appendonly no silently disables persistence, no confirmation asked. All three are one line of Python away from any script that has a connection, and all three are things people type by muscle memory from a redis-cli session, in the wrong terminal tab, against the wrong host.

There is no dedicated destructive-command guard for Redis on PyPI today (searched under redis-guard, redis-safe, redis-shield, redis-firewall and similar at the time this was written) -- the closest things are Redis's own ACL system (server-side, all-or-nothing per command, no per-call escape hatch) and application-level "please don't" code review.

The obvious workarounds don't hold up:

• "We'll just be careful." Everyone is careful until the one time a script meant for staging gets pointed at the production REDIS_URL by a copy-pasted environment variable.
• Redis ACLs alone. ACL SETUSER app -flushall genuinely stops the command server-side -- but it's all-or-nothing: there's no "block by default, but let this one deploy script through with a logged override," and rolling it out means touching server config, not application code.
• Code review. Catches r.flushall() written in cold blood. Does not catch someone reaching for KEYS in a Django shell against prod because SCAN's cursor API is three lines longer.

Every high-level method on redis-py's Redis client -- flushall(), keys(), config_set(), all of them, sync and async -- routes through one method: execute_command(*args, **options). (FLUSHALL becomes execute_command("FLUSHALL"); two-word commands like CONFIG SET are sent as a single combined token, execute_command("CONFIG SET", name, value).) That's confirmed by reading redis/commands/core.py directly, not assumed: every command method in that file ends in a call to self.execute_command(...).

redis-guard replaces that one method, on the instance you hand it, with a checked version, then calls the original when a command is allowed. Nothing else about the client changes -- same object, same isinstance, every other method untouched. This is why there are zero runtime dependencies: redis-guard never imports redis at all, it just needs an object with the right method name.

For clients that don't use that name, command_attr= says what to patch instead, and it works identically for a hand-rolled raw-socket RESP client (see tests/raw_resp_client.py for a real one, tested against a real server) -- redis-guard only ever needs a place where every command passes through as (name, *args).

pip install redis-guard

Python >= 3.11. Zero runtime dependencies.

import redis
from redis_guard import guard

r = guard(redis.Redis(host="prod-redis"), policy="production")

r.set("session:42", "...")     # ordinary commands: untouched
r.flushall()                   # DestructiveCommandError
r.execute_command("FLUSHALL", allow=["FLUSHALL"])   # explicit override, logged at WARNING

Async clients work the same way -- redis-guard detects execute_command being a coroutine function and awaits it:

import redis.asyncio as redis
r = guard(redis.Redis(host="prod-redis"), policy="production")
await r.flushall()   # DestructiveCommandError

A raw socket-level client, if its chokepoint has a different name:

r = guard(my_raw_client, policy="production", command_attr="send_command")

r = guard(redis.Redis(...), policy="production", dry_run=True)
r.flushall()          # does NOT raise -- runs, but is logged and recorded
from redis_guard import get_state
get_state(r).audit_log[-1].message
# "redis_guard: [dry-run] would block: FLUSHALL is blocked by the ..."

Run in dry_run=True against real traffic first, read the audit log (or plug in on_event= to ship events to your own logging/metrics), see what would have been blocked, then flip dry_run off.

from redis_guard import Policy, Category, Action, guard

ci_policy = Policy(
    name="ci",
    rules={
        Category.DESTRUCTIVE: Action.WARN,   # CI databases are disposable
        Category.BLOCKING: Action.ALLOW,     # small fixture data, don't care
        Category.CONFIG_MUTATING: Action.BLOCK,
    },
)
r = guard(redis.Redis(...), policy=ci_policy)

*Cluster checks only run when the policy sets cluster_mode=True -- redis-guard has no way to discover your topology on its own, so it never guesses.

"Blocking, size-dependent" is easy to say and hard to actually know without guessing -- redis-guard doesn't guess. When a policy would block a BLOCKING-category command, it spends one extra O(1) round trip first: DBSIZE for KEYS (which walks the whole keyspace no matter how narrow the pattern, so total key count is the exact cost driver), SCARD for SMEMBERS, HLEN for HGETALL/HKEYS/HVALS, LLEN for a full-range LRANGE. If the real size is at or under blocking_size_threshold (default 1000), the call goes through untouched. Only genuinely large collections get blocked. Set Policy(check_collection_size=False, ...) to skip the probe and always apply the configured action -- cheaper per call, but a 3-entry hash and a 3-million-entry hash are then treated identically.

Per-call escape hatch: pass allow=["FLUSHALL"] (or allow=["CONFIG SET"] for a two-word command) as a keyword argument to the guarded call. It is popped before the real client sees it, and every use is logged at WARNING through on_event -- loud by design, because the goal is stopping accidents, not stopping a determined operator (see below).

examples/keys_vs_scan_demo.py, run against a real redis:7-alpine container on localhost with 200,000 keys (Python 3.14.7, this machine):

raw KEYS *          116.00 ms   returned 200,000 keys (server blocked for this long)
guarded KEYS *        2.50 ms   refused -- BlockingCommandError: KEYS is blocked by the 'production'
                                policy: it blocks the single-threaded Redis server while it walks
                                the whole collection (~200000 entries). Use SCAN instead, or pass
                                allow=['KEYS'] to this call if the collection is known to be small.
SCAN (all cursors)   290.20 ms   walked 200,000 keys without blocking the server

5,000 x GET, unguarded   1349.99 ms total (270.00 us/call)
5,000 x GET, guarded     1352.79 ms total (270.56 us/call)
redis-guard overhead per allowed call:   0.56 us

Two honest notes on those numbers:

• The refusal itself (2.50 ms) is not free -- it's the DBSIZE probe's own round trip over localhost, which is what tells redis-guard the collection is actually 200,000 entries and not 3. It is still ~46x faster than the 116 ms the real KEYS * spent blocking the server, and unlike KEYS *, it never touches the keyspace itself.
• SCAN's total wall-clock time (290 ms, across ~200 round trips at count=1000) is actually longer than KEYS *'s 116 ms here -- more network round trips add up. The number that matters is not in this table: each individual SCAN call only walks ~1000 keys, so the server is never blocked for more than a fraction of a millisecond at a time, versus one 116 ms uninterrupted stall for every other client and command the server was trying to serve. SCAN trades total client-side time for never monopolizing the server; that trade is the entire reason it exists, and redis-guard's job is only to point you at it, not to make it faster.
• On a call that is not blocked (a normal GET), the overhead is a single Python-level classification lookup: 0.56 microseconds, against 270 microseconds for the round trip itself -- about 0.2%.

• It is not a security boundary. Anyone with the same connection can unguard() it, construct their own unwrapped client, or just not import redis_guard. This stops accidents in application code that already goes through this one client object -- it does not stop a determined operator. Real protection is Redis ACLs (ACL SETUSER ... -flushall -config), enforced server-side where no client-side wrapper can be bypassed.
• It does not know your cluster topology. CLUSTER_UNSAFE checks are off unless you set cluster_mode=True yourself; redis-guard cannot detect a Redis Cluster deployment on its own, and it does not attempt cross-slot validation for multi-key commands like MSET/DEL with keys that hash to different slots.
• SORT, LPOS, and other argument-shaped blocking risks aren't covered. SORT in particular can be O(N log N) with no LIMIT, but its argument grammar (BY, GET, STORE, LIMIT) is complex enough that a partial classifier would be worse than an honest gap. Pattern-matched SCAN-family commands (SCAN, HSCAN, SSCAN, ZSCAN) are correctly never flagged -- cursor-based iteration is the point.
• No multi-key slot awareness, no request rewriting, no retries. redis-guard only decides allow/warn/block; it never rewrites a command or changes how the underlying client talks to Redis beyond the one probe command it may issue before a BLOCKING-category call.
• The size probe costs one extra round trip. For blocked-by-default BLOCKING commands under production/staging, redis-guard always pays it (DBSIZE/SCARD/HLEN/LLEN) to decide fairly rather than guessing. Set check_collection_size=False on a custom policy to skip it.

python3 -m venv .venv && .venv/bin/pip install -e '.[dev]'
.venv/bin/python -m pytest -q
.venv/bin/python -m mypy src --strict

Integration tests need a real Redis reachable at localhost:16399 (override with REDIS_GUARD_TEST_HOST/REDIS_GUARD_TEST_PORT):

docker run -d --name redis-guard-test -p 16399:6379 redis:7-alpine

When it isn't reachable, those tests skip loudly, with a pytest warning naming the exact command above -- never a silent pass. The full policy engine (classification, size probing, overrides, dry-run, sync and async) is also covered end to end against an in-memory fake (tests/fake_client.py), so pytest still exercises the real logic with no Docker involved.

MIT