# RedisPydantic
A Python package that provides Pydantic models with Redis as the backend storage, enabling automatic synchronization between your Python objects and Redis with full type validation.
## Features
- **Async/Await Support**: Built with asyncio for high-performance applications
- **Pydantic Integration**: Full type validation and serialization using Pydantic v2
- **Redis Backend**: Efficient storage using Redis JSON with support for various data types
- **Automatic Serialization**: Handles lists, dicts, BaseModel instances, and primitive types
- **Atomic Operations**: Built-in methods for atomic list and dictionary operations
- **Type Safety**: Full type hints and validation for all Redis operations
## Installation
```bash
pip install redis-pydantic
```
## Requirements
- Python 3.10+
- Redis server with JSON module
- Pydantic v2
- redis-py with async support
## Quick Start
```python
import asyncio
from redis_pydantic.base import BaseRedisModel
from typing import List, Dict
class User(BaseRedisModel):
name: str
age: int
tags: List[str] = []
metadata: Dict[str, str] = {}
score: int = 0
async def main():
# Create a new user
user = User(name="John", age=30)
await user.save()
# Retrieve user by key
retrieved_user = User()
retrieved_user.pk = user.pk
await retrieved_user.name.load() # Load specific field
print(f"Retrieved: {retrieved_user.name}")
# Update field directly in Redis
await user.name.set("John Doe")
# Work with lists
await user.tags.aappend("python")
await user.tags.aappend("redis")
await user.tags.aextend(["pydantic", "async"])
# Work with dictionaries
await user.metadata.aset_item("department", "engineering")
await user.metadata.aupdate(role="developer", level="senior")
# Increment counters
await user.score.set(100)
if __name__ == "__main__":
asyncio.run(main())
```
## Redis Connection Setup
### Default Connection
By default, RedisPydantic connects to `redis://localhost:6379/0`.
### Custom Connection
Configure Redis connection in your model's `Meta` class:
```python
import redis.asyncio as redis
from redis_pydantic.base import BaseRedisModel
class MyModel(BaseRedisModel):
name: str
class Meta:
redis = redis.from_url("redis://your-redis-host:6379/1")
```
### Environment-based Configuration
```python
import os
import redis.asyncio as redis
from redis_pydantic.base import BaseRedisModel
redis_url = os.getenv("REDIS_URL", "redis://localhost:6379/0")
class MyModel(BaseRedisModel):
name: str
class Meta:
redis = redis.from_url(redis_url)
```
### Connection with Authentication
```python
import redis.asyncio as redis
from redis_pydantic.base import BaseRedisModel
class MyModel(BaseRedisModel):
name: str
class Meta:
redis = redis.from_url(
"redis://username:password@your-redis-host:6379/0",
decode_responses=True
)
```
## Supported Types and Operations
### String (RedisStr)
```python
class MyModel(BaseRedisModel):
name: str = "default"
# Operations
await model.name.set("new_value") # Set string value
await model.name.load() # Load from Redis
```
### Integer (RedisInt)
```python
class MyModel(BaseRedisModel):
counter: int = 0
# Operations
await model.counter.set(42) # Set integer value
await model.counter.load() # Load from Redis
```
### Boolean (RedisBool)
```python
class MyModel(BaseRedisModel):
is_active: bool = True
# Operations
await model.is_active.set(False) # Set boolean value
await model.is_active.load() # Load from Redis
```
### Bytes (RedisBytes)
```python
class MyModel(BaseRedisModel):
data: bytes = b""
# Operations
await model.data.set(b"binary_data") # Set bytes value
await model.data.load() # Load from Redis
```
### List (RedisList)
```python
class MyModel(BaseRedisModel):
items: List[str] = []
# Operations
await model.items.aappend("item1") # Append single item
await model.items.aextend(["item2", "item3"]) # Extend with multiple items
await model.items.ainsert(0, "first") # Insert at specific index
popped = await model.items.apop() # Pop last item
popped = await model.items.apop(0) # Pop item at index
await model.items.aclear() # Clear all items
await model.items.load() # Load from Redis
```
### Dictionary (RedisDict)
```python
class MyModel(BaseRedisModel):
metadata: Dict[str, str] = {}
# Operations
await model.metadata.aset_item("key", "value") # Set single item
await model.metadata.aupdate(key1="val1", key2="val2") # Update multiple items
await model.metadata.adel_item("key") # Delete item
popped = await model.metadata.apop("key") # Pop item by key
popped = await model.metadata.apop("key", "default") # Pop with default
key, value = await model.metadata.apopitem() # Pop arbitrary item
await model.metadata.aclear() # Clear all items
await model.metadata.load() # Load from Redis
```
## Advanced Usage
### Lock Context Manager
Use the lock context manager to ensure atomic operations across multiple fields with automatic model synchronization:
```python
class User(BaseRedisModel):
name: str
balance: int = 0
transaction_count: int = 0
user = User(name="John", balance=1000)
await user.save()
# Lock with default action
async with user.lock() as locked_user:
# The locked_user is automatically refreshed from Redis
locked_user.balance -= 50
locked_user.transaction_count += 1
# Changes are automatically saved when exiting the context
# Lock with specific action name
async with user.lock("transfer") as locked_user:
# This creates a lock with key "User:user_id/transfer"
locked_user.balance -= 100
locked_user.transaction_count += 1
# Automatic save on context exit
```
The lock context manager:
- Creates a Redis lock with key `{model_key}/{action}`
- Automatically refreshes the model with latest Redis data on entry
- Saves all changes back to Redis on successful exit
- Ensures atomic operations across multiple field updates
### Working with Nested Models
RedisPydantic automatically supports nested Pydantic models by converting regular `BaseModel` classes into Redis-enabled versions. This allows you to use all Redis field operations on nested model fields.
```python
from pydantic import BaseModel, Field
class UserProfile(BaseModel):
bio: str = ""
skills: List[str] = Field(default_factory=list)
settings: Dict[str, bool] = Field(default_factory=dict)
class Address(BaseModel):
street: str
city: str
country: str = "US"
class User(BaseRedisModel):
name: str
profile: UserProfile = Field(default_factory=UserProfile)
address: Address
tags: List[str] = Field(default_factory=list)
user = User(name="John", address=Address(street="123 Main St", city="Boston"))
await user.save()
# Access Redis operations on nested model fields
await user.profile.skills.aappend("Python")
await user.profile.skills.aextend(["Redis", "AsyncIO"])
await user.profile.settings.aupdate(dark_mode=True, notifications=False)
# Even deeply nested operations work
await user.profile.skills.ainsert(0, "Leadership")
popped_skill = await user.profile.skills.apop()
# All Redis list/dict operations are available on nested fields
await user.profile.settings.aset_item("email_updates", True)
await user.profile.settings.adel_item("notifications")
# Load specific nested fields
await user.profile.skills.load()
await user.profile.settings.load()
print(user.profile.skills) # Reflects Redis state
print(user.profile.settings) # Reflects Redis state
```
#### Deep Nesting Support
RedisPydantic supports unlimited nesting depth:
```python
class InnerModel(BaseModel):
items: List[str] = Field(default_factory=list)
counter: int = 0
class MiddleModel(BaseModel):
inner: InnerModel = Field(default_factory=InnerModel)
tags: List[str] = Field(default_factory=list)
class OuterModel(BaseRedisModel):
middle: MiddleModel = Field(default_factory=MiddleModel)
data: Dict[str, int] = Field(default_factory=dict)
outer = OuterModel()
await outer.save()
# All Redis operations work at any nesting level
await outer.middle.inner.items.aappend("deep_item")
await outer.middle.tags.aextend(["tag1", "tag2"])
await outer.data.aset_item("count", 42)
# Load nested data
await outer.middle.inner.items.load()
await outer.middle.tags.load()
```
#### Nested Model Persistence
Nested models maintain full persistence and consistency:
```python
# Create and modify nested data
user1 = User(name="Alice", address=Address(street="456 Oak Ave", city="Seattle"))
await user1.save()
await user1.profile.skills.aextend(["JavaScript", "TypeScript"])
# Access from different instance
user2 = User()
user2.pk = user1.pk
await user2.profile.skills.load()
print(user2.profile.skills) # ["JavaScript", "TypeScript"]
# All operations are atomic and persistent
await user2.profile.skills.aappend("React")
await user1.profile.skills.load() # user1 now sees the new skill
```
### Working with Nested Types
```python
class User(BaseRedisModel):
preferences: Dict[str, List[str]] = {}
scores: List[int] = []
user = User()
# Nested operations
await user.preferences.aset_item("languages", ["python", "rust"])
await user.scores.aextend([95, 87, 92])
```
### Loading Specific Fields
```python
user = User()
user.pk = "some-existing-id"
# Load only specific fields from Redis
await user.name.load()
await user.tags.load()
# Other fields remain unloaded
```
### Atomic Operations
All Redis operations are atomic. For example:
```python
# This is atomic - either both operations succeed or both fail
await user.metadata.aupdate(
last_login=str(datetime.now()),
session_count=str(session_count + 1)
)
# This is also atomic
popped_item = await user.items.apop() # Atomically removes and returns
```
### Model Serialization
```python
# Get the model as a Redis-compatible dict
redis_data = user.redis_dump()
# Save entire model to Redis
await user.save()
```
## Key Features
### Automatic Type Conversion
RedisPydantic handles serialization and deserialization automatically:
```python
class MyModel(BaseRedisModel):
data: bytes = b""
model = MyModel()
await model.data.set(b"binary_data") # Automatically base64 encoded in Redis
loaded_data = await model.data.load() # Automatically decoded back to bytes
```
### Type Safety
All operations maintain type safety:
```python
class MyModel(BaseRedisModel):
count: int = 0
# This will raise TypeError
await model.count.set("not_a_number") # ❌ TypeError: Value must be int
# This works
await model.count.set(42) # ✅ Valid
```
### Consistent Local and Redis State
RedisPydantic keeps your local Python objects in sync with Redis:
```python
user = User(tags=["python"])
await user.save()
await user.tags.aappend("redis") # Updates both local list and Redis
print(user.tags) # ["python", "redis"] - local state is updated
```
## Error Handling
```python
try:
# Attempt to pop from empty list
item = await user.tags.apop()
except IndexError:
print("List is empty")
try:
# Attempt to pop non-existent key from dict
value = await user.metadata.apop("nonexistent_key")
except KeyError:
print("Key not found")
# Using default values
value = await user.metadata.apop("key", "default_value") # Returns default if key missing
```
## Performance Tips
1. **Batch Operations**: Use `aupdate()` for multiple dict updates and `aextend()` for multiple list items
2. **Load Only What You Need**: Load specific fields instead of entire models when possible
3. **Use Appropriate Data Types**: Choose the right Redis type for your use case
4. **Connection Pooling**: Configure Redis connection pools for high-concurrency applications
## Examples
### User Session Management
```python
class UserSession(BaseRedisModel):
user_id: str
session_data: Dict[str, str] = {}
activity_log: List[str] = []
is_active: bool = True
last_seen: str = ""
session = UserSession(user_id="user123")
await session.save()
# Track user activity
await session.activity_log.aappend(f"login:{datetime.now()}")
await session.session_data.aupdate(
ip_address="192.168.1.1",
user_agent="Chrome/91.0"
)
```
### Shopping Cart
```python
class ShoppingCart(BaseRedisModel):
user_id: str
items: List[str] = [] # product IDs
quantities: Dict[str, int] = {}
total_amount: int = 0 # in cents
cart = ShoppingCart(user_id="user456")
# Add items
await cart.items.aappend("product123")
await cart.quantities.aset_item("product123", 2)
# Update totals
await cart.total_amount.set(4999) # $49.99
```
### Configuration Management
```python
class AppConfig(BaseRedisModel):
features: Dict[str, bool] = {}
limits: Dict[str, int] = {}
allowed_ips: List[str] = []
config = AppConfig()
# Enable feature flags
await config.features.aupdate(
new_ui=True,
beta_features=False
)
# Set rate limits
await config.limits.aupdate(
requests_per_minute=1000,
max_file_size=10485760
)
```
## License
MIT License
## Contributing
Contributions are welcome! Please feel free to submit a Pull Request.
Raw data
{
"_id": null,
"home_page": null,
"name": "redis-pydantic",
"maintainer": null,
"docs_url": null,
"requires_python": "<4.0,>=3.10",
"maintainer_email": null,
"keywords": "redis, pydantic, orm, database, async",
"author": "YedidyaHKfir",
"author_email": "your.email@example.com",
"download_url": "https://files.pythonhosted.org/packages/0f/36/87331f2f666aa452f6a1f80dfad2479808cefb5fb760418f04ac23c4738f/redis_pydantic-1.0.0.tar.gz",
"platform": null,
"description": "# RedisPydantic\n\nA Python package that provides Pydantic models with Redis as the backend storage, enabling automatic synchronization between your Python objects and Redis with full type validation.\n\n## Features\n\n- **Async/Await Support**: Built with asyncio for high-performance applications\n- **Pydantic Integration**: Full type validation and serialization using Pydantic v2\n- **Redis Backend**: Efficient storage using Redis JSON with support for various data types\n- **Automatic Serialization**: Handles lists, dicts, BaseModel instances, and primitive types\n- **Atomic Operations**: Built-in methods for atomic list and dictionary operations\n- **Type Safety**: Full type hints and validation for all Redis operations\n\n## Installation\n\n```bash\npip install redis-pydantic\n```\n\n## Requirements\n\n- Python 3.10+\n- Redis server with JSON module\n- Pydantic v2\n- redis-py with async support\n\n## Quick Start\n\n```python\nimport asyncio\nfrom redis_pydantic.base import BaseRedisModel\nfrom typing import List, Dict\n\nclass User(BaseRedisModel):\n name: str\n age: int\n tags: List[str] = []\n metadata: Dict[str, str] = {}\n score: int = 0\n\nasync def main():\n # Create a new user\n user = User(name=\"John\", age=30)\n await user.save()\n \n # Retrieve user by key\n retrieved_user = User()\n retrieved_user.pk = user.pk\n await retrieved_user.name.load() # Load specific field\n print(f\"Retrieved: {retrieved_user.name}\")\n \n # Update field directly in Redis\n await user.name.set(\"John Doe\")\n \n # Work with lists\n await user.tags.aappend(\"python\")\n await user.tags.aappend(\"redis\")\n await user.tags.aextend([\"pydantic\", \"async\"])\n \n # Work with dictionaries\n await user.metadata.aset_item(\"department\", \"engineering\")\n await user.metadata.aupdate(role=\"developer\", level=\"senior\")\n \n # Increment counters\n await user.score.set(100)\n\nif __name__ == \"__main__\":\n asyncio.run(main())\n```\n\n## Redis Connection Setup\n\n### Default Connection\n\nBy default, RedisPydantic connects to `redis://localhost:6379/0`. \n\n### Custom Connection\n\nConfigure Redis connection in your model's `Meta` class:\n\n```python\nimport redis.asyncio as redis\nfrom redis_pydantic.base import BaseRedisModel\n\nclass MyModel(BaseRedisModel):\n name: str\n \n class Meta:\n redis = redis.from_url(\"redis://your-redis-host:6379/1\")\n```\n\n### Environment-based Configuration\n\n```python\nimport os\nimport redis.asyncio as redis\nfrom redis_pydantic.base import BaseRedisModel\n\nredis_url = os.getenv(\"REDIS_URL\", \"redis://localhost:6379/0\")\n\nclass MyModel(BaseRedisModel):\n name: str\n \n class Meta:\n redis = redis.from_url(redis_url)\n```\n\n### Connection with Authentication\n\n```python\nimport redis.asyncio as redis\nfrom redis_pydantic.base import BaseRedisModel\n\nclass MyModel(BaseRedisModel):\n name: str\n \n class Meta:\n redis = redis.from_url(\n \"redis://username:password@your-redis-host:6379/0\",\n decode_responses=True\n )\n```\n\n## Supported Types and Operations\n\n### String (RedisStr)\n\n```python\nclass MyModel(BaseRedisModel):\n name: str = \"default\"\n\n# Operations\nawait model.name.set(\"new_value\") # Set string value\nawait model.name.load() # Load from Redis\n```\n\n### Integer (RedisInt)\n\n```python\nclass MyModel(BaseRedisModel):\n counter: int = 0\n\n# Operations\nawait model.counter.set(42) # Set integer value\nawait model.counter.load() # Load from Redis\n```\n\n### Boolean (RedisBool)\n\n```python\nclass MyModel(BaseRedisModel):\n is_active: bool = True\n\n# Operations\nawait model.is_active.set(False) # Set boolean value\nawait model.is_active.load() # Load from Redis\n```\n\n### Bytes (RedisBytes)\n\n```python\nclass MyModel(BaseRedisModel):\n data: bytes = b\"\"\n\n# Operations\nawait model.data.set(b\"binary_data\") # Set bytes value\nawait model.data.load() # Load from Redis\n```\n\n### List (RedisList)\n\n```python\nclass MyModel(BaseRedisModel):\n items: List[str] = []\n\n# Operations\nawait model.items.aappend(\"item1\") # Append single item\nawait model.items.aextend([\"item2\", \"item3\"]) # Extend with multiple items\nawait model.items.ainsert(0, \"first\") # Insert at specific index\npopped = await model.items.apop() # Pop last item\npopped = await model.items.apop(0) # Pop item at index\nawait model.items.aclear() # Clear all items\nawait model.items.load() # Load from Redis\n```\n\n### Dictionary (RedisDict)\n\n```python\nclass MyModel(BaseRedisModel):\n metadata: Dict[str, str] = {}\n\n# Operations\nawait model.metadata.aset_item(\"key\", \"value\") # Set single item\nawait model.metadata.aupdate(key1=\"val1\", key2=\"val2\") # Update multiple items\nawait model.metadata.adel_item(\"key\") # Delete item\npopped = await model.metadata.apop(\"key\") # Pop item by key\npopped = await model.metadata.apop(\"key\", \"default\") # Pop with default\nkey, value = await model.metadata.apopitem() # Pop arbitrary item\nawait model.metadata.aclear() # Clear all items\nawait model.metadata.load() # Load from Redis\n```\n\n## Advanced Usage\n\n### Lock Context Manager\n\nUse the lock context manager to ensure atomic operations across multiple fields with automatic model synchronization:\n\n```python\nclass User(BaseRedisModel):\n name: str\n balance: int = 0\n transaction_count: int = 0\n\nuser = User(name=\"John\", balance=1000)\nawait user.save()\n\n# Lock with default action\nasync with user.lock() as locked_user:\n # The locked_user is automatically refreshed from Redis\n locked_user.balance -= 50\n locked_user.transaction_count += 1\n # Changes are automatically saved when exiting the context\n\n# Lock with specific action name\nasync with user.lock(\"transfer\") as locked_user:\n # This creates a lock with key \"User:user_id/transfer\"\n locked_user.balance -= 100\n locked_user.transaction_count += 1\n # Automatic save on context exit\n```\n\nThe lock context manager:\n- Creates a Redis lock with key `{model_key}/{action}`\n- Automatically refreshes the model with latest Redis data on entry\n- Saves all changes back to Redis on successful exit\n- Ensures atomic operations across multiple field updates\n\n### Working with Nested Models\n\nRedisPydantic automatically supports nested Pydantic models by converting regular `BaseModel` classes into Redis-enabled versions. This allows you to use all Redis field operations on nested model fields.\n\n```python\nfrom pydantic import BaseModel, Field\n\nclass UserProfile(BaseModel):\n bio: str = \"\"\n skills: List[str] = Field(default_factory=list)\n settings: Dict[str, bool] = Field(default_factory=dict)\n\nclass Address(BaseModel):\n street: str\n city: str\n country: str = \"US\"\n\nclass User(BaseRedisModel):\n name: str\n profile: UserProfile = Field(default_factory=UserProfile)\n address: Address\n tags: List[str] = Field(default_factory=list)\n\nuser = User(name=\"John\", address=Address(street=\"123 Main St\", city=\"Boston\"))\nawait user.save()\n\n# Access Redis operations on nested model fields\nawait user.profile.skills.aappend(\"Python\")\nawait user.profile.skills.aextend([\"Redis\", \"AsyncIO\"])\nawait user.profile.settings.aupdate(dark_mode=True, notifications=False)\n\n# Even deeply nested operations work\nawait user.profile.skills.ainsert(0, \"Leadership\")\npopped_skill = await user.profile.skills.apop()\n\n# All Redis list/dict operations are available on nested fields\nawait user.profile.settings.aset_item(\"email_updates\", True)\nawait user.profile.settings.adel_item(\"notifications\")\n\n# Load specific nested fields\nawait user.profile.skills.load()\nawait user.profile.settings.load()\n\nprint(user.profile.skills) # Reflects Redis state\nprint(user.profile.settings) # Reflects Redis state\n```\n\n#### Deep Nesting Support\n\nRedisPydantic supports unlimited nesting depth:\n\n```python\nclass InnerModel(BaseModel):\n items: List[str] = Field(default_factory=list)\n counter: int = 0\n\nclass MiddleModel(BaseModel):\n inner: InnerModel = Field(default_factory=InnerModel)\n tags: List[str] = Field(default_factory=list)\n\nclass OuterModel(BaseRedisModel):\n middle: MiddleModel = Field(default_factory=MiddleModel)\n data: Dict[str, int] = Field(default_factory=dict)\n\nouter = OuterModel()\nawait outer.save()\n\n# All Redis operations work at any nesting level\nawait outer.middle.inner.items.aappend(\"deep_item\")\nawait outer.middle.tags.aextend([\"tag1\", \"tag2\"])\nawait outer.data.aset_item(\"count\", 42)\n\n# Load nested data\nawait outer.middle.inner.items.load()\nawait outer.middle.tags.load()\n```\n\n#### Nested Model Persistence\n\nNested models maintain full persistence and consistency:\n\n```python\n# Create and modify nested data\nuser1 = User(name=\"Alice\", address=Address(street=\"456 Oak Ave\", city=\"Seattle\"))\nawait user1.save()\nawait user1.profile.skills.aextend([\"JavaScript\", \"TypeScript\"])\n\n# Access from different instance\nuser2 = User()\nuser2.pk = user1.pk\nawait user2.profile.skills.load()\nprint(user2.profile.skills) # [\"JavaScript\", \"TypeScript\"]\n\n# All operations are atomic and persistent\nawait user2.profile.skills.aappend(\"React\")\nawait user1.profile.skills.load() # user1 now sees the new skill\n```\n\n### Working with Nested Types\n\n```python\nclass User(BaseRedisModel):\n preferences: Dict[str, List[str]] = {}\n scores: List[int] = []\n\nuser = User()\n\n# Nested operations\nawait user.preferences.aset_item(\"languages\", [\"python\", \"rust\"])\nawait user.scores.aextend([95, 87, 92])\n```\n\n### Loading Specific Fields\n\n```python\nuser = User()\nuser.pk = \"some-existing-id\"\n\n# Load only specific fields from Redis\nawait user.name.load()\nawait user.tags.load()\n# Other fields remain unloaded\n```\n\n### Atomic Operations\n\nAll Redis operations are atomic. For example:\n\n```python\n# This is atomic - either both operations succeed or both fail\nawait user.metadata.aupdate(\n last_login=str(datetime.now()),\n session_count=str(session_count + 1)\n)\n\n# This is also atomic\npopped_item = await user.items.apop() # Atomically removes and returns\n```\n\n### Model Serialization\n\n```python\n# Get the model as a Redis-compatible dict\nredis_data = user.redis_dump()\n\n# Save entire model to Redis\nawait user.save()\n```\n\n## Key Features\n\n### Automatic Type Conversion\n\nRedisPydantic handles serialization and deserialization automatically:\n\n```python\nclass MyModel(BaseRedisModel):\n data: bytes = b\"\"\n\nmodel = MyModel()\nawait model.data.set(b\"binary_data\") # Automatically base64 encoded in Redis\nloaded_data = await model.data.load() # Automatically decoded back to bytes\n```\n\n### Type Safety\n\nAll operations maintain type safety:\n\n```python\nclass MyModel(BaseRedisModel):\n count: int = 0\n\n# This will raise TypeError\nawait model.count.set(\"not_a_number\") # \u274c TypeError: Value must be int\n\n# This works\nawait model.count.set(42) # \u2705 Valid\n```\n\n### Consistent Local and Redis State\n\nRedisPydantic keeps your local Python objects in sync with Redis:\n\n```python\nuser = User(tags=[\"python\"])\nawait user.save()\n\nawait user.tags.aappend(\"redis\") # Updates both local list and Redis\nprint(user.tags) # [\"python\", \"redis\"] - local state is updated\n```\n\n## Error Handling\n\n```python\ntry:\n # Attempt to pop from empty list\n item = await user.tags.apop()\nexcept IndexError:\n print(\"List is empty\")\n\ntry:\n # Attempt to pop non-existent key from dict\n value = await user.metadata.apop(\"nonexistent_key\")\nexcept KeyError:\n print(\"Key not found\")\n\n# Using default values\nvalue = await user.metadata.apop(\"key\", \"default_value\") # Returns default if key missing\n```\n\n## Performance Tips\n\n1. **Batch Operations**: Use `aupdate()` for multiple dict updates and `aextend()` for multiple list items\n2. **Load Only What You Need**: Load specific fields instead of entire models when possible\n3. **Use Appropriate Data Types**: Choose the right Redis type for your use case\n4. **Connection Pooling**: Configure Redis connection pools for high-concurrency applications\n\n## Examples\n\n### User Session Management\n\n```python\nclass UserSession(BaseRedisModel):\n user_id: str\n session_data: Dict[str, str] = {}\n activity_log: List[str] = []\n is_active: bool = True\n last_seen: str = \"\"\n\nsession = UserSession(user_id=\"user123\")\nawait session.save()\n\n# Track user activity\nawait session.activity_log.aappend(f\"login:{datetime.now()}\")\nawait session.session_data.aupdate(\n ip_address=\"192.168.1.1\",\n user_agent=\"Chrome/91.0\"\n)\n```\n\n### Shopping Cart\n\n```python\nclass ShoppingCart(BaseRedisModel):\n user_id: str\n items: List[str] = [] # product IDs\n quantities: Dict[str, int] = {}\n total_amount: int = 0 # in cents\n\ncart = ShoppingCart(user_id=\"user456\")\n\n# Add items\nawait cart.items.aappend(\"product123\")\nawait cart.quantities.aset_item(\"product123\", 2)\n\n# Update totals\nawait cart.total_amount.set(4999) # $49.99\n```\n\n### Configuration Management\n\n```python\nclass AppConfig(BaseRedisModel):\n features: Dict[str, bool] = {}\n limits: Dict[str, int] = {}\n allowed_ips: List[str] = []\n\nconfig = AppConfig()\n\n# Enable feature flags\nawait config.features.aupdate(\n new_ui=True,\n beta_features=False\n)\n\n# Set rate limits\nawait config.limits.aupdate(\n requests_per_minute=1000,\n max_file_size=10485760\n)\n```\n\n## License\n\nMIT License\n\n## Contributing\n\nContributions are welcome! Please feel free to submit a Pull Request.\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "Pydantic models with Redis as the backend",
"version": "1.0.0",
"project_urls": {
"Documentation": "https://github.com/YedidyaHKfir/redis-pydantic",
"Homepage": "https://github.com/YedidyaHKfir/redis-pydantic",
"Repository": "https://github.com/YedidyaHKfir/redis-pydantic"
},
"split_keywords": [
"redis",
" pydantic",
" orm",
" database",
" async"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "8fdcb653a4226e67cff1f04344c7aa0e5fdc80c4ef8f99938422d5f3cb7b26b4",
"md5": "e7ca64964e4b17d33d22670e7b884b68",
"sha256": "817604943275897581fb67e1ce866c853669d82d4b5b152f3fa9458f6eee8dfe"
},
"downloads": -1,
"filename": "redis_pydantic-1.0.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "e7ca64964e4b17d33d22670e7b884b68",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4.0,>=3.10",
"size": 17475,
"upload_time": "2025-10-09T19:03:41",
"upload_time_iso_8601": "2025-10-09T19:03:41.328521Z",
"url": "https://files.pythonhosted.org/packages/8f/dc/b653a4226e67cff1f04344c7aa0e5fdc80c4ef8f99938422d5f3cb7b26b4/redis_pydantic-1.0.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "0f3687331f2f666aa452f6a1f80dfad2479808cefb5fb760418f04ac23c4738f",
"md5": "83daac474c02da334cc63ec147050dce",
"sha256": "98c4e2f08857ef1765e54f46c4823a9c0120f8d8b0e62d520d082549a25fcf47"
},
"downloads": -1,
"filename": "redis_pydantic-1.0.0.tar.gz",
"has_sig": false,
"md5_digest": "83daac474c02da334cc63ec147050dce",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4.0,>=3.10",
"size": 16124,
"upload_time": "2025-10-09T19:03:42",
"upload_time_iso_8601": "2025-10-09T19:03:42.672603Z",
"url": "https://files.pythonhosted.org/packages/0f/36/87331f2f666aa452f6a1f80dfad2479808cefb5fb760418f04ac23c4738f/redis_pydantic-1.0.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-10-09 19:03:42",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "YedidyaHKfir",
"github_project": "redis-pydantic",
"github_not_found": true,
"lcname": "redis-pydantic"
}
JSON
