Redis

Redis is an in-memory data store, commonly used for caching, rate limiting, session storage, leaderboards and queues.

With Encore.py you declare a Redis database in code, and Encore provisions it: a local Redis emulator when you run your app with encore run, and a managed Redis when you deploy. Every command is traced automatically.

sessions/store.py
from encore.storage import redis

sessions = redis.Database("sessions")


async def save_session(session_id: str, user_id: str) -> None:
    await sessions.string.set(f"session:{session_id}", user_id, ex=3600)


async def load_session(session_id: str) -> str | None:
    return await sessions.string.get[str](f"session:{session_id}")

Creating a database

Create a database with redis.Database, assigned to a module-level variable:

from encore.storage import redis

sessions = redis.Database("sessions")

Encore reads the declaration from your source code to provision the database, so:

• Declare it as a module-level variable, not inside a function, class, or if block.
• The name must be a string literal. It's a kebab-case name such as sessions, unique within the app.

Database takes no other options yet: persistence and eviction policies can't be configured.

Import the variable from other modules to use the database there.

Running commands

Commands are grouped into families, which are properties of the database. Call a command on its family and await it:

await sessions.string.set("user:1", "Ada", ex=60)
name = await sessions.string.get("user:1")  # b"Ada"

await sessions.hash.hset("user:1:profile", ("name", "Ada"), ("lang", "en"))
await sessions.list.rpush("queue", "job-1", "job-2")
await sessions.sorted_set.zadd("leaderboard", (100, "ada"), (85, "grace"))
await sessions.generic.expire("queue", 300)

Methods are named after the Redis commands they run, and their options are keyword arguments, such as set(key, value, ex=60, nx=True). Redis errors raise encore.EncoreError.

Keys are plain strings. Encore namespaces them behind the scenes, for example to isolate tests from each other, but you always read and write the unprefixed names: generic.scan returns your keys as you wrote them.

Multiple values

Commands that take several values accept them as extra positional arguments, with a tuple for each grouped value such as a hash field and its value, or a sorted-set score and its member:

await sessions.string.mset(("first", "Ada"), ("last", "Lovelace"))
values = await sessions.string.mget("first", "last")

await sessions.set.sadd("tags", "python", "redis")

To pass an existing collection, unpack it with *, as in sessions.set.sadd("tags", *tags).

Values

You can store bytes, str, int and float values. Strings are encoded as UTF-8, integers as decimal text, floats using repr(), and bytes are stored unchanged. Booleans aren't supported as values.

Reads return bytes by default. To decode a stored value as another type, put the type in square brackets after the command:

await sessions.string.set("visits", 42)
visits = await sessions.string.get[int]("visits")  # int | None

await sessions.list.rpush("names", "Ada", "Grace")
names = await sessions.list.lrange[str]("names", 0, -1)  # list[str]

top = await sessions.sorted_set.zrange[str]("leaderboard", 0, 9, withscores=True)
# list[tuple[str, float]]

The type applies to stored values: list elements, hash values, set members, and so on. Keys, hash field names, scores and other metadata keep their own types; for example, hash.hgetall[int] returns dict[bytes, int].

A missing value is None, and an empty collection is empty. A value that can't be decoded as the requested type (invalid UTF-8, or text that isn't a number) raises instead of returning None. Commands that change data and also return it, such as getdel or lpop, run before the result is decoded, so a decoding error doesn't undo them.

Custom value types

To store your own types, give them an encode method that returns bytes and a decode class method that builds an instance from bytes. This is the redis.RedisValue protocol; you don't need to inherit from anything:

import json
from dataclasses import dataclass


@dataclass
class User:
    name: str

    def encode(self) -> bytes:
        return json.dumps({"name": self.name}).encode()

    @classmethod
    def decode(cls, data: bytes) -> "User":
        return cls(**json.loads(data))


await sessions.string.set("user:1", User("Ada"))
user = await sessions.string.get[User]("user:1")  # User | None

Redis doesn't record which type a value was stored as, so use the same type when you read it back.

Read and write handles

Encore reads which Redis commands each service uses from your code, to know which services read and which write each database. That's why a command must be called directly on its family, as in sessions.string.get(...): Encore can see both the family and the command at that call. If you store a family in a variable or pass it to a function, Encore can't tell which commands run through it, and reports an error.

To hand access to a database to other code, such as a helper function or a FastAPI dependency, take a handle first:

• sessions.read() returns a redis.ReadOnlyDatabaseRef, which has only the read commands of each family.
• sessions.write() returns a redis.DatabaseRef, which has every command.

Handles can be passed around and stored freely; Encore records the access at the read() or write() call.

from typing import Annotated

from encore.storage import redis
from fastapi import Depends, FastAPI

from sessions.store import sessions

app = FastAPI()


async def cache_reader() -> redis.ReadOnlyDatabaseRef:
    return sessions.read()


async def count_visits(db: redis.DatabaseRef, page: str) -> int:
    return await db.string.incr(f"visits:{page}")


@app.get("/pages/{page}")
async def get_page(
    page: str, reader: Annotated[redis.ReadOnlyDatabaseRef, Depends(cache_reader)]
):
    cached = await reader.string.get[str](f"page:{page}")
    visits = await count_visits(sessions.write(), page)
    return {"cached": cached, "visits": visits}

A read handle rejects writes at run time too, not only in your type checker.

Redis channels

The pubsub family is Redis's own publish/subscribe: fire-and-forget messages to whoever is subscribed to a channel at that moment. It's separate from Encore's Pub/Sub topics, which store messages durably and retry failed deliveries.

Publish with publish, which returns how many subscribers received the message:

await sessions.pubsub.publish("chat:lobby", "hello")

subscribe opens a dedicated connection and returns a subscription you iterate asynchronously. Use it as an async context manager so the connection is closed when you're done, even if your code raises or is cancelled:

async def listen() -> None:
    async with await sessions.pubsub.subscribe("chat:lobby") as messages:
        async for message in messages:
            print(message.channel, message.payload)  # both bytes

psubscribe subscribes to channels matching glob patterns, such as chat:*. Both take several channels or patterns, and a type for the payloads, as in subscribe[str]("chat:lobby"). Each message has a channel (always bytes) and a decoded payload. You can also call receive(), which returns the next message or None once the subscription is closed, and close().

JSON

The json family works with RedisJSON documents. Values are Python JSON values, encoded and decoded for you:

await sessions.json.set("profile:1", "$", {"name": "Ada", "active": True})
names = await sessions.json.get("profile:1", "$.name")  # ["Ada"]
await sessions.json.numincrby("profile:1", "$.logins", 1)

A JSONPath (starting with $) can select several values, so replies to one are lists of matches.

Connecting other clients

connection_string returns the database's Redis URL, including credentials, for use with another Redis client:

url = sessions.connection_string

It's available on the database and on write handles, not on read handles, and grants full read-write access. Clients using it bypass Encore: they don't apply Encore's key namespacing, so they may not see the same keys as Encore's commands (under encore test, for example), and their commands aren't traced.

Not supported yet

The Python SDK doesn't yet have:

• Persistence or eviction options on Database.
• Transactions (MULTI/EXEC), Lua scripts, or pipelining.
• Read replica routing (see read(primary=True) above).
• A way to reference a Redis database by name, as Bucket.named and Topic.named do. Import the declaring module's variable instead.

Local development and testing

encore run runs a local Redis emulator for your databases, so you don't need to install Redis. Under encore test, each test gets its own namespace of keys and channels on a shared local server, so tests don't see each other's data. See Testing.