Tuples of Nested Models

Sometimes, one might need to have models (schemas) that have tuples of other models (schemas).

An example is a ScoreBoard model that can have Tuples of player name and Scores'.

This can easily be pulled off with pydantic-redis.

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 typing import Tuple
from pydantic_redis.asyncio import RedisConfig, Model, Store


class Score(Model):
    _primary_key_field: str = "id"
    id: str
    total: int


class ScoreBoard(Model):
    _primary_key_field: str = "id"
    id: str
    scores: Tuple[str, Score]


async def main():
    pp = pprint.PrettyPrinter(indent=4)
    store = Store(name="test", redis_config=RedisConfig())

    store.register_model(Score)
    store.register_model(ScoreBoard)

    await ScoreBoard.insert(
        data=ScoreBoard(
            id="test",
            scores=(
                "mark",
                Score(id="some id", total=50),
            ),
        )
    )
    score_board_response = await ScoreBoard.select(ids=["test"])
    scores_response = await Score.select(ids=["some id"])

    await Score.update(_id="some id", data={"total": 78})
    updated_score_board_response = await ScoreBoard.select(ids=["test"])

    await ScoreBoard.update(
        _id="test",
        data={
            "scores": (
                "tom",
                Score(id="some id", total=60),
            )
        },
    )
    updated_score_response = await Score.select(ids=["some id"])

    print("score board:")
    pp.pprint(score_board_response)
    print("\nscores:")
    pp.pprint(scores_response)

    print("\nindirectly updated score board:")
    pp.pprint(updated_score_board_response)
    print("\nindirectly updated score:")
    pp.pprint(updated_score_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 typing import Tuple
from pydantic_redis.asyncio import RedisConfig, Model, Store


class Score(Model):
    _primary_key_field: str = "id"
    id: str
    total: int


class ScoreBoard(Model):
    _primary_key_field: str = "id"
    id: str
    scores: Tuple[str, Score]


async def main():
    pp = pprint.PrettyPrinter(indent=4)
    store = Store(name="test", redis_config=RedisConfig())

    store.register_model(Score)
    store.register_model(ScoreBoard)

    await ScoreBoard.insert(
        data=ScoreBoard(
            id="test",
            scores=(
                "mark",
                Score(id="some id", total=50),
            ),
        )
    )
    score_board_response = await ScoreBoard.select(ids=["test"])
    scores_response = await Score.select(ids=["some id"])

    await Score.update(_id="some id", data={"total": 78})
    updated_score_board_response = await ScoreBoard.select(ids=["test"])

    await ScoreBoard.update(
        _id="test",
        data={
            "scores": (
                "tom",
                Score(id="some id", total=60),
            )
        },
    )
    updated_score_response = await Score.select(ids=["some id"])

    print("score board:")
    pp.pprint(score_board_response)
    print("\nscores:")
    pp.pprint(scores_response)

    print("\nindirectly updated score board:")
    pp.pprint(updated_score_board_response)
    print("\nindirectly updated score:")
    pp.pprint(updated_score_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 Score's with the same id.

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


class Score(Model):
    _primary_key_field: str = "id"
    id: str
    total: int


class ScoreBoard(Model):
    _primary_key_field: str = "id"
    id: str
    scores: Tuple[str, Score]


async def main():
    pp = pprint.PrettyPrinter(indent=4)
    store = Store(name="test", redis_config=RedisConfig())

    store.register_model(Score)
    store.register_model(ScoreBoard)

    await ScoreBoard.insert(
        data=ScoreBoard(
            id="test",
            scores=(
                "mark",
                Score(id="some id", total=50),
            ),
        )
    )
    score_board_response = await ScoreBoard.select(ids=["test"])
    scores_response = await Score.select(ids=["some id"])

    await Score.update(_id="some id", data={"total": 78})
    updated_score_board_response = await ScoreBoard.select(ids=["test"])

    await ScoreBoard.update(
        _id="test",
        data={
            "scores": (
                "tom",
                Score(id="some id", total=60),
            )
        },
    )
    updated_score_response = await Score.select(ids=["some id"])

    print("score board:")
    pp.pprint(score_board_response)
    print("\nscores:")
    pp.pprint(scores_response)

    print("\nindirectly updated score board:")
    pp.pprint(updated_score_board_response)
    print("\nindirectly updated score:")
    pp.pprint(updated_score_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 typing import Tuple
from pydantic_redis.asyncio import RedisConfig, Model, Store


class Score(Model):
    _primary_key_field: str = "id"
    id: str
    total: int


class ScoreBoard(Model):
    _primary_key_field: str = "id"
    id: str
    scores: Tuple[str, Score]


async def main():
    pp = pprint.PrettyPrinter(indent=4)
    store = Store(name="test", redis_config=RedisConfig())

    store.register_model(Score)
    store.register_model(ScoreBoard)

    await ScoreBoard.insert(
        data=ScoreBoard(
            id="test",
            scores=(
                "mark",
                Score(id="some id", total=50),
            ),
        )
    )
    score_board_response = await ScoreBoard.select(ids=["test"])
    scores_response = await Score.select(ids=["some id"])

    await Score.update(_id="some id", data={"total": 78})
    updated_score_board_response = await ScoreBoard.select(ids=["test"])

    await ScoreBoard.update(
        _id="test",
        data={
            "scores": (
                "tom",
                Score(id="some id", total=60),
            )
        },
    )
    updated_score_response = await Score.select(ids=["some id"])

    print("score board:")
    pp.pprint(score_board_response)
    print("\nscores:")
    pp.pprint(scores_response)

    print("\nindirectly updated score board:")
    pp.pprint(updated_score_board_response)
    print("\nindirectly updated score:")
    pp.pprint(updated_score_response)


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

Add the Nested Model Tuple to the Parent Model

Annotate the field that is to hold the tuple of child models with the Tuple of child class.

Example

In this case, the field scores is annotated with Tuple[str, Score] class.

Info

The str is the player's name.

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


class Score(Model):
    _primary_key_field: str = "id"
    id: str
    total: int


class ScoreBoard(Model):
    _primary_key_field: str = "id"
    id: str
    scores: Tuple[str, Score]


async def main():
    pp = pprint.PrettyPrinter(indent=4)
    store = Store(name="test", redis_config=RedisConfig())

    store.register_model(Score)
    store.register_model(ScoreBoard)

    await ScoreBoard.insert(
        data=ScoreBoard(
            id="test",
            scores=(
                "mark",
                Score(id="some id", total=50),
            ),
        )
    )
    score_board_response = await ScoreBoard.select(ids=["test"])
    scores_response = await Score.select(ids=["some id"])

    await Score.update(_id="some id", data={"total": 78})
    updated_score_board_response = await ScoreBoard.select(ids=["test"])

    await ScoreBoard.update(
        _id="test",
        data={
            "scores": (
                "tom",
                Score(id="some id", total=60),
            )
        },
    )
    updated_score_response = await Score.select(ids=["some id"])

    print("score board:")
    pp.pprint(score_board_response)
    print("\nscores:")
    pp.pprint(scores_response)

    print("\nindirectly updated score board:")
    pp.pprint(updated_score_board_response)
    print("\nindirectly updated score:")
    pp.pprint(updated_score_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 ScoreBoard's with the same id.

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


class Score(Model):
    _primary_key_field: str = "id"
    id: str
    total: int


class ScoreBoard(Model):
    _primary_key_field: str = "id"
    id: str
    scores: Tuple[str, Score]


async def main():
    pp = pprint.PrettyPrinter(indent=4)
    store = Store(name="test", redis_config=RedisConfig())

    store.register_model(Score)
    store.register_model(ScoreBoard)

    await ScoreBoard.insert(
        data=ScoreBoard(
            id="test",
            scores=(
                "mark",
                Score(id="some id", total=50),
            ),
        )
    )
    score_board_response = await ScoreBoard.select(ids=["test"])
    scores_response = await Score.select(ids=["some id"])

    await Score.update(_id="some id", data={"total": 78})
    updated_score_board_response = await ScoreBoard.select(ids=["test"])

    await ScoreBoard.update(
        _id="test",
        data={
            "scores": (
                "tom",
                Score(id="some id", total=60),
            )
        },
    )
    updated_score_response = await Score.select(ids=["some id"])

    print("score board:")
    pp.pprint(score_board_response)
    print("\nscores:")
    pp.pprint(scores_response)

    print("\nindirectly updated score board:")
    pp.pprint(updated_score_board_response)
    print("\nindirectly updated score:")
    pp.pprint(updated_score_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 typing import Tuple
from pydantic_redis.asyncio import RedisConfig, Model, Store


class Score(Model):
    _primary_key_field: str = "id"
    id: str
    total: int


class ScoreBoard(Model):
    _primary_key_field: str = "id"
    id: str
    scores: Tuple[str, Score]


async def main():
    pp = pprint.PrettyPrinter(indent=4)
    store = Store(name="test", redis_config=RedisConfig())

    store.register_model(Score)
    store.register_model(ScoreBoard)

    await ScoreBoard.insert(
        data=ScoreBoard(
            id="test",
            scores=(
                "mark",
                Score(id="some id", total=50),
            ),
        )
    )
    score_board_response = await ScoreBoard.select(ids=["test"])
    scores_response = await Score.select(ids=["some id"])

    await Score.update(_id="some id", data={"total": 78})
    updated_score_board_response = await ScoreBoard.select(ids=["test"])

    await ScoreBoard.update(
        _id="test",
        data={
            "scores": (
                "tom",
                Score(id="some id", total=60),
            )
        },
    )
    updated_score_response = await Score.select(ids=["some id"])

    print("score board:")
    pp.pprint(score_board_response)
    print("\nscores:")
    pp.pprint(scores_response)

    print("\nindirectly updated score board:")
    pp.pprint(updated_score_board_response)
    print("\nindirectly updated score:")
    pp.pprint(updated_score_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

Info

The child models will be automatically inserted, or updated if they already exist

Info

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

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


class Score(Model):
    _primary_key_field: str = "id"
    id: str
    total: int


class ScoreBoard(Model):
    _primary_key_field: str = "id"
    id: str
    scores: Tuple[str, Score]


async def main():
    pp = pprint.PrettyPrinter(indent=4)
    store = Store(name="test", redis_config=RedisConfig())

    store.register_model(Score)
    store.register_model(ScoreBoard)

    await ScoreBoard.insert(
        data=ScoreBoard(
            id="test",
            scores=(
                "mark",
                Score(id="some id", total=50),
            ),
        )
    )
    score_board_response = await ScoreBoard.select(ids=["test"])
    scores_response = await Score.select(ids=["some id"])

    await Score.update(_id="some id", data={"total": 78})
    updated_score_board_response = await ScoreBoard.select(ids=["test"])

    await ScoreBoard.update(
        _id="test",
        data={
            "scores": (
                "tom",
                Score(id="some id", total=60),
            )
        },
    )
    updated_score_response = await Score.select(ids=["some id"])

    print("score board:")
    pp.pprint(score_board_response)
    print("\nscores:")
    pp.pprint(scores_response)

    print("\nindirectly updated score board:")
    pp.pprint(updated_score_board_response)
    print("\nindirectly updated score:")
    pp.pprint(updated_score_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 typing import Tuple
from pydantic_redis.asyncio import RedisConfig, Model, Store


class Score(Model):
    _primary_key_field: str = "id"
    id: str
    total: int


class ScoreBoard(Model):
    _primary_key_field: str = "id"
    id: str
    scores: Tuple[str, Score]


async def main():
    pp = pprint.PrettyPrinter(indent=4)
    store = Store(name="test", redis_config=RedisConfig())

    store.register_model(Score)
    store.register_model(ScoreBoard)

    await ScoreBoard.insert(
        data=ScoreBoard(
            id="test",
            scores=(
                "mark",
                Score(id="some id", total=50),
            ),
        )
    )
    score_board_response = await ScoreBoard.select(ids=["test"])
    scores_response = await Score.select(ids=["some id"])

    await Score.update(_id="some id", data={"total": 78})
    updated_score_board_response = await ScoreBoard.select(ids=["test"])

    await ScoreBoard.update(
        _id="test",
        data={
            "scores": (
                "tom",
                Score(id="some id", total=60),
            )
        },
    )
    updated_score_response = await Score.select(ids=["some id"])

    print("score board:")
    pp.pprint(score_board_response)
    print("\nscores:")
    pp.pprint(scores_response)

    print("\nindirectly updated score board:")
    pp.pprint(updated_score_board_response)
    print("\nindirectly updated score:")
    pp.pprint(updated_score_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 tuple with a tuple of instances of the child model

If there is any new instance of the child model that has a pre-existing primary key, it will be updated in redis.

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


class Score(Model):
    _primary_key_field: str = "id"
    id: str
    total: int


class ScoreBoard(Model):
    _primary_key_field: str = "id"
    id: str
    scores: Tuple[str, Score]


async def main():
    pp = pprint.PrettyPrinter(indent=4)
    store = Store(name="test", redis_config=RedisConfig())

    store.register_model(Score)
    store.register_model(ScoreBoard)

    await ScoreBoard.insert(
        data=ScoreBoard(
            id="test",
            scores=(
                "mark",
                Score(id="some id", total=50),
            ),
        )
    )
    score_board_response = await ScoreBoard.select(ids=["test"])
    scores_response = await Score.select(ids=["some id"])

    await Score.update(_id="some id", data={"total": 78})
    updated_score_board_response = await ScoreBoard.select(ids=["test"])

    await ScoreBoard.update(
        _id="test",
        data={
            "scores": (
                "tom",
                Score(id="some id", total=60),
            )
        },
    )
    updated_score_response = await Score.select(ids=["some id"])

    print("score board:")
    pp.pprint(score_board_response)
    print("\nscores:")
    pp.pprint(scores_response)

    print("\nindirectly updated score board:")
    pp.pprint(updated_score_board_response)
    print("\nindirectly updated score:")
    pp.pprint(updated_score_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
score board:
[ScoreBoard(id='test', scores=('mark', Score(id='some id', total=50)))]

scores:
[Score(id='some id', total=50)]

indirectly updated score board:
[ScoreBoard(id='test', scores=('mark', Score(id='some id', total=78)))]

indirectly updated score:
[Score(id='some id', total=60)]

Tuples of 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 Tuple 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