upstash-redis Python: sync and async clients, from_env, init outside the handler

TL;DR

Use pipeline() to batch independent commands into fewer round trips, or multi() when you need atomicity: pipe = redis.pipeline() pipe.set("foo", 1) pipe.incr("foo") result = pipe.exec() Commands chain too: redis.pipeline().set("a", 1).incr("a").exec().

Steps

1. pip install upstash-redis
2. from upstash_redis import Redis
3. redis = Redis.from_env()
4. fromenv reads UPSTASHREDISRESTURL and UPSTASHREDISREST_TOKEN from

your environment. For async code:

5. from upstash_redis.asyncio import Redis
6. redis = Redis.from_env()
7. If your platform allows it, create the client at module level so warm

function instances reuse it:

8. from upstash_redis import Redis
9. redis = Redis.from_env()

10. def handler(event, context): redis.incr("hits") return {"ok": True}

11. The REST proxy can base64-encode responses server side when values are

not valid JSON. For large payloads this adds a few milliseconds, so if you know your data is valid JSON you can set rest_encoding to None.

12. Use pipeline() to batch independent commands into fewer round trips, or

multi() when you need atomicity:

13. pipe = redis.pipeline()

pipe.set("foo", 1) pipe.incr("foo") result = pipe.exec()

14. Commands chain too: redis.pipeline().set("a", 1).incr("a").exec().

When to use

You are seeing this: For async code: from upstashredis.asyncio import Redis redis = Redis.fromenv() Use this skill when you run into "upstash-redis Python: sync and async clients, from_env, init outside the handler".

When not to use

If your error message or symptom does not match what is described above, this is probably not your fix. Search for your exact error text instead of forcing this one to fit.

Versions

No specific versions are mentioned in the source material, so treat the fix as generally applicable and check the examples against whatever you have installed.

Why this happens

The original report does not dig into a root cause. It documents the symptom and the fix that resolved it.

Verified against Upstash Python SDK docs: pip install upstash-redis; sync client from upstashredis import Redis, async from upstashredis.asyncio import Redis; Redis.fromenv() loads credentials; docs recommend initializing outside the request handler in serverless; default retry is one retry 3 seconds after error, customizable via restretries and restretryinterval. Sources: https://upstash.com/docs/redis/sdks/py/gettingstarted https://upstash.com/docs/redis/sdks/py/features