Nested Models

The very first thing you need to create for pydantic-redis are the models (or schemas) that the data you are to save in redis is to be based on.

It is possible to refer one model in another model in a parent-child relationship.

Import Pydantic-redis' `Model`

First, import pydantic-redis.asyncio's Model.

Warning

The imports are from pydantic_redis.asyncio NOT pydantic_redis

import asyncio
import pprint
from datetime import date
from typing import List, Tuple

from pydantic_redis.asyncio import Model, Store, 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
    tags: List[str] = []
    in_stock: bool = True


async def 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(Author)
    store.register_model(Book)

    await Book.insert(
        Book(
            title="Oliver Twist",
            author=Author(name="Charles Dickens", active_years=(1999, 2007)),
            published_on=date(year=1215, month=4, day=4),
            in_stock=False,
            rating=2,
            tags=["Classic"],
        )
    )

    book_response = await Book.select(ids=["Oliver Twist"])
    author_response = await Author.select(ids=["Charles Dickens"])

    await Author.update(_id="Charles Dickens", data={"active_years": (1227, 1277)})
    updated_book_response = await Book.select(ids=["Oliver Twist"])

    await Book.update(
        _id="Oliver Twist",
        data={"author": Author(name="Charles Dickens", active_years=(1969, 1999))},
    )
    updated_author_response = await Author.select(ids=["Charles Dickens"])

    print("book:")
    pp.pprint(book_response)
    print("\nauthor:")
    pp.pprint(author_response)

    print("\nindirectly updated book:")
    pp.pprint(updated_book_response)
    print("\nindirectly updated author:")
    pp.pprint(updated_author_response)


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

Create the Child Model

Next, declare a new model as a class that inherits from Model.

Use standard Python types for all attributes.

import asyncio
import pprint
from datetime import date
from typing import List, Tuple

from pydantic_redis.asyncio import Model, Store, 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
    tags: List[str] = []
    in_stock: bool = True


async def 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(Author)
    store.register_model(Book)

    await Book.insert(
        Book(
            title="Oliver Twist",
            author=Author(name="Charles Dickens", active_years=(1999, 2007)),
            published_on=date(year=1215, month=4, day=4),
            in_stock=False,
            rating=2,
            tags=["Classic"],
        )
    )

    book_response = await Book.select(ids=["Oliver Twist"])
    author_response = await Author.select(ids=["Charles Dickens"])

    await Author.update(_id="Charles Dickens", data={"active_years": (1227, 1277)})
    updated_book_response = await Book.select(ids=["Oliver Twist"])

    await Book.update(
        _id="Oliver Twist",
        data={"author": Author(name="Charles Dickens", active_years=(1969, 1999))},
    )
    updated_author_response = await Author.select(ids=["Charles Dickens"])

    print("book:")
    pp.pprint(book_response)
    print("\nauthor:")
    pp.pprint(author_response)

    print("\nindirectly updated book:")
    pp.pprint(updated_book_response)
    print("\nindirectly updated author:")
    pp.pprint(updated_author_response)


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

Set the `_primary_key_field` of the Child Model

Set the _primary_key_field attribute to the name of the attribute that is to act as a unique identifier for each instance of the Model.

Example

In this case, there can be no two authors with the same name.

import asyncio
import pprint
from datetime import date
from typing import List, Tuple

from pydantic_redis.asyncio import Model, Store, 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
    tags: List[str] = []
    in_stock: bool = True


async def 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(Author)
    store.register_model(Book)

    await Book.insert(
        Book(
            title="Oliver Twist",
            author=Author(name="Charles Dickens", active_years=(1999, 2007)),
            published_on=date(year=1215, month=4, day=4),
            in_stock=False,
            rating=2,
            tags=["Classic"],
        )
    )

    book_response = await Book.select(ids=["Oliver Twist"])
    author_response = await Author.select(ids=["Charles Dickens"])

    await Author.update(_id="Charles Dickens", data={"active_years": (1227, 1277)})
    updated_book_response = await Book.select(ids=["Oliver Twist"])

    await Book.update(
        _id="Oliver Twist",
        data={"author": Author(name="Charles Dickens", active_years=(1969, 1999))},
    )
    updated_author_response = await Author.select(ids=["Charles Dickens"])

    print("book:")
    pp.pprint(book_response)
    print("\nauthor:")
    pp.pprint(author_response)

    print("\nindirectly updated book:")
    pp.pprint(updated_book_response)
    print("\nindirectly updated author:")
    pp.pprint(updated_author_response)


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

Create the Parent Model

Next, declare another model as a class that inherits from Model.

Use standard Python types for all attributes, as before.

import asyncio
import pprint
from datetime import date
from typing import List, Tuple

from pydantic_redis.asyncio import Model, Store, 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
    tags: List[str] = []
    in_stock: bool = True


async def 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(Author)
    store.register_model(Book)

    await Book.insert(
        Book(
            title="Oliver Twist",
            author=Author(name="Charles Dickens", active_years=(1999, 2007)),
            published_on=date(year=1215, month=4, day=4),
            in_stock=False,
            rating=2,
            tags=["Classic"],
        )
    )

    book_response = await Book.select(ids=["Oliver Twist"])
    author_response = await Author.select(ids=["Charles Dickens"])

    await Author.update(_id="Charles Dickens", data={"active_years": (1227, 1277)})
    updated_book_response = await Book.select(ids=["Oliver Twist"])

    await Book.update(
        _id="Oliver Twist",
        data={"author": Author(name="Charles Dickens", active_years=(1969, 1999))},
    )
    updated_author_response = await Author.select(ids=["Charles Dickens"])

    print("book:")
    pp.pprint(book_response)
    print("\nauthor:")
    pp.pprint(author_response)

    print("\nindirectly updated book:")
    pp.pprint(updated_book_response)
    print("\nindirectly updated author:")
    pp.pprint(updated_author_response)


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

Add the Nested Model to the Parent Model

Annotate the field that is to hold the child model with the child class.

Example

In this case, the field author is annotated with Author class.

import asyncio
import pprint
from datetime import date
from typing import List, Tuple

from pydantic_redis.asyncio import Model, Store, 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
    tags: List[str] = []
    in_stock: bool = True


async def 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(Author)
    store.register_model(Book)

    await Book.insert(
        Book(
            title="Oliver Twist",
            author=Author(name="Charles Dickens", active_years=(1999, 2007)),
            published_on=date(year=1215, month=4, day=4),
            in_stock=False,
            rating=2,
            tags=["Classic"],
        )
    )

    book_response = await Book.select(ids=["Oliver Twist"])
    author_response = await Author.select(ids=["Charles Dickens"])

    await Author.update(_id="Charles Dickens", data={"active_years": (1227, 1277)})
    updated_book_response = await Book.select(ids=["Oliver Twist"])

    await Book.update(
        _id="Oliver Twist",
        data={"author": Author(name="Charles Dickens", active_years=(1969, 1999))},
    )
    updated_author_response = await Author.select(ids=["Charles Dickens"])

    print("book:")
    pp.pprint(book_response)
    print("\nauthor:")
    pp.pprint(author_response)

    print("\nindirectly updated book:")
    pp.pprint(updated_book_response)
    print("\nindirectly updated author:")
    pp.pprint(updated_author_response)


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

Set the `_primary_key_field` of the Parent Model

Set the _primary_key_field attribute to the name of the attribute that is to act as a unique identifier for each instance of the parent Model.

Example

In this case, there can be no two books with the same title.

import asyncio
import pprint
from datetime import date
from typing import List, Tuple

from pydantic_redis.asyncio import Model, Store, 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
    tags: List[str] = []
    in_stock: bool = True


async def 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(Author)
    store.register_model(Book)

    await Book.insert(
        Book(
            title="Oliver Twist",
            author=Author(name="Charles Dickens", active_years=(1999, 2007)),
            published_on=date(year=1215, month=4, day=4),
            in_stock=False,
            rating=2,
            tags=["Classic"],
        )
    )

    book_response = await Book.select(ids=["Oliver Twist"])
    author_response = await Author.select(ids=["Charles Dickens"])

    await Author.update(_id="Charles Dickens", data={"active_years": (1227, 1277)})
    updated_book_response = await Book.select(ids=["Oliver Twist"])

    await Book.update(
        _id="Oliver Twist",
        data={"author": Author(name="Charles Dickens", active_years=(1969, 1999))},
    )
    updated_author_response = await Author.select(ids=["Charles Dickens"])

    print("book:")
    pp.pprint(book_response)
    print("\nauthor:")
    pp.pprint(author_response)

    print("\nindirectly updated book:")
    pp.pprint(updated_book_response)
    print("\nindirectly updated author:")
    pp.pprint(updated_author_response)


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

Register the Models in the Store

Then, in order for the store to know the existence of each given model, register it using the register_model method of Store

import asyncio
import pprint
from datetime import date
from typing import List, Tuple

from pydantic_redis.asyncio import Model, Store, 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
    tags: List[str] = []
    in_stock: bool = True


async def 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(Author)
    store.register_model(Book)

    await Book.insert(
        Book(
            title="Oliver Twist",
            author=Author(name="Charles Dickens", active_years=(1999, 2007)),
            published_on=date(year=1215, month=4, day=4),
            in_stock=False,
            rating=2,
            tags=["Classic"],
        )
    )

    book_response = await Book.select(ids=["Oliver Twist"])
    author_response = await Author.select(ids=["Charles Dickens"])

    await Author.update(_id="Charles Dickens", data={"active_years": (1227, 1277)})
    updated_book_response = await Book.select(ids=["Oliver Twist"])

    await Book.update(
        _id="Oliver Twist",
        data={"author": Author(name="Charles Dickens", active_years=(1969, 1999))},
    )
    updated_author_response = await Author.select(ids=["Charles Dickens"])

    print("book:")
    pp.pprint(book_response)
    print("\nauthor:")
    pp.pprint(author_response)

    print("\nindirectly updated book:")
    pp.pprint(updated_book_response)
    print("\nindirectly updated author:")
    pp.pprint(updated_author_response)


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

Use the Parent Model

Then you can use the parent model class to:

insert into the store
update an instance of the model
delete from store
select from store

Note

The child model will be automatically inserted, or updated if it already exists

Info

The store is connected to the Redis instance, so any changes you make will reflect in redis itself.

import asyncio
import pprint
from datetime import date
from typing import List, Tuple

from pydantic_redis.asyncio import Model, Store, 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
    tags: List[str] = []
    in_stock: bool = True


async def 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(Author)
    store.register_model(Book)

    await Book.insert(
        Book(
            title="Oliver Twist",
            author=Author(name="Charles Dickens", active_years=(1999, 2007)),
            published_on=date(year=1215, month=4, day=4),
            in_stock=False,
            rating=2,
            tags=["Classic"],
        )
    )

    book_response = await Book.select(ids=["Oliver Twist"])
    author_response = await Author.select(ids=["Charles Dickens"])

    await Author.update(_id="Charles Dickens", data={"active_years": (1227, 1277)})
    updated_book_response = await Book.select(ids=["Oliver Twist"])

    await Book.update(
        _id="Oliver Twist",
        data={"author": Author(name="Charles Dickens", active_years=(1969, 1999))},
    )
    updated_author_response = await Author.select(ids=["Charles Dickens"])

    print("book:")
    pp.pprint(book_response)
    print("\nauthor:")
    pp.pprint(author_response)

    print("\nindirectly updated book:")
    pp.pprint(updated_book_response)
    print("\nindirectly updated author:")
    pp.pprint(updated_author_response)


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

Use the Child Model Independently

You can also use the child model independently.

Info

Any mutation on the child model will also be reflected in the any parent model instances fetched from redis after that mutation.

import asyncio
import pprint
from datetime import date
from typing import List, Tuple

from pydantic_redis.asyncio import Model, Store, 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
    tags: List[str] = []
    in_stock: bool = True


async def 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(Author)
    store.register_model(Book)

    await Book.insert(
        Book(
            title="Oliver Twist",
            author=Author(name="Charles Dickens", active_years=(1999, 2007)),
            published_on=date(year=1215, month=4, day=4),
            in_stock=False,
            rating=2,
            tags=["Classic"],
        )
    )

    book_response = await Book.select(ids=["Oliver Twist"])
    author_response = await Author.select(ids=["Charles Dickens"])

    await Author.update(_id="Charles Dickens", data={"active_years": (1227, 1277)})
    updated_book_response = await Book.select(ids=["Oliver Twist"])

    await Book.update(
        _id="Oliver Twist",
        data={"author": Author(name="Charles Dickens", active_years=(1969, 1999))},
    )
    updated_author_response = await Author.select(ids=["Charles Dickens"])

    print("book:")
    pp.pprint(book_response)
    print("\nauthor:")
    pp.pprint(author_response)

    print("\nindirectly updated book:")
    pp.pprint(updated_book_response)
    print("\nindirectly updated author:")
    pp.pprint(updated_author_response)


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

Indirectly Update Child Model

A child model can be indirectly updated via the parent model.

Set the attribute containing the child model with an instance of the child model

Note

The new instance of the child model should have the SAME primary key as the original child model.

import asyncio
import pprint
from datetime import date
from typing import List, Tuple

from pydantic_redis.asyncio import Model, Store, 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
    tags: List[str] = []
    in_stock: bool = True


async def 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(Author)
    store.register_model(Book)

    await Book.insert(
        Book(
            title="Oliver Twist",
            author=Author(name="Charles Dickens", active_years=(1999, 2007)),
            published_on=date(year=1215, month=4, day=4),
            in_stock=False,
            rating=2,
            tags=["Classic"],
        )
    )

    book_response = await Book.select(ids=["Oliver Twist"])
    author_response = await Author.select(ids=["Charles Dickens"])

    await Author.update(_id="Charles Dickens", data={"active_years": (1227, 1277)})
    updated_book_response = await Book.select(ids=["Oliver Twist"])

    await Book.update(
        _id="Oliver Twist",
        data={"author": Author(name="Charles Dickens", active_years=(1969, 1999))},
    )
    updated_author_response = await Author.select(ids=["Charles Dickens"])

    print("book:")
    pp.pprint(book_response)
    print("\nauthor:")
    pp.pprint(author_response)

    print("\nindirectly updated book:")
    pp.pprint(updated_book_response)
    print("\nindirectly updated author:")
    pp.pprint(updated_author_response)


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

Run the App

Running the above code in a file main.py would produce:

Tip

Probably FLUSHALL redis first

$ python main.py
book:
[   Book(title='Oliver Twist', author=Author(name='Charles Dickens', active_years=(1999, 2007)), rating=2.0, published_on=datetime.date(1215, 4, 4), tags=['Classic'], in_stock=False)]

author:
[Author(name='Charles Dickens', active_years=(1999, 2007))]

indirectly updated book:
[   Book(title='Oliver Twist', author=Author(name='Charles Dickens', active_years=(1227, 1277)), rating=2.0, published_on=datetime.date(1215, 4, 4), tags=['Classic'], in_stock=False)]

indirectly updated author:
[Author(name='Charles Dickens', active_years=(1969, 1999))]

Nested Models

Import Pydantic-redis' Model

Create the Child Model

Set the _primary_key_field of the Child Model

Create the Parent Model

Add the Nested Model to the Parent Model

Set the _primary_key_field of the Parent Model

Register the Models in the Store

Use the Parent Model

Use the Child Model Independently

Indirectly Update Child Model

Run the App

Import Pydantic-redis' `Model`

Set the `_primary_key_field` of the Child Model

Set the `_primary_key_field` of the Parent Model