Upstash Redis Python SDK

Warnings

• breaking In version 1.0.0, the `set` and `hset` commands changed their `value` type from `Any` to `ValueT` (Union[str, int, float, bool]). Directly passing dictionaries or other complex objects for JSON storage will now result in a type error. You must explicitly `json.dumps()` such values before setting them. [2]
  Fix: Explicitly serialize non-scalar values (e.g., dictionaries) to JSON strings using `json.dumps()` before passing them to `set()` or `hset()`.
• breaking With version 1.0.0, the return types for set-related commands like `sdiff`, `sunion`, `sinter`, and `smembers` were changed from `Set` to `List`. This was done to avoid unnecessary set allocations. [2]
  Fix: If your code relies on the previous `Set` return type, you will need to explicitly convert the `List` result to a `set` (e.g., `set(redis.smembers('my_set'))`).
• gotcha The SDK collects anonymous telemetry data by default (e.g., SDK version, platform, Python runtime version). This feature was introduced around v1.5.0. [1, 3, 5, 7, 9, 11]
  Fix: To opt-out, set `allow_telemetry=False` when initializing the `Redis` client (e.g., `Redis.from_env(allow_telemetry=False)`) or set the environment variable `UPSTASH_DISABLE_TELEMETRY=true`. [5, 7, 11]
• gotcha The 'Read Your Writes' consistency feature, which ensures that write operations are completed before subsequent reads, is automatically managed by the SDK for versions 1.2.0 and later. If you are using an older SDK version or interacting directly with the REST API, you need to manually manage the `upstash-sync-token` header for strong consistency. [13]
  Fix: Upgrade to SDK v1.2.0+ for automatic handling. If using older versions or the REST API directly, capture the `upstash-sync-token` from a write response and include it in subsequent read requests.
• gotcha By default, the Upstash REST proxy may base64 encode/decode data, especially when dealing with JSON. While this ensures data integrity, it can introduce slight latency, particularly for very large payloads. [1, 5]
  Fix: If you are certain your data is valid JSON and you want to avoid the encoding overhead, you can set `rest_encoding=None` or `rest_encoding=False` when initializing the client (e.g., `Redis.from_env(rest_encoding=None)`). [5]

Install

• pip install upstash-redis Install stable version

Imports

• Redis
  wrong:

  from upstash_redis.client import Redis

  correct:

  from upstash_redis import Redis

  The direct import `from upstash_redis import Redis` is the current and recommended path for the synchronous client. The `from upstash_redis.client import Redis` path was used in earlier versions but is generally not recommended anymore. [4, 9]
• Redis (async)

  from upstash_redis.asyncio import Redis

  This is the correct import path for the asynchronous Redis client.

Quickstart

This quickstart demonstrates how to initialize both synchronous and asynchronous Upstash Redis clients using environment variables (`UPSTASH_REDIS_REST_URL` and `UPSTASH_REDIS_REST_TOKEN`). It performs a simple `SET` and `GET` operation. In serverless environments, it is recommended to initialize the client outside the request handler to maximize reuse and efficiency. [1, 3, 8, 9]

import os
from upstash_redis import Redis

# Ensure environment variables are set for demonstration
# In a real application, set UPSTASH_REDIS_REST_URL and UPSTASH_REDIS_REST_TOKEN
# in your environment or pass them directly to Redis().
os.environ['UPSTASH_REDIS_REST_URL'] = os.environ.get('UPSTASH_REDIS_REST_URL', 'YOUR_UPSTASH_REDIS_REST_URL')
os.environ['UPSTASH_REDIS_REST_TOKEN'] = os.environ.get('UPSTASH_REDIS_REST_TOKEN', 'YOUR_UPSTASH_REDIS_REST_TOKEN')

def run_sync_example():
    try:
        redis_sync = Redis.from_env()
        redis_sync.set('mykey_sync', 'myvalue_sync')
        value = redis_sync.get('mykey_sync')
        print(f"Sync: Set 'mykey_sync' to 'myvalue_sync', retrieved: {value}")
        redis_sync.delete('mykey_sync')
    except Exception as e:
        print(f"Sync example failed: {e}")

if __name__ == '__main__':
    print("Running synchronous example...")
    run_sync_example()

    # Async example (requires asyncio and aiohttp)
    import asyncio
    from upstash_redis.asyncio import Redis

    async def run_async_example():
        try:
            redis_async = Redis.from_env()
            await redis_async.set('mykey_async', 'myvalue_async')
            value = await redis_async.get('mykey_async')
            print(f"Async: Set 'mykey_async' to 'myvalue_async', retrieved: {value}")
            await redis_async.delete('mykey_async')
        except Exception as e:
            print(f"Async example failed: {e}")

    print("\nRunning asynchronous example...")
    asyncio.run(run_async_example())