coredis

import anyio import coredis import os async def main() -> None: # Connect to Redis. Use a URL from an environment variable or default to localhost redis_url = os.environ.get('COREDIS_URL', 'redis://localhost:6379/0') # Optionally, decode responses to get Python strings instead of bytes client = coredis.Redis.from_url(redis_url, decode_responses=True) async with client: # Clear the database (use with caution in production!) print(f"Flushing database...") await client.flushdb() # Basic SET and GET operations print(f"Setting 'mykey' to 'hello'") await client.set("mykey", "hello") value = await client.get("mykey") print(f"Value of 'mykey': {value}") assert value == "hello" # Increment a numerical value print(f"Incrementing 'counter'") await client.set("counter", 1) assert await client.incr("counter") == 2 print(f"Value of 'counter' after increment: {await client.get('counter')}") # Using a pipeline for multiple commands in a single round trip print("Running a pipeline...") async with client.pipeline() as pipeline: pipeline.incr("pipeline_counter") pipeline.get("pipeline_counter") pipeline.delete(["pipeline_counter"]) results = await pipeline.execute() print(f"Pipeline results: {results}") # Expected: [1, '1', 1] (if decode_responses=True) if __name__ == "__main__": # coredis uses anyio, supporting asyncio and trio. Specify your preferred backend. anyio.run(main, backend="asyncio")