Redis

Expanse supports natively and provides a convenient interface for interacting with Redis databases.

Before you can use Redis in your application, you need to do two things:

1. Install the redis package:

   pippoetryuv

   pip install redis
   

   poetry add redis
   

   uv add redis
   

   Alternatively, you can install Expanse with the redis extra, which includes the redis package:

   pippoetryuv

   pip install expanse[redis]
   

   poetry add expanse[redis]
   

   uv add expanse[redis]
   

2. Add the RedisServiceProvider to the list of service providers in your application configuration (app/app.py file):

   from expanse.support.service_providers_list import ServiceProvidersList
   
   providers = ServiceProvidersList.default().merge(
       [
           # Package-provided providers
           "expanse.redis.redis_service_provider.RedisServiceProvider"
       ]
   ).merge(
       [
           # Application-specific providers
       ]
   )
   

Configuration

The configuration of your database is done through environment variables prefixed with REDIS_. Default configurations are provided in the example .env.example file at the root of you application directory. You can edit and/or remove any value as you see fit.

The default connection that will be used by Expanse is controlled via the REDIS_CONNECTION environment variable.

REDIS_CONNECTION=default

You can configure as many connections as you want by following the following pattern:

REDIS_CONNECTIONS__CONNECTION_NAME__URL=redis://127.0.0.1:6379/0

The minimum required configuration for a connection is the URL, which should be a valid Redis URL. The following URL schemes are supported:

• redis://: creates a TCP socket connection.
• rediss://: creates a SSL wrapped TCP socket connection.

The database to connect to can be specified in the URL path (redis://127.0.0.1:6379/0) or the query string (redis://127.0.0.1:6379?db=0).

Using Redis

To use Redis in your application, you can inject a Connection instance. The Connection instance provides a convenient interface for interacting with your Redis database.

from expanse.redis.asynchronous.connections.connection import Connection


async def my_route(redis: Connection) -> Response:
    await redis.set("key", "value")
    value = await redis.get("key")

from expanse.redis.synchronous.connections.connection import Connection


def my_route(redis: Connection) -> Response:
    redis.set("key", "value")
    value = redis.get("key")

Using multiple connections

If you have configured multiple connections, you can specify which connection to use by injecting an annotated Connection instance:

from typing import Annotated

from expanse.redis.asynchronous.connections.connection import Connection

async def my_route(redis: Annotated[Connection, "cache"]) -> Response:
    value = await redis.get("key")

from typing import Annotated

from expanse.redis.synchronous.connections.connection import Connection


def my_route(redis: Annotated[Connection, "cache"]) -> Response:
    value = redis.get("key")