Lists of Nested Models

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

An example is a Folder model that can have child Folder's and File's.

This can easily be pulled off with pydantic-redis.

Import Pydantic-redis' `Model`

First, import pydantic-redis' Model

import pprint
from enum import Enum
from typing import List

from pydantic_redis import Model, Store, RedisConfig


class FileType(Enum):
    TEXT = "text"
    IMAGE = "image"
    EXEC = "executable"


class File(Model):
    _primary_key_field: str = "path"
    path: str
    type: FileType


class Folder(Model):
    _primary_key_field: str = "path"
    path: str
    files: List[File] = []
    folders: List["Folder"] = []


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(File)
    store.register_model(Folder)

    child_folder = Folder(
        path="path/to/child-folder",
        files=[
            File(path="path/to/foo.txt", type=FileType.TEXT),
            File(path="path/to/foo.jpg", type=FileType.IMAGE),
        ],
    )

    Folder.insert(
        Folder(
            path="path/to/parent-folder",
            files=[
                File(path="path/to/bar.txt", type=FileType.TEXT),
                File(path="path/to/bar.jpg", type=FileType.IMAGE),
            ],
            folders=[child_folder],
        )
    )

    parent_folder_response = Folder.select(ids=["path/to/parent-folder"])
    files_response = File.select(
        ids=["path/to/foo.txt", "path/to/foo.jpg", "path/to/bar.txt", "path/to/bar.jpg"]
    )

    Folder.update(
        _id="path/to/child-folder",
        data={"files": [File(path="path/to/foo.txt", type=FileType.EXEC)]},
    )
    updated_parent_folder_response = Folder.select(ids=["path/to/parent-folder"])
    updated_file_response = File.select(ids=["path/to/foo.txt"])

    print("parent folder:")
    pp.pprint(parent_folder_response)
    print("\nfiles:")
    pp.pprint(files_response)

    print("\nindirectly updated parent folder:")
    pp.pprint(updated_parent_folder_response)
    print("\nindirectly updated files:")
    pp.pprint(updated_file_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 enum import Enum
from typing import List

from pydantic_redis import Model, Store, RedisConfig


class FileType(Enum):
    TEXT = "text"
    IMAGE = "image"
    EXEC = "executable"


class File(Model):
    _primary_key_field: str = "path"
    path: str
    type: FileType


class Folder(Model):
    _primary_key_field: str = "path"
    path: str
    files: List[File] = []
    folders: List["Folder"] = []


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(File)
    store.register_model(Folder)

    child_folder = Folder(
        path="path/to/child-folder",
        files=[
            File(path="path/to/foo.txt", type=FileType.TEXT),
            File(path="path/to/foo.jpg", type=FileType.IMAGE),
        ],
    )

    Folder.insert(
        Folder(
            path="path/to/parent-folder",
            files=[
                File(path="path/to/bar.txt", type=FileType.TEXT),
                File(path="path/to/bar.jpg", type=FileType.IMAGE),
            ],
            folders=[child_folder],
        )
    )

    parent_folder_response = Folder.select(ids=["path/to/parent-folder"])
    files_response = File.select(
        ids=["path/to/foo.txt", "path/to/foo.jpg", "path/to/bar.txt", "path/to/bar.jpg"]
    )

    Folder.update(
        _id="path/to/child-folder",
        data={"files": [File(path="path/to/foo.txt", type=FileType.EXEC)]},
    )
    updated_parent_folder_response = Folder.select(ids=["path/to/parent-folder"])
    updated_file_response = File.select(ids=["path/to/foo.txt"])

    print("parent folder:")
    pp.pprint(parent_folder_response)
    print("\nfiles:")
    pp.pprint(files_response)

    print("\nindirectly updated parent folder:")
    pp.pprint(updated_parent_folder_response)
    print("\nindirectly updated files:")
    pp.pprint(updated_file_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 File's with the same path.

import pprint
from enum import Enum
from typing import List

from pydantic_redis import Model, Store, RedisConfig


class FileType(Enum):
    TEXT = "text"
    IMAGE = "image"
    EXEC = "executable"


class File(Model):
    _primary_key_field: str = "path"
    path: str
    type: FileType


class Folder(Model):
    _primary_key_field: str = "path"
    path: str
    files: List[File] = []
    folders: List["Folder"] = []


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(File)
    store.register_model(Folder)

    child_folder = Folder(
        path="path/to/child-folder",
        files=[
            File(path="path/to/foo.txt", type=FileType.TEXT),
            File(path="path/to/foo.jpg", type=FileType.IMAGE),
        ],
    )

    Folder.insert(
        Folder(
            path="path/to/parent-folder",
            files=[
                File(path="path/to/bar.txt", type=FileType.TEXT),
                File(path="path/to/bar.jpg", type=FileType.IMAGE),
            ],
            folders=[child_folder],
        )
    )

    parent_folder_response = Folder.select(ids=["path/to/parent-folder"])
    files_response = File.select(
        ids=["path/to/foo.txt", "path/to/foo.jpg", "path/to/bar.txt", "path/to/bar.jpg"]
    )

    Folder.update(
        _id="path/to/child-folder",
        data={"files": [File(path="path/to/foo.txt", type=FileType.EXEC)]},
    )
    updated_parent_folder_response = Folder.select(ids=["path/to/parent-folder"])
    updated_file_response = File.select(ids=["path/to/foo.txt"])

    print("parent folder:")
    pp.pprint(parent_folder_response)
    print("\nfiles:")
    pp.pprint(files_response)

    print("\nindirectly updated parent folder:")
    pp.pprint(updated_parent_folder_response)
    print("\nindirectly updated files:")
    pp.pprint(updated_file_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 enum import Enum
from typing import List

from pydantic_redis import Model, Store, RedisConfig


class FileType(Enum):
    TEXT = "text"
    IMAGE = "image"
    EXEC = "executable"


class File(Model):
    _primary_key_field: str = "path"
    path: str
    type: FileType


class Folder(Model):
    _primary_key_field: str = "path"
    path: str
    files: List[File] = []
    folders: List["Folder"] = []


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(File)
    store.register_model(Folder)

    child_folder = Folder(
        path="path/to/child-folder",
        files=[
            File(path="path/to/foo.txt", type=FileType.TEXT),
            File(path="path/to/foo.jpg", type=FileType.IMAGE),
        ],
    )

    Folder.insert(
        Folder(
            path="path/to/parent-folder",
            files=[
                File(path="path/to/bar.txt", type=FileType.TEXT),
                File(path="path/to/bar.jpg", type=FileType.IMAGE),
            ],
            folders=[child_folder],
        )
    )

    parent_folder_response = Folder.select(ids=["path/to/parent-folder"])
    files_response = File.select(
        ids=["path/to/foo.txt", "path/to/foo.jpg", "path/to/bar.txt", "path/to/bar.jpg"]
    )

    Folder.update(
        _id="path/to/child-folder",
        data={"files": [File(path="path/to/foo.txt", type=FileType.EXEC)]},
    )
    updated_parent_folder_response = Folder.select(ids=["path/to/parent-folder"])
    updated_file_response = File.select(ids=["path/to/foo.txt"])

    print("parent folder:")
    pp.pprint(parent_folder_response)
    print("\nfiles:")
    pp.pprint(files_response)

    print("\nindirectly updated parent folder:")
    pp.pprint(updated_parent_folder_response)
    print("\nindirectly updated files:")
    pp.pprint(updated_file_response)

Add the Nested Model List to the Parent Model

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

Example

In this case, the field files is annotated with List[File].

And the field folders is annotated with "Folder" class i.e. itself.

import pprint
from enum import Enum
from typing import List

from pydantic_redis import Model, Store, RedisConfig


class FileType(Enum):
    TEXT = "text"
    IMAGE = "image"
    EXEC = "executable"


class File(Model):
    _primary_key_field: str = "path"
    path: str
    type: FileType


class Folder(Model):
    _primary_key_field: str = "path"
    path: str
    files: List[File] = []
    folders: List["Folder"] = []


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(File)
    store.register_model(Folder)

    child_folder = Folder(
        path="path/to/child-folder",
        files=[
            File(path="path/to/foo.txt", type=FileType.TEXT),
            File(path="path/to/foo.jpg", type=FileType.IMAGE),
        ],
    )

    Folder.insert(
        Folder(
            path="path/to/parent-folder",
            files=[
                File(path="path/to/bar.txt", type=FileType.TEXT),
                File(path="path/to/bar.jpg", type=FileType.IMAGE),
            ],
            folders=[child_folder],
        )
    )

    parent_folder_response = Folder.select(ids=["path/to/parent-folder"])
    files_response = File.select(
        ids=["path/to/foo.txt", "path/to/foo.jpg", "path/to/bar.txt", "path/to/bar.jpg"]
    )

    Folder.update(
        _id="path/to/child-folder",
        data={"files": [File(path="path/to/foo.txt", type=FileType.EXEC)]},
    )
    updated_parent_folder_response = Folder.select(ids=["path/to/parent-folder"])
    updated_file_response = File.select(ids=["path/to/foo.txt"])

    print("parent folder:")
    pp.pprint(parent_folder_response)
    print("\nfiles:")
    pp.pprint(files_response)

    print("\nindirectly updated parent folder:")
    pp.pprint(updated_parent_folder_response)
    print("\nindirectly updated files:")
    pp.pprint(updated_file_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 Folder's with the same path.

import pprint
from enum import Enum
from typing import List

from pydantic_redis import Model, Store, RedisConfig


class FileType(Enum):
    TEXT = "text"
    IMAGE = "image"
    EXEC = "executable"


class File(Model):
    _primary_key_field: str = "path"
    path: str
    type: FileType


class Folder(Model):
    _primary_key_field: str = "path"
    path: str
    files: List[File] = []
    folders: List["Folder"] = []


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(File)
    store.register_model(Folder)

    child_folder = Folder(
        path="path/to/child-folder",
        files=[
            File(path="path/to/foo.txt", type=FileType.TEXT),
            File(path="path/to/foo.jpg", type=FileType.IMAGE),
        ],
    )

    Folder.insert(
        Folder(
            path="path/to/parent-folder",
            files=[
                File(path="path/to/bar.txt", type=FileType.TEXT),
                File(path="path/to/bar.jpg", type=FileType.IMAGE),
            ],
            folders=[child_folder],
        )
    )

    parent_folder_response = Folder.select(ids=["path/to/parent-folder"])
    files_response = File.select(
        ids=["path/to/foo.txt", "path/to/foo.jpg", "path/to/bar.txt", "path/to/bar.jpg"]
    )

    Folder.update(
        _id="path/to/child-folder",
        data={"files": [File(path="path/to/foo.txt", type=FileType.EXEC)]},
    )
    updated_parent_folder_response = Folder.select(ids=["path/to/parent-folder"])
    updated_file_response = File.select(ids=["path/to/foo.txt"])

    print("parent folder:")
    pp.pprint(parent_folder_response)
    print("\nfiles:")
    pp.pprint(files_response)

    print("\nindirectly updated parent folder:")
    pp.pprint(updated_parent_folder_response)
    print("\nindirectly updated files:")
    pp.pprint(updated_file_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 enum import Enum
from typing import List

from pydantic_redis import Model, Store, RedisConfig


class FileType(Enum):
    TEXT = "text"
    IMAGE = "image"
    EXEC = "executable"


class File(Model):
    _primary_key_field: str = "path"
    path: str
    type: FileType


class Folder(Model):
    _primary_key_field: str = "path"
    path: str
    files: List[File] = []
    folders: List["Folder"] = []


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(File)
    store.register_model(Folder)

    child_folder = Folder(
        path="path/to/child-folder",
        files=[
            File(path="path/to/foo.txt", type=FileType.TEXT),
            File(path="path/to/foo.jpg", type=FileType.IMAGE),
        ],
    )

    Folder.insert(
        Folder(
            path="path/to/parent-folder",
            files=[
                File(path="path/to/bar.txt", type=FileType.TEXT),
                File(path="path/to/bar.jpg", type=FileType.IMAGE),
            ],
            folders=[child_folder],
        )
    )

    parent_folder_response = Folder.select(ids=["path/to/parent-folder"])
    files_response = File.select(
        ids=["path/to/foo.txt", "path/to/foo.jpg", "path/to/bar.txt", "path/to/bar.jpg"]
    )

    Folder.update(
        _id="path/to/child-folder",
        data={"files": [File(path="path/to/foo.txt", type=FileType.EXEC)]},
    )
    updated_parent_folder_response = Folder.select(ids=["path/to/parent-folder"])
    updated_file_response = File.select(ids=["path/to/foo.txt"])

    print("parent folder:")
    pp.pprint(parent_folder_response)
    print("\nfiles:")
    pp.pprint(files_response)

    print("\nindirectly updated parent folder:")
    pp.pprint(updated_parent_folder_response)
    print("\nindirectly updated files:")
    pp.pprint(updated_file_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 enum import Enum
from typing import List

from pydantic_redis import Model, Store, RedisConfig


class FileType(Enum):
    TEXT = "text"
    IMAGE = "image"
    EXEC = "executable"


class File(Model):
    _primary_key_field: str = "path"
    path: str
    type: FileType


class Folder(Model):
    _primary_key_field: str = "path"
    path: str
    files: List[File] = []
    folders: List["Folder"] = []


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(File)
    store.register_model(Folder)

    child_folder = Folder(
        path="path/to/child-folder",
        files=[
            File(path="path/to/foo.txt", type=FileType.TEXT),
            File(path="path/to/foo.jpg", type=FileType.IMAGE),
        ],
    )

    Folder.insert(
        Folder(
            path="path/to/parent-folder",
            files=[
                File(path="path/to/bar.txt", type=FileType.TEXT),
                File(path="path/to/bar.jpg", type=FileType.IMAGE),
            ],
            folders=[child_folder],
        )
    )

    parent_folder_response = Folder.select(ids=["path/to/parent-folder"])
    files_response = File.select(
        ids=["path/to/foo.txt", "path/to/foo.jpg", "path/to/bar.txt", "path/to/bar.jpg"]
    )

    Folder.update(
        _id="path/to/child-folder",
        data={"files": [File(path="path/to/foo.txt", type=FileType.EXEC)]},
    )
    updated_parent_folder_response = Folder.select(ids=["path/to/parent-folder"])
    updated_file_response = File.select(ids=["path/to/foo.txt"])

    print("parent folder:")
    pp.pprint(parent_folder_response)
    print("\nfiles:")
    pp.pprint(files_response)

    print("\nindirectly updated parent folder:")
    pp.pprint(updated_parent_folder_response)
    print("\nindirectly updated files:")
    pp.pprint(updated_file_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 enum import Enum
from typing import List

from pydantic_redis import Model, Store, RedisConfig


class FileType(Enum):
    TEXT = "text"
    IMAGE = "image"
    EXEC = "executable"


class File(Model):
    _primary_key_field: str = "path"
    path: str
    type: FileType


class Folder(Model):
    _primary_key_field: str = "path"
    path: str
    files: List[File] = []
    folders: List["Folder"] = []


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(File)
    store.register_model(Folder)

    child_folder = Folder(
        path="path/to/child-folder",
        files=[
            File(path="path/to/foo.txt", type=FileType.TEXT),
            File(path="path/to/foo.jpg", type=FileType.IMAGE),
        ],
    )

    Folder.insert(
        Folder(
            path="path/to/parent-folder",
            files=[
                File(path="path/to/bar.txt", type=FileType.TEXT),
                File(path="path/to/bar.jpg", type=FileType.IMAGE),
            ],
            folders=[child_folder],
        )
    )

    parent_folder_response = Folder.select(ids=["path/to/parent-folder"])
    files_response = File.select(
        ids=["path/to/foo.txt", "path/to/foo.jpg", "path/to/bar.txt", "path/to/bar.jpg"]
    )

    Folder.update(
        _id="path/to/child-folder",
        data={"files": [File(path="path/to/foo.txt", type=FileType.EXEC)]},
    )
    updated_parent_folder_response = Folder.select(ids=["path/to/parent-folder"])
    updated_file_response = File.select(ids=["path/to/foo.txt"])

    print("parent folder:")
    pp.pprint(parent_folder_response)
    print("\nfiles:")
    pp.pprint(files_response)

    print("\nindirectly updated parent folder:")
    pp.pprint(updated_parent_folder_response)
    print("\nindirectly updated files:")
    pp.pprint(updated_file_response)

Indirectly Update Child Model

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

Set the attribute containing the child model list with a list 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 enum import Enum
from typing import List

from pydantic_redis import Model, Store, RedisConfig


class FileType(Enum):
    TEXT = "text"
    IMAGE = "image"
    EXEC = "executable"


class File(Model):
    _primary_key_field: str = "path"
    path: str
    type: FileType


class Folder(Model):
    _primary_key_field: str = "path"
    path: str
    files: List[File] = []
    folders: List["Folder"] = []


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(File)
    store.register_model(Folder)

    child_folder = Folder(
        path="path/to/child-folder",
        files=[
            File(path="path/to/foo.txt", type=FileType.TEXT),
            File(path="path/to/foo.jpg", type=FileType.IMAGE),
        ],
    )

    Folder.insert(
        Folder(
            path="path/to/parent-folder",
            files=[
                File(path="path/to/bar.txt", type=FileType.TEXT),
                File(path="path/to/bar.jpg", type=FileType.IMAGE),
            ],
            folders=[child_folder],
        )
    )

    parent_folder_response = Folder.select(ids=["path/to/parent-folder"])
    files_response = File.select(
        ids=["path/to/foo.txt", "path/to/foo.jpg", "path/to/bar.txt", "path/to/bar.jpg"]
    )

    Folder.update(
        _id="path/to/child-folder",
        data={"files": [File(path="path/to/foo.txt", type=FileType.EXEC)]},
    )
    updated_parent_folder_response = Folder.select(ids=["path/to/parent-folder"])
    updated_file_response = File.select(ids=["path/to/foo.txt"])

    print("parent folder:")
    pp.pprint(parent_folder_response)
    print("\nfiles:")
    pp.pprint(files_response)

    print("\nindirectly updated parent folder:")
    pp.pprint(updated_parent_folder_response)
    print("\nindirectly updated files:")
    pp.pprint(updated_file_response)

Run the App

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

Tip

Probably FLUSHALL redis first

$ python main.py
parent folder:
[   Folder(path='path/to/parent-folder', files=[File(path='path/to/bar.txt', type=<FileType.TEXT: 'text'>), File(path='path/to/bar.jpg', type=<FileType.IMAGE: 'image'>)], folders=[Folder(path='path/to/child-folder', files=[File(path='path/to/foo.txt', type=<FileType.TEXT: 'text'>), File(path='path/to/foo.jpg', type=<FileType.IMAGE: 'image'>)], folders=[])])]

files:
[   File(path='path/to/foo.txt', type=<FileType.TEXT: 'text'>),
    File(path='path/to/foo.jpg', type=<FileType.IMAGE: 'image'>),
    File(path='path/to/bar.txt', type=<FileType.TEXT: 'text'>),
    File(path='path/to/bar.jpg', type=<FileType.IMAGE: 'image'>)]

indirectly updated parent folder:
[   Folder(path='path/to/parent-folder', files=[File(path='path/to/bar.txt', type=<FileType.TEXT: 'text'>), File(path='path/to/bar.jpg', type=<FileType.IMAGE: 'image'>)], folders=[Folder(path='path/to/child-folder', files=[File(path='path/to/foo.txt', type=<FileType.EXEC: 'executable'>)], folders=[])])]

indirectly updated files:
[File(path='path/to/foo.txt', type=<FileType.EXEC: 'executable'>)]

Lists 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 List 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