The fantastic pydantic-aioredis package enables an async API that closely resembles pydantic-redis. However, currently, it lacks some features that are already in pydantic-redis and even as the packages mature, the divide between the two is expected to increase.
Add an async API like:
import asyncio
from datetime import date, datetime
from typing import Tuple, List, Dict
from pydantic_redis.asyncio import Store, Model, RedisConfig
class Author(Model):
_primary_key_field: str = 'name'
name: str
active_years: Tuple[int, int]
class Book(Model):
_primary_key_field: str = 'title'
title: str
author: Author
rating: float
published_on: date
last_updated: datetime = datetime.utcnow()
tags: List[str] = []
meta: Dict[str, str] = {}
in_stock: bool = True
authors = {
"charles": Author(name="Charles Dickens", active_years=(1220, 1280)),
"jane": Author(name="Jane Austen", active_years=(1580, 1640)),
}
books = [
Book(title="Oliver Twist", author=authors["charles"], published_on=date(year=1215, month=4, day=4),
in_stock=False, rating=2, tags=["Classic"], meta={"isbn": "63598943843898IN"}),
Book(title="Great Expectations", author=authors["charles"], published_on=date(year=1220, month=4, day=4), rating=5,
tags=["Classic"], meta={"isbn": "629389285uy", "description": "Some fancy book"}),
Book(title="Jane Eyre", author=authors["charles"], published_on=date(year=1225, month=6, day=4), in_stock=False,
rating=3.4, tags=["Classic", "Romance"], meta={"isbn": "787873ty"}),
Book(title="Wuthering Heights", author=authors["jane"], published_on=date(year=1600, month=4, day=4), rating=4.0,
tags=["Classic", "Romance"], last_updated=datetime.utcnow()),
]
async def run_async():
store = Store(name="test", redis_config=RedisConfig())
await store.redis_store.flushall()
store.register_model(Author)
store.register_model(Book)
# clear the store
await store.clear()
# Insert books and their associated models into redis
await Book.insert(books)
# Select all books to view them. A list of Model instances will be returned
# Will print [Book(title="Oliver Twist", author="Charles Dickens", ...), Book(...]
all_books = await Book.select()
print(all_books)
# Or select some books
some_books = await Book.select(ids=["Oliver Twist", "Jane Eyre"])
print(some_books) # Will print only those two books
# Or select some authors
# Will print Jane Austen even though you didn't explicitly insert her in the Author's collection
some_authors = await Author.select(ids=["Jane Austen"])
print(some_authors)
# Or select some columns. THIS RETURNS DICTIONARIES not MODEL Instances
# The Dictionaries have values in string form so you might need to do some extra work
books_with_few_fields = await Book.select(columns=["author", "in_stock"])
print(books_with_few_fields) # Will print [{"author": "'Charles Dickens", "in_stock": "True"},...]
# Update any book or library
await Book.update(_id="Oliver Twist", data={"author": authors["jane"]})
# You could even update a given author's details by nesting their new data in a book update
updated_jane = Author(**authors["jane"].dict())
updated_jane.active_years = (1999, 2008)
await Book.update(_id="Oliver Twist", data={"author": updated_jane})
# Trying to retrieve jane directly will return her with the new details
# All other books that have Jane Austen as author will also have their data updated. (like a real relationship)
new_jane = await Author.select(ids=["Jane Austen"])
print(new_jane)
# Delete any number of items
await Book.delete(ids=["Oliver Twist", "Jane Eyre"])
asyncio.run(run_async())
The fantastic pydantic-aioredis package enables an async API that closely resembles pydantic-redis. However, currently, it lacks some features that are already in pydantic-redis and even as the packages mature, the divide between the two is expected to increase.
Add an async API like:
This is especially useful for asyncio in FastAPI