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' Model

import pprint
from typing import Tuple
from pydantic_redis 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]


if __name__ == "__main__":
    pp = pprint.PrettyPrinter(indent=4)
    store = Store(name="test", redis_config=RedisConfig())

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

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

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

    ScoreBoard.update(
        _id="test",
        data={
            "scores": (
                "tom",
                Score(id="some id", total=60),
            )
        },
    )
    updated_score_response = 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)

Create the Child Model

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

Use standard Python types for all attributes.

import pprint
from typing import Tuple
from pydantic_redis 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]


if __name__ == "__main__":
    pp = pprint.PrettyPrinter(indent=4)
    store = Store(name="test", redis_config=RedisConfig())

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

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

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

    ScoreBoard.update(
        _id="test",
        data={
            "scores": (
                "tom",
                Score(id="some id", total=60),
            )
        },
    )
    updated_score_response = 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)

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 pprint
from typing import Tuple
from pydantic_redis 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]


if __name__ == "__main__":
    pp = pprint.PrettyPrinter(indent=4)
    store = Store(name="test", redis_config=RedisConfig())

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

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

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

    ScoreBoard.update(
        _id="test",
        data={
            "scores": (
                "tom",
                Score(id="some id", total=60),
            )
        },
    )
    updated_score_response = 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)

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 pprint
from typing import Tuple
from pydantic_redis 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]


if __name__ == "__main__":
    pp = pprint.PrettyPrinter(indent=4)
    store = Store(name="test", redis_config=RedisConfig())

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

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

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

    ScoreBoard.update(
        _id="test",
        data={
            "scores": (
                "tom",
                Score(id="some id", total=60),
            )
        },
    )
    updated_score_response = 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)

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 pprint
from typing import Tuple
from pydantic_redis 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]


if __name__ == "__main__":
    pp = pprint.PrettyPrinter(indent=4)
    store = Store(name="test", redis_config=RedisConfig())

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

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

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

    ScoreBoard.update(
        _id="test",
        data={
            "scores": (
                "tom",
                Score(id="some id", total=60),
            )
        },
    )
    updated_score_response = 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)

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 pprint
from typing import Tuple
from pydantic_redis 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]


if __name__ == "__main__":
    pp = pprint.PrettyPrinter(indent=4)
    store = Store(name="test", redis_config=RedisConfig())

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

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

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

    ScoreBoard.update(
        _id="test",
        data={
            "scores": (
                "tom",
                Score(id="some id", total=60),
            )
        },
    )
    updated_score_response = 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)

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 pprint
from typing import Tuple
from pydantic_redis 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]


if __name__ == "__main__":
    pp = pprint.PrettyPrinter(indent=4)
    store = Store(name="test", redis_config=RedisConfig())

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

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

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

    ScoreBoard.update(
        _id="test",
        data={
            "scores": (
                "tom",
                Score(id="some id", total=60),
            )
        },
    )
    updated_score_response = 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)

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 pprint
from typing import Tuple
from pydantic_redis 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]


if __name__ == "__main__":
    pp = pprint.PrettyPrinter(indent=4)
    store = Store(name="test", redis_config=RedisConfig())

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

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

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

    ScoreBoard.update(
        _id="test",
        data={
            "scores": (
                "tom",
                Score(id="some id", total=60),
            )
        },
    )
    updated_score_response = 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)

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 pprint
from typing import Tuple
from pydantic_redis 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]


if __name__ == "__main__":
    pp = pprint.PrettyPrinter(indent=4)
    store = Store(name="test", redis_config=RedisConfig())

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

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

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

    ScoreBoard.update(
        _id="test",
        data={
            "scores": (
                "tom",
                Score(id="some id", total=60),
            )
        },
    )
    updated_score_response = 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)

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 pprint
from typing import Tuple
from pydantic_redis 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]


if __name__ == "__main__":
    pp = pprint.PrettyPrinter(indent=4)
    store = Store(name="test", redis_config=RedisConfig())

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

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

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

    ScoreBoard.update(
        _id="test",
        data={
            "scores": (
                "tom",
                Score(id="some id", total=60),
            )
        },
    )
    updated_score_response = 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)

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)]