Pydantic-redis

A simple declarative ORM for redis based on pydantic

Features

1. A subclass-able Model class to create Object Relational Mapping to redis hashes
2. A redis Store class to mutate and query Model's registered in it
3. A RedisConfig class to pass to the Store constructor to connect to a redis instance
4. A synchronous syncio and an asynchronous asyncio interface to the above classes

Installation

$ pip install pydantic-redis

---> 100%

Synchronous Example

Create it

• Create a file main.py with:

import pprint
from datetime import date
from typing import Tuple, List
from pydantic_redis import RedisConfig, Model, Store


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
    tags: List[str] = []
    in_stock: bool = True


if __name__ == "__main__":
    pp = pprint.PrettyPrinter(indent=4)
    store = Store(
        name="some_name",
        redis_config=RedisConfig(db=5, host="localhost", port=6379),
        life_span_in_seconds=3600,
    )

    store.register_model(Book)
    store.register_model(Author)

    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"],
        ),
        Book(
            title="Great Expectations",
            author=authors["charles"],
            published_on=date(year=1220, month=4, day=4),
            rating=5,
            tags=["Classic"],
        ),
        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"],
        ),
        Book(
            title="Wuthering Heights",
            author=authors["jane"],
            published_on=date(year=1600, month=4, day=4),
            rating=4.0,
            tags=["Classic", "Romance"],
        ),
    ]

    Book.insert(books, life_span_seconds=3600)
    all_books = Book.select()
    paginated_books = Book.select(skip=2, limit=2)
    paginated_books_with_few_fields = Book.select(
        columns=["author", "in_stock"], skip=2, limit=2
    )
    print("All:")
    pp.pprint(all_books)
    print("\nPaginated:")
    pp.pprint(paginated_books)
    print("\nPaginated but with few fields:")
    pp.pprint(paginated_books_with_few_fields)

Run it

Run the example with:

$ python main.py
All:
[   Book(title='Wuthering Heights', author=Author(name='Jane Austen', active_years=(1580, 1640)), rating=4.0, published_on=datetime.date(1600, 4, 4), tags=['Classic', 'Romance'], in_stock=True),
    Book(title='Oliver Twist', author=Author(name='Charles Dickens', active_years=(1220, 1280)), rating=2.0, published_on=datetime.date(1215, 4, 4), tags=['Classic'], in_stock=False),
    Book(title='Jane Eyre', author=Author(name='Charles Dickens', active_years=(1220, 1280)), rating=3.4, published_on=datetime.date(1225, 6, 4), tags=['Classic', 'Romance'], in_stock=False),
    Book(title='Great Expectations', author=Author(name='Charles Dickens', active_years=(1220, 1280)), rating=5.0, published_on=datetime.date(1220, 4, 4), tags=['Classic'], in_stock=True)]

Paginated:
[   Book(title='Jane Eyre', author=Author(name='Charles Dickens', active_years=(1220, 1280)), rating=3.4, published_on=datetime.date(1225, 6, 4), tags=['Classic', 'Romance'], in_stock=False),
    Book(title='Wuthering Heights', author=Author(name='Jane Austen', active_years=(1580, 1640)), rating=4.0, published_on=datetime.date(1600, 4, 4), tags=['Classic', 'Romance'], in_stock=True)]

Paginated but with few fields:
[   {   'author': Author(name='Charles Dickens', active_years=(1220, 1280)),
        'in_stock': False},
    {   'author': Author(name='Jane Austen', active_years=(1580, 1640)),
        'in_stock': True}]

Asynchronous Example

Create it

• Create a file main.py with:

import asyncio
import pprint
from datetime import date
from typing import Tuple, List
from pydantic_redis.asyncio import RedisConfig, Model, Store


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
    tags: List[str] = []
    in_stock: bool = True


async def run_async():
    pp = pprint.PrettyPrinter(indent=4)
    store = Store(
        name="some_name",
        redis_config=RedisConfig(db=5, host="localhost", port=6379),
        life_span_in_seconds=3600,
    )

    store.register_model(Book)
    store.register_model(Author)

    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"],
        ),
        Book(
            title="Great Expectations",
            author=authors["charles"],
            published_on=date(year=1220, month=4, day=4),
            rating=5,
            tags=["Classic"],
        ),
        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"],
        ),
        Book(
            title="Wuthering Heights",
            author=authors["jane"],
            published_on=date(year=1600, month=4, day=4),
            rating=4.0,
            tags=["Classic", "Romance"],
        ),
    ]

    await Book.insert(books, life_span_seconds=3600)
    all_books = await Book.select()
    paginated_books = await Book.select(skip=2, limit=2)
    paginated_books_with_few_fields = await Book.select(
        columns=["author", "in_stock"], skip=2, limit=2
    )
    print("All:")
    pp.pprint(all_books)
    print("\nPaginated:")
    pp.pprint(paginated_books)
    print("\nPaginated but with few fields:")
    pp.pprint(paginated_books_with_few_fields)


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(run_async())

Run it

Run the example with:

$ python main.py
All:
[   Book(title='Wuthering Heights', author=Author(name='Jane Austen', active_years=(1580, 1640)), rating=4.0, published_on=datetime.date(1600, 4, 4), tags=['Classic', 'Romance'], in_stock=True),
    Book(title='Oliver Twist', author=Author(name='Charles Dickens', active_years=(1220, 1280)), rating=2.0, published_on=datetime.date(1215, 4, 4), tags=['Classic'], in_stock=False),
    Book(title='Jane Eyre', author=Author(name='Charles Dickens', active_years=(1220, 1280)), rating=3.4, published_on=datetime.date(1225, 6, 4), tags=['Classic', 'Romance'], in_stock=False),
    Book(title='Great Expectations', author=Author(name='Charles Dickens', active_years=(1220, 1280)), rating=5.0, published_on=datetime.date(1220, 4, 4), tags=['Classic'], in_stock=True)]

Paginated:
[   Book(title='Jane Eyre', author=Author(name='Charles Dickens', active_years=(1220, 1280)), rating=3.4, published_on=datetime.date(1225, 6, 4), tags=['Classic', 'Romance'], in_stock=False),
    Book(title='Wuthering Heights', author=Author(name='Jane Austen', active_years=(1580, 1640)), rating=4.0, published_on=datetime.date(1600, 4, 4), tags=['Classic', 'Romance'], in_stock=True)]

Paginated but with few fields:
[   {   'author': Author(name='Charles Dickens', active_years=(1220, 1280)),
        'in_stock': False},
    {   'author': Author(name='Jane Austen', active_years=(1580, 1640)),
        'in_stock': True}]