Function start_temporary_redis_server

examples/run_examples.py:304–341

()

def start_temporary_redis_server() -> TemporaryRedisServer | None:
    redis_server = shutil.which("redis-server")
    if redis_server is None:
        return None

    port = choose_loopback_port()
    temp_dir = tempfile.TemporaryDirectory(prefix="examples-redis-")
    url = f"redis://127.0.0.1:{port}/0"
    process = subprocess.Popen(
        [
            redis_server,
            "--bind",
            "127.0.0.1",
            "--port",
            str(port),
            "--save",
            "",
            "--appendonly",
            "no",
            "--dir",
            temp_dir.name,
        ],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )

    server = TemporaryRedisServer(process=process, temp_dir=temp_dir, url=url)
    deadline = time.monotonic() + 5
    while time.monotonic() < deadline:
        if process.poll() is not None:
            server.close()
            return None
        if redis_ping_url(url, timeout=0.2):
            return server
        time.sleep(0.1)

    server.close()
    return None


def prepare_redis_for_example(