Redis OM Python

If your FastAPI application is built using asyncio, you should use the aredis_om module instead of redis_om.

While the modules are nearly identical, aredis_om returns coroutines for its methods, meaning you must use the await keyword when calling them.

Key differences:

• Import: Use from aredis_om import HashModel, ...
• Execution: Use await model.save(), await model.get(pk), etc.

This is essential for non-blocking I/O in high-concurrency FastAPI applications.

from aredis_om import HashModel, NotFoundError, get_redis_connection

class Customer(HashModel):
    first_name: str
    # ... other fields

@app.post("/customer")
async def save_customer(customer: Customer):
    # Use await with aredis_om
    return await customer.save()

@app.get("/customer/{pk}")
async def get_customer(pk: str):
    try:
        return await Customer.get(pk)
    except NotFoundError:
        raise HTTPException(status_code=404, detail="Customer not found")

In Redis OM 1.0, datetime fields are indexed as NUMERIC (Unix timestamps) instead of TAG (ISO strings). This change enables powerful new capabilities that were previously unavailable or inefficient:

• Range Queries: Use comparison operators (e.g., >, <, >=, <=) on datetime fields.
• Sorting: Use .sort_by() on datetime fields.
• Between Queries: Perform queries within a specific time range.

Data Format Change:

• Before: Stored as ISO strings (e.g., "2023-12-01T14:30:22.123456").
• After: Stored as Unix timestamps (e.g., 1701435022).

Note: When migrating existing data, you must run the schema migration and the data migration commands to convert existing ISO strings to timestamps.

# Range queries and sorting now work natively
users = await User.find(User.created_at > datetime.now() - timedelta(days=7)).sort_by('created_at').all()

# Between queries
start = datetime(2023, 1, 1)
end = datetime(2023, 12, 31)
users = await User.find(
    (User.created_at >= start) & (User.created_at <= end)
).all()

When indexing List or Tuple fields in a JsonModel, there are two critical constraints:

1. Elements must be strings: You cannot index a list of integers or other types. If you need to index non-string elements, you must use a different approach or store them without indexing (Error E12).
2. No Full-Text Search: List and Tuple fields are indexed as TAG fields. They support exact matching but do not support the full_text_search=True option (Error E13).

Error E12: List and tuple fields can only contain strings. Error E13: List and tuple fields cannot be indexed for full-text search.

from typing import List
from redis_om import JsonModel, Field

# This works - list of strings
class Article(JsonModel):
    tags: List[str] = Field(index=True)

# This does NOT work - list of integers (Raises E12)
class Article(JsonModel):
    scores: List[int] = Field(index=True)

# This does NOT work - list with full-text search (Raises E13)
class Article(JsonModel):
    tags: List[str] = Field(index=True, full_text_search=True)

You can create abstract models to gather shared configuration (like a global key prefix) for your application's models. Abstract models must subclass both ABC (from abc) and either HashModel or JsonModel. You cannot instantiate abstract models directly.

When a subclass is created from a base class with a Meta object, Redis OM copies the parent's fields into the child's Meta object. A subclass can override specific fields in its Meta class without redefining the entire object.

from abc import ABC
from redis_om import HashModel, get_redis_connection


redis = get_redis_connection(port=6380)
other_redis = get_redis_connection(port=6381)


class BaseModel(HashModel, ABC):
    class Meta:
        global_key_prefix = "customer-dashboard"
        database = redis


class Customer(BaseModel):
    first_name: str
    last_name: str

    class Meta:
        database = other_redis


print(Customer.global_key_prefix)
# > "customer-dashboard"  # Inherited from BaseModel

Every Redis OM model is also a Pydantic model. You can use standard Pydantic type annotations and validators (like EmailStr, Pattern, etc.) to ensure data integrity. If you attempt to instantiate a model with invalid data, a pydantic.ValidationError will be raised.

By default, Redis OM will attempt to coerce input values to the correct type (e.g., parsing a valid ISO date string into a datetime.date object).

import datetime
from typing import Optional
from redis_om import HashModel
from pydantic import ValidationError

class Customer(HashModel):
    first_name: str
    last_name: str
    email: str
    join_date: datetime.date
    age: int
    bio: Optional[str] = "Super dope"

try:
    Customer(
        first_name="Andrew",
        last_name="Brookins",
        email="a@example.com",
        join_date="not a date!",  # This will trigger ValidationError
        age=38
    )
except ValidationError as e:
    print(e)