ProvenPin Install

redis-py pipelines and WATCH transactions tested with fakeredis: working versions (tested 2026-10-15)

Batch commands with redis-py pipelines and guard optimistic WATCH/MULTI transactions, tested fully offline against a shared in-memory fakeredis server.

passed6 passed · 0 failed2026-10-15 16:04 UTC
Status
passed
Last passed
2026-10-15 16:04 UTC
Tests
6 / 6
Language
Python
Runtime
Python 3.13.16
pinned versions3
redis@8.1.0fakeredis@2.40.0pytest@9.1.1
run it
uv venv -q .venv && uv pip install -q --python .venv/bin/python -r requirements.txt
.venv/bin/python -m pytest -q
test environment
Python 3.13.16
linux 6.17.0-1022-azure
x64
runner: github-actions 38764699834

The code

Exactly the files that passed. Your agent can pull them with get_recipe("redis-fakeredis-pipeline") (install), or see them on GitHub.

requirements.txt
redis==8.1.0
fakeredis==2.40.0
pytest==9.1.1
pipeline.py
import fakeredis
import redis


def make_server():
    # a single shared FakeServer is what makes WATCH visible across clients
    server = fakeredis.FakeServer()
    return server, fakeredis.FakeRedis(server=server, decode_responses=True), fakeredis.FakeRedis(server=server, decode_responses=True)


def transfer(r, src, dst, amount, retries=3):
    with r.pipeline() as pipe:
        for _ in range(retries):
            try:
                pipe.watch(src, dst)
                if int(pipe.get(src) or 0) < amount:
                    pipe.unwatch()
                    return False
                pipe.multi()
                pipe.decrby(src, amount)
                pipe.incrby(dst, amount)
                pipe.execute()
                return True
            except redis.WatchError:
                continue
    raise RuntimeError("too much contention")
tests/test_pipeline.py
import fakeredis
import pytest
import redis

from pipeline import make_server, transfer


def test_pipeline_buffers_then_executes_in_order():
    _, r, _ = make_server()
    with r.pipeline() as pipe:
        pipe.set("n", 1)
        pipe.incr("n")
        pipe.get("n")
        assert pipe.execute() == [True, 2, "2"]


def test_watch_turns_reads_into_immediate_calls():
    _, r, _ = make_server()
    r.set("k", "v")
    with r.pipeline() as pipe:
        pipe.watch("k")
        # while watching, commands run now instead of being queued
        assert pipe.get("k") == "v"
        assert pipe.command_stack == []
        pipe.multi()
        pipe.get("k")
        assert len(pipe.command_stack) == 1
        assert pipe.execute() == ["v"]


def test_watch_error_on_concurrent_write():
    _, r1, r2 = make_server()
    r1.set("balance", 10)
    with r1.pipeline() as pipe:
        pipe.watch("balance")
        r2.set("balance", 999)
        pipe.multi()
        pipe.set("balance", 0)
        with pytest.raises(redis.WatchError):
            pipe.execute()
    assert r1.get("balance") == "999"


def test_transfer_moves_funds_atomically():
    _, r, _ = make_server()
    r.mset({"a": 10, "b": 0})
    assert transfer(r, "a", "b", 4) is True
    assert r.mget("a", "b") == ["6", "4"]


def test_transfer_aborts_on_insufficient_funds():
    _, r, _ = make_server()
    r.set("a", 1)
    assert transfer(r, "a", "b", 5) is False
    assert r.get("a") == "1"


def test_default_instances_do_not_share_state():
    a = fakeredis.FakeRedis(decode_responses=True)
    b = fakeredis.FakeRedis(decode_responses=True)
    a.set("k", "v")
    assert b.get("k") is None
pytest.ini
[pytest]
pythonpath = .

Connect to ProvenPin

Add the free MCP server so your agent can pull this recipe and others like it. Same setup as the docs.

terminal
claude mcp add --transport http provenpin https://mcp.provenpin.com/mcp

Any other MCP client: Streamable HTTP at https://mcp.provenpin.com/mcp. No API key, no signup.

Related stacks

Celery 5 on Redis with fakeredis-backed idempotency, tested offline in eager mode

Configure a Celery app on a Redis broker and result backend for production, guard a task with a Redis SET NX idempotency lock, and test it offline with task_always_eager and fakeredis.

Celery · redis · fakeredis

RQ background jobs tested with fakeredis

Enqueue Redis Queue (RQ) jobs and process them with a SimpleWorker against fakeredis, so queue tests run offline with no Redis server.

RQ · fakeredis

All redis recipesAll fakeredis recipes