Beanis - Redis ODM for Humans

📚 Documentation | 🚀 Getting Started | 📖 Tutorial | 🔧 API Reference

---

Stop writing boilerplate Redis code. Focus on your application logic.

Beanis is an async Python ODM (Object-Document Mapper) for Redis that gives you Pydantic models, type safety, and a clean API - while staying fast and working with vanilla Redis.

Why Beanis?

The Problem with Vanilla Redis

❌ Manual serialization - You write json.dumps() and json.loads() everywhere ❌ Type conversions - Strings from Redis need manual float(), int() conversions ❌ Key management - You track "Product:123", "all:Product" keys manually ❌ No validation - Bad data silently corrupts your Redis database ❌ Boilerplate code - 15-20 lines for simple CRUD operations

The Solution: Beanis

✅ Automatic serialization - Nested objects, lists, custom types - all handled ✅ Type safety - Full Pydantic validation + IDE autocomplete ✅ Smart key management - Focus on your data, not Redis internals ✅ Data validation - Catch errors before they hit Redis ✅ Write 70% less code - 5-7 lines for the same operations

AND it's fast: Only 8% overhead vs vanilla Redis

Who Should Use Beanis?

✅ You're building a production app that needs Redis but not the boilerplate ✅ You want type safety and validation without sacrificing performance ✅ You're using vanilla Redis (no RedisJSON/RediSearch modules) ✅ You like Beanie's MongoDB API and want the same for Redis ✅ You're storing complex data (nested objects, NumPy arrays, etc.)

When NOT to Use Beanis?

❌ You need every microsecond of performance (use raw redis-py) ❌ You need RedisJSON/RediSearch features (use Redis OM) ❌ You're only storing simple key-value pairs (use raw redis-py)

Show Me The Code

Basic CRUD Operation

Vanilla Redis (20 lines)

Beanis (7 lines)

importjsonimporttimefromredis.asyncioimportRedisredis=Redis(decode_responses=True) # Insertproduct_data= { "name": "Tony's Chocolonely", "price": "5.95", "category": json.dumps({ "name": "Chocolate", "description": "Roasted cacao" }) } awaitredis.hset("Product:prod_123", mapping=product_data) awaitredis.zadd("all:Product", {"prod_123": time.time()}) # Retrieveraw=awaitredis.hgetall("Product:prod_123") product= { "name": raw["name"], "price": float(raw["price"]), "category": json.loads(raw["category"]) }

frombeanisimportDocumentfrompydanticimportBaseModelclassCategory(BaseModel): name: strdescription: strclassProduct(Document): name: strprice: floatcategory: Category# Insertproduct=Product( name="Tony's Chocolonely", price=5.95, category=Category( name="Chocolate", description="Roasted cacao" ) ) awaitproduct.insert() # Retrievefound=awaitProduct.get(product.id)

Result: Type-safe, validated, 65% less code

Search/Query Operation

Vanilla Redis (25 lines)

Beanis (4 lines)

# Find products between $10-50keys=awaitredis.zrangebyscore( "idx:Product:price", min=10.0, max=50.0 ) # Fetch each product using pipelinepipe=redis.pipeline() forkeyinkeys: pipe.hgetall(f"Product:{key}") results=awaitpipe.execute() # Parse manuallyproducts= [] fordatainresults: ifdata: products.append({ "name": data["name"], "price": float(data["price"]), "stock": int(data["stock"]), "category": json.loads( data.get("category", "{}") ) })

# Find products between $10-50products=awaitProduct.find( Product.price>=10.0, Product.price<=50.0 ).to_list()

Result:84% less code, fully typed results

Update Operation

Vanilla Redis (10 lines)	Beanis (6 lines)
# Update price and stockawaitredis.hset("Product:123", mapping={ "price": "6.95", "stock": "150" }) # Atomic incrementnew_stock=awaitredis.hincrby( "Product:123", "stock", -1 )	# Update fieldsawaitproduct.update( price=6.95, stock=150 ) # Atomic incrementnew_stock=awaitproduct.increment_field( "stock", -1 )

Result: Same functionality, cleaner API, type-safe

Batch Operations

Vanilla Redis (14 lines)	Beanis (9 lines)
# Insert 100 productspipe=redis.pipeline() foriinrange(100): product_id=f"prod_{i}"data= { "name": f"Product {i}", "price": str(i*10), "stock": "100", "category": json.dumps({ "name": "Category" }) } pipe.hset(f"Product:{product_id}", mapping=data) pipe.zadd("all:Product", {product_id: time.time()}) awaitpipe.execute()	# Insert 100 productsproducts= [ Product( name=f"Product {i}", price=i*10, stock=100, category=Category(name="Category") ) foriinrange(100) ] awaitProduct.insert_many(products)

Result:35% less code, no manual key management

Installation

PIP

pip install beanis

Poetry

poetry add beanis

Quick Start

importasynciofromtypingimportOptionalfromredis.asyncioimportRedisfrompydanticimportBaseModelfrombeanisimportDocument, init_beanis, IndexedclassCategory(BaseModel):
name: strdescription: strclassProduct(Document):
name: strdescription: Optional[str] =Noneprice: Indexed(float) # Indexed for range queriescategory: Categorystock: int=0classSettings:
name="products"asyncdefmain():
# Initialize Redis clientclient=Redis(host="localhost", port=6379, db=0, decode_responses=True)
# Initialize Beanisawaitinit_beanis(database=client, document_models=[Product])
# Create a productchocolate=Category(
name="Chocolate",
description="A preparation of roasted and ground cacao seeds."
)
product=Product(
name="Tony's Chocolonely",
price=5.95,
category=chocolate,
stock=100
)
# Insert into Redisawaitproduct.insert()
# Retrieve by IDfound=awaitProduct.get(product.id)
print(f"Found: {found.name} - ${found.price}")
# Query by price rangeaffordable=awaitProduct.find(
Product.price<10.0
).to_list()
print(f"Affordable products: {len(affordable)}")
# Update specific fieldsawaitproduct.update(price=6.95, stock=150)
# Atomic incrementnew_stock=awaitproduct.increment_field("stock", -1)
print(f"Stock after sale: {new_stock}")
# Get all productsall_products=awaitProduct.all()
print(f"Total products: {len(all_products)}")
# Deleteawaitproduct.delete_self()
awaitclient.close()
if__name__=="__main__":
asyncio.run(main())

Core Features

🚀 Type Safety & Validation

Beanis uses Pydantic models, giving you automatic validation and type checking:

classProduct(Document):
name: strprice: float# Must be a numberstock: int# Must be an integercategory: Category# Must be a valid Category object# This will raise a validation error BEFORE hitting Redisproduct=Product(
name="Invalid",
price="not a number", # ❌ ValidationError!stock=100
)

⚡ High Performance

Beanis is optimized for speed with minimal overhead:

• 8% overhead vs vanilla Redis (benchmarked)
• Uses msgspec for ultra-fast JSON parsing (2x faster than orjson)
• Skips Pydantic validation on reads by default (data from Redis is trusted)
• Efficient pipeline usage for batch operations

Benchmark Results (Get by ID):

• Vanilla Redis: 1.00x (baseline)
• Beanis: 1.08x (only 8% slower)
• Redis OM: 1.20x (20% slower)

🎯 Pythonic API

Familiar Beanie-style interface for MongoDB developers:

# Query with Pythonic operatorsproducts=awaitProduct.find(
Product.price>=10.0,
Product.price<=50.0,
Product.stock>0
).to_list()
# Chaining operationsexpensive=awaitProduct.find(
Product.price>100
).sort(Product.price).limit(10).to_list()
# Batch operationsawaitProduct.insert_many([product1, product2, product3])
products=awaitProduct.get_many([id1, id2, id3])
awaitProduct.delete_many([id1, id2])

📦 Store Anything

Beanis handles complex types automatically:

Built-in support:

• Nested Pydantic models
• Lists, dicts, tuples, sets
• Decimal, UUID, Enum
• datetime, date, time, timedelta

Custom types via encoders:

frombeanisimportDocument, register_typeimportnumpyasnp# NumPy arrays work automatically (auto-registered)classMLModel(Document):
name: strweights: Any# Stores np.ndarray!model=MLModel(name="v1", weights=np.random.rand(100, 100))
awaitmodel.insert() # Just works!# Custom typesregister_type(
MyCustomType,
encoder=lambdaobj: str(obj),
decoder=lambdas: MyCustomType.from_string(s)
)

See tutorial on how to create them!

🔧 Production Ready Features

TTL Support:

# Insert with TTLawaitproduct.insert(ttl=3600) # Expires in 1 hour# Set TTL on existing documentawaitproduct.set_ttl(7200)
ttl=awaitproduct.get_ttl()
awaitproduct.persist() # Remove TTL

Event Hooks:

frombeanisimportbefore_event, after_event, Insert, UpdateclassProduct(Document):
name: strprice: float@before_event(Insert)asyncdefvalidate_price(self):
ifself.price<0:
raiseValueError("Price cannot be negative")
@after_event(Insert)asyncdeflog_creation(self):
print(f"Created product: {self.name}")

Field-Level Operations:

# Get/set single field without loading entire documentprice=awaitproduct.get_field("price")
awaitproduct.set_field("stock", 200)
# Atomic incrementnew_stock=awaitproduct.increment_field("stock", 5)

Document Tracking:

# Get all documents (sorted by insertion time)all_products=awaitProduct.all()
# Paginationpage1=awaitProduct.all(limit=10)
page2=awaitProduct.all(skip=10, limit=10)
# Count and delete allcount=awaitProduct.count()
awaitProduct.delete_all()

Comparison

Feature	Vanilla Redis	Beanis	Redis OM
Code volume	100%	30% ⭐	50%
Type safety	Manual	Automatic ⭐	Automatic
Performance	100% ⭐	108%	120%
Vanilla Redis	✅	✅ ⭐	❌ Requires modules
Validation	Manual	Automatic ⭐	Automatic
API Style	Redis commands	Pythonic ⭐	Redis OM
Learning curve	Medium	Easy ⭐	Medium
Nested objects	Manual	Automatic ⭐	Automatic
Custom types	Manual	Easy ⭐	Limited
Event hooks	❌	✅ ⭐	❌
All DBs (0-15)	✅	✅ ⭐	❌ DB 0 only

Choosing the Right Tool

Choose Vanilla Redis when:

• Every microsecond matters (high-frequency trading, etc.)
• Simple key-value storage
• You're a Redis expert and don't need abstractions

Choose Beanis when: ⭐

• Building production applications with complex data models
• Want type safety + performance (8% overhead is acceptable)
• Using vanilla Redis (no RedisJSON/RediSearch modules)
• Need to store nested objects, custom types, NumPy arrays, etc.
• Coming from MongoDB/Beanie and want familiar API
• Want event hooks for validation and lifecycle management

Choose Redis OM when:

• You need RedisJSON/RediSearch features
• Don't mind installing Redis modules
• Want Redis Stack integration
• Need advanced full-text search

Requirements

• Python 3.8+
• Redis 5.0+
• Pydantic 1.10+ or 2.0+

Testing

# Run tests
pytest
# Run with coverage
pytest --cov=beanis
# Run specific test
pytest tests/test_core.py::test_insert_and_get

Credits

Beanis is a fork of Beanie - the amazing MongoDB ODM created by Roman Right and contributors.

We took the Beanie codebase and completely reimagined it for Redis, replacing MongoDB operations with Redis commands while preserving the elegant API design. If you're using MongoDB, check out the original Beanie - it's awesome!

Special thanks to:

• Roman Right and the Beanie community for creating the foundation
• All Beanie contributors whose code inspired this project
• The Redis and Pydantic teams for their excellent libraries

License

Apache License 2.0