API Reference¶

This page provides a comprehensive reference of all available functions and methods in the Rapyer package.

AtomicRedisModel¶

The AtomicRedisModel is the core class that all Redis-backed models inherit from. It provides atomic operations and seamless integration with Redis as a backend storage.

Properties¶

`pk`¶

Type: str
Description: Primary key of the model. If a key field is defined using KeyAnnotation, returns that field's value; otherwise returns the internal UUID.

user = User(name="John", age=30)
print(user.pk)  # Returns UUID if no key field defined

`key`¶

Type: RapyerKey Description: The Redis key used to store this model instance. Format: {ClassName}:{pk}. Returns a RapyerKey, a str subclass that identifies the value as a Rapyer-managed key.

user = User(name="John")
print(user.key)  # "User:abc-123-def-456"

`field_name`¶

Type: str
Description: The field name when this model is used as a nested field in another model.

`field_path`¶

Type: str
Description: The full JSON path to this model when nested, including parent paths.

`json_path`¶

Type: str
Description: The JSON path used for Redis JSON operations. Returns "$" for root level or "${field_path}" for nested models.

Instance Methods¶

`asave()`¶

Type: async method
Returns: Self
Description: Saves the current model instance to Redis using JSON.SET operation.

user = User(name="John", age=30)
await user.asave()

`aload()`¶

Type: async method Returns: Self Raises: CantSerializeRedisValueError if the value in Redis cannot be deserialized (Corruption or missing resources) Description: Loads the latest data from Redis for this model instance, updating the current instance.

await user.aload()  # Refreshes user with latest Redis data

`adelete()`¶

Type: async method
Returns: bool
Description: Deletes this model instance from Redis. Can only be called on top-level models (not nested ones). Returns True only if at least one Redis key was actually removed; returns False if the model did not exist.

Warning: When called inside an active pipeline context (e.g. inside apipeline()), the deletion is queued on the outer caller's pipeline and executed later. In that case the actual Redis result is not observable here, so adelete() returns True unconditionally — it does not reflect whether a key was truly removed. Only the standalone (non-pipeline) call is guaranteed to return False for a missing key.

success = await user.adelete()

`aset_ttl(ttl)`¶

Type: async method Parameters: - ttl (int): Time-to-live in seconds Description: Sets the TTL for this model instance in Redis. Can only be called on top-level models. Supports pipeline context for atomic TTL updates.

# Direct usage
await user.aset_ttl(3600)  # Expire in 1 hour

# Within pipeline context
async with user.apipeline():
    user.score += 100
    await user.aset_ttl(7200)
    await user2.aset_ttl(7200)  # TTL set atomically with other changes as part of the transaction

`aduplicate()`¶

Description: Creates a duplicate of the current model with a new primary key and saves it to Redis.

user_copy = await user.aduplicate()

`aduplicate_many(num)`¶

Type: async method
Parameters: - num (int): Number of duplicates to create
Returns: list[Self]
Description: Creates multiple duplicates of the current model.

user_copies = await user.aduplicate_many(5)  # Creates 5 duplicates

`update(**kwargs)`¶

Type: Synchronous method
Parameters: - **kwargs: Field values to update
Description: Updates model fields locally (does not save to Redis).

user.update(name="Jane", age=25)

`aupdate(**kwargs)`¶

Type: async method
Parameters: - **kwargs: Field values to update
Description: Updates specified fields both locally and in Redis atomically.

await user.aupdate(name="Jane", age=25)

`is_inner_model()`¶

Type: Synchronous method
Returns: bool
Description: Returns True if this model is a nested field within another model.

if user.is_inner_model():
    print("This is a nested model")

Class Methods¶

`aget(key)`¶

Type: async class method Parameters: - key (str): The Redis key to retrieve Returns: Self Raises:

KeyNotFound if key doesn't exist
CantSerializeRedisValueError if the value in Redis cannot be deserialized (Corruption or missing resources)

Description: Retrieves a model instance from Redis by its key.

user = await User.aget("User:abc-123")

`adelete_by_key(key)`¶

Type: async class method Parameters: - key (str): The Redis key to delete Returns: bool Description: Deletes a model from Redis by its key without needing to load it first. Returns True only if at least one Redis key was actually removed; returns False if no key matched.

Warning: When called inside an active pipeline context (e.g. inside apipeline()), the deletion is queued on the outer caller's pipeline and executed later. In that case the actual Redis result is not observable here, so adelete_by_key() returns True unconditionally — it does not reflect whether a key was truly removed. Only the standalone (non-pipeline) call is guaranteed to return False for a missing key.

success = await User.adelete_by_key("User:abc-123")

`aexists(key)`¶

Type: async class method Parameters: - key (str | AtomicRedisModel): The Redis key or a model instance to check. When a model instance is passed, its .key is extracted automatically. Returns: bool Description: Checks whether a key exists in Redis. If the model uses a Key[] annotation and the provided key has no : separator, the class key prefix is automatically prepended.

# Check with full key
exists = await User.aexists("User:abc-123")

# With Key[] annotation, just the primary key value works too
exists = await User.aexists("abc-123")

if not exists:
    print("User not found")

`afind(*args, max_results=None)`¶

Type: async class method Parameters: - *args (str | Expression): Keys (strings) or filter expressions - max_results (int | None): Maximum number of models to return. None returns all matches. Returns: list of redis models Raises: CantSerializeRedisValueError if a value in Redis cannot be deserialized (Corruption or missing resources) Description: Retrieves model instances from Redis. Supports three modes:

No arguments - Returns all instances
Keys - Pass string keys to retrieve specific models
Expressions - Pass filter expressions (requires Index annotation)

Note

Keys and expressions are mutually exclusive. If both are provided, keys take priority and expressions are ignored (a warning is logged).

Supported Filter Operators: - == - Equal to - != - Not equal to - > - Greater than - < - Less than - >= - Greater than or equal to - <= - Less than or equal to - & - Logical AND - | - Logical OR - ~ - Logical NOT

# Get all users
all_users = await User.afind()

# Find by keys (full key or just primary key)
users = await User.afind("User:abc123", "User:def456")
users = await User.afind("abc123", "def456")  # prefix added automatically

# Filter with expressions (requires Index annotation on fields)
active_users = await User.afind(User.status == "active")
adult_users = await User.afind(User.age >= 18)

# Complex filters
filtered = await User.afind(
    (User.age > 18) & (User.status == "active") | (User.role == "admin")
)

# Limit results
top_5 = await User.afind(max_results=5)
recent_active = await User.afind(User.status == "active", max_results=10)

`afind_one(*args)`¶

Type: async class method Parameters: - *args (str | Expression): Keys (strings) or filter expressions (same as afind()) Returns: Self | None - Returns a single model instance or None if no match is found. Description: Retrieves a single model instance. Returns None when no match is found. Equivalent to calling afind(*args, max_results=1) and returning the first result.

user = await User.afind_one(User.name == "Alice")
if user is not None:
    print(user.name)

`afind_keys(max_results=None)`¶

Type: async class method Parameters: - max_results (int | None): Maximum number of keys to return. None returns all keys. Returns: list[RapyerKey] Description: Retrieves Redis keys for instances of this model class.

user_keys = await User.afind_keys()  # Returns ["User:123", "User:456", ...]
first_10 = await User.afind_keys(max_results=10)

`ainsert(*models)`¶

Type: async class method
Parameters: - *models (Self): Variable number of model instances to insert
Returns: None
Description: Inserts multiple model instances to Redis in a single atomic transaction. This is significantly more efficient than calling save() on each model individually, as it uses Redis pipelining to batch all operations.

Benefits: - Transactional: All models are saved atomically - either all succeed or all fail - Performance: Uses a single Redis pipeline instead of multiple round trips - Network Efficiency: Reduces network latency by batching operations - Consistency: Prevents partial saves if an error occurs during bulk operations

# Create multiple user instances
users = [
    User(name="Alice", email="alice@example.com"),
    User(name="Bob", email="bob@example.com"),
    User(name="Charlie", email="charlie@example.com")
]

# Insert all users atomically
await User.ainsert(*users)

# Alternative syntax
await User.ainsert(user1, user2, user3)

`adelete_many(*args)`¶

Type: async class method Parameters: - *args (Self | RapyerKey | str | Expression): Model instances, keys obtained from .key/.pk, or filter expressions Returns: DeleteResult - contains count (number of deleted keys) and was_committed (bool) Raises: UnsupportArgumentTypeError if no arguments provided, if expressions are mixed with keys/instances, or if an unsupported type is passed

Deletes multiple model instances from Redis. Supports three input modes: model instances, RapyerKey strings (from .key or .pk), or filter expressions (requires Index fields). Large deletes are automatically batched based on max_delete_per_transaction.

# Delete using model instances
result = await User.adelete_many(*users)
print(result.models_deleted)  # 4

# Delete using keys from .key or .pk or a string
result = await User.adelete_many(users[0].key, users[1].pk, "User:123")

# Mix models and keys
result = await User.adelete_many(user1, user2.key, user3.pk)

# Delete with filter expressions (requires indexed fields)
result = await User.adelete_many(User.status == "inactive")
result = await User.adelete_many((User.age < 18) & (User.status == "pending"))

`class_key_initials()`¶

Type: Class method
Returns: str
Description: Returns the class name used as the key prefix. Override to customize key naming.

class User(AtomicRedisModel):
    @classmethod
    def class_key_initials(cls):
        return "USR"  # Keys will be "USR:pk" instead of "User:pk"

Context Managers¶

`alock_from_key(key, action="default", save_at_end=False)`¶

Type: async context manager
Parameters: - key (str): Redis key to lock - action (str): Lock action name for different operation types - save_at_end (bool): Whether to save changes when context exits
Returns: AsyncGenerator[Self, None]
Description: Acquires an exclusive lock on the model and yields the loaded instance.

async with User.alock_from_key("User:123", "profile_update", save_at_end=True) as user:
    user.name = "Updated Name"
    # Automatically saved when context exits

`alock(action="default", save_at_end=False)`¶

Type: async context manager
Parameters: - action (str): Lock action name - save_at_end (bool): Whether to save changes when context exits
Returns: AsyncGenerator[Self, None]
Description: Acquires an exclusive lock on the current model instance.

async with user.alock("settings_update", save_at_end=True) as locked_user:
    locked_user.settings = {"theme": "dark"}

`apipeline(ignore_redis_error=False, use_existing_pipe=False)`¶

Type: async context manager Parameters: - ignore_redis_error (bool): If True, suppresses redis.exceptions.ResponseError raised during pipeline execution (for example, if the model was deleted during the pipeline) and continues. - use_existing_pipe (bool): If True, reuses an already active pipeline from an outer context instead of creating a new one. Changes are deferred until the outer pipeline exits. Returns: AsyncGenerator[Self, None] Description: Batches all operations into a Redis pipeline for atomic execution.

async with user.apipeline() as pipeline_user:
    user.score += 100
    user.achievements.append("New Achievement")
    # All operations executed atomically

Configuration¶

`Meta`¶

Type: ClassVar[RedisConfig]
Description: Configuration class for Redis connection settings and model behavior.

class User(AtomicRedisModel):
    name: str
    age: int

    Meta = RedisConfig(redis=redis_client, ttl=3600)  # Expire after 1 hour

Available options:

redis - Redis client instance
ttl - Time-to-live in seconds
refresh_ttl - Controls TTL refresh behavior. Accepts True (refresh on every operation, default), False (never refresh), or an ActionGroup flag for fine-grained per-category control.
safe_load_all - Treat all non-Redis fields as SafeLoad (default: False)
prefer_normal_json_dump - Use JSON serialization for compatible fields instead of pickle (default: False)
max_delete_per_transaction - Maximum keys deleted per pipeline transaction in adelete_many() (default: 1000, set to None to disable batching)

Special Behaviors¶

Key Field Definition¶

Use KeyAnnotation to specify which field should be used as the primary key:

from rapyer.fields import KeyAnnotation
from typing import Annotated

class User(AtomicRedisModel):
    username: Annotated[str, KeyAnnotation] 
    email: str

user = User(username="john_doe", email="john@example.com")
print(user.pk)   # "john_doe"
print(user.key)  # "User:john_doe"

Automatic Redis Type Conversion¶

Model fields are automatically converted to Redis-compatible types:

class User(AtomicRedisModel):
    name: str          # Becomes RedisStr
    tags: List[str]    # Becomes RedisList[str]
    settings: Dict[str, str]  # Becomes RedisDict[str, str]

Field Linking¶

Redis type fields are automatically linked to their parent model for proper key resolution:

user = User(name="John", tags=["python", "redis"])
print(user.tags.key)  # "User:abc-123.tags"

Error Handling¶

`KeyNotFound`¶

Raised when attempting to get or load a model that doesn't exist in Redis:

from rapyer.errors import KeyNotFound

try:
    user = await User.aget("User:nonexistent")
except KeyNotFound:
    print("User not found in Redis")

Example Usage¶

from rapyer import AtomicRedisModel
from typing import List, Dict


class User(AtomicRedisModel):
    name: str
    email: str
    age: int = 0
    tags: List[str] = []
    settings: Dict[str, str] = {}


# Create and save
user = User(name="John", email="john@example.com", age=30)
await user.asave()

# Retrieve
retrieved_user = await User.aget(user.key)

# Update atomically
await user.aupdate(age=31, tags=["python", "redis"])

# Batch operations
async with user.apipeline():
    user.age += 1
    user.tags.append("asyncio")
    user.settings["theme"] = "dark"

# Lock for exclusive access
async with user.alock("profile_update", save_at_end=True):
    if user.age >= 25:
        user.tags.append("adult")
        user.settings["account_type"] = "premium"

API Reference¶

AtomicRedisModel¶

Properties¶

pk¶

key¶

field_name¶

field_path¶

json_path¶

Instance Methods¶

asave()¶

aload()¶

adelete()¶

aset_ttl(ttl)¶

aduplicate()¶

aduplicate_many(num)¶

update(**kwargs)¶

aupdate(**kwargs)¶

is_inner_model()¶