Skip to content

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.asyncio's Model

Warning

The imports are from pydantic_redis.asyncio NOT pydantic_redis

import asyncio
import pprint
from enum import Enum
from typing import List

from pydantic_redis.asyncio 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"] = []


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

    await 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 = await Folder.select(ids=["path/to/parent-folder"])
    files_response = await File.select(
        ids=["path/to/foo.txt", "path/to/foo.jpg", "path/to/bar.txt", "path/to/bar.jpg"]
    )

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


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 enum import Enum
from typing import List

from pydantic_redis.asyncio 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"] = []


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

    await 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 = await Folder.select(ids=["path/to/parent-folder"])
    files_response = await File.select(
        ids=["path/to/foo.txt", "path/to/foo.jpg", "path/to/bar.txt", "path/to/bar.jpg"]
    )

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


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 File's with the same path.

import asyncio
import pprint
from enum import Enum
from typing import List

from pydantic_redis.asyncio 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"] = []


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

    await 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 = await Folder.select(ids=["path/to/parent-folder"])
    files_response = await File.select(
        ids=["path/to/foo.txt", "path/to/foo.jpg", "path/to/bar.txt", "path/to/bar.jpg"]
    )

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


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 enum import Enum
from typing import List

from pydantic_redis.asyncio 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"] = []


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

    await 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 = await Folder.select(ids=["path/to/parent-folder"])
    files_response = await File.select(
        ids=["path/to/foo.txt", "path/to/foo.jpg", "path/to/bar.txt", "path/to/bar.jpg"]
    )

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


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

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 asyncio
import pprint
from enum import Enum
from typing import List

from pydantic_redis.asyncio 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"] = []


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

    await 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 = await Folder.select(ids=["path/to/parent-folder"])
    files_response = await File.select(
        ids=["path/to/foo.txt", "path/to/foo.jpg", "path/to/bar.txt", "path/to/bar.jpg"]
    )

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


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 Folder's with the same path.

import asyncio
import pprint
from enum import Enum
from typing import List

from pydantic_redis.asyncio 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"] = []


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

    await 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 = await Folder.select(ids=["path/to/parent-folder"])
    files_response = await File.select(
        ids=["path/to/foo.txt", "path/to/foo.jpg", "path/to/bar.txt", "path/to/bar.jpg"]
    )

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


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 enum import Enum
from typing import List

from pydantic_redis.asyncio 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"] = []


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

    await 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 = await Folder.select(ids=["path/to/parent-folder"])
    files_response = await File.select(
        ids=["path/to/foo.txt", "path/to/foo.jpg", "path/to/bar.txt", "path/to/bar.jpg"]
    )

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


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 enum import Enum
from typing import List

from pydantic_redis.asyncio 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"] = []


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

    await 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 = await Folder.select(ids=["path/to/parent-folder"])
    files_response = await File.select(
        ids=["path/to/foo.txt", "path/to/foo.jpg", "path/to/bar.txt", "path/to/bar.jpg"]
    )

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


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 enum import Enum
from typing import List

from pydantic_redis.asyncio 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"] = []


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

    await 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 = await Folder.select(ids=["path/to/parent-folder"])
    files_response = await File.select(
        ids=["path/to/foo.txt", "path/to/foo.jpg", "path/to/bar.txt", "path/to/bar.jpg"]
    )

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


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 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 asyncio
import pprint
from enum import Enum
from typing import List

from pydantic_redis.asyncio 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"] = []


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

    await 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 = await Folder.select(ids=["path/to/parent-folder"])
    files_response = await File.select(
        ids=["path/to/foo.txt", "path/to/foo.jpg", "path/to/bar.txt", "path/to/bar.jpg"]
    )

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


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