Skip to content

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.

These models are derived from pydantic's BaseModel.

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

from pydantic_redis.asyncio import Model, Store, RedisConfig


class Book(Model):
    _primary_key_field: str = "title"
    title: str
    author: str
    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(Book)

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

    response = await Book.select(ids=["Oliver Twist"])
    pp.pprint(response)


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

Create the 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

from pydantic_redis.asyncio import Model, Store, RedisConfig


class Book(Model):
    _primary_key_field: str = "title"
    title: str
    author: str
    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(Book)

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

    response = await Book.select(ids=["Oliver Twist"])
    pp.pprint(response)


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

Specify the _primary_key_field Attribute

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 books with the same title.

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

from pydantic_redis.asyncio import Model, Store, RedisConfig


class Book(Model):
    _primary_key_field: str = "title"
    title: str
    author: str
    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(Book)

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

    response = await Book.select(ids=["Oliver Twist"])
    pp.pprint(response)


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

Register the Model in the Store

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

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

from pydantic_redis.asyncio import Model, Store, RedisConfig


class Book(Model):
    _primary_key_field: str = "title"
    title: str
    author: str
    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(Book)

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

    response = await Book.select(ids=["Oliver Twist"])
    pp.pprint(response)


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

Use the Model

Then you can use the model class to:

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

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

from pydantic_redis.asyncio import Model, Store, RedisConfig


class Book(Model):
    _primary_key_field: str = "title"
    title: str
    author: str
    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(Book)

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

    response = await Book.select(ids=["Oliver Twist"])
    pp.pprint(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(title='Oliver Twist', author='Charles Dickens', rating=2.0, published_on=datetime.date(1215, 4, 4), tags=['Classic'], in_stock=False)]