Update

Pydantic-redis can be used to update model instances in redis.

Create and register the Model

A model is a class that inherits from Model with its _primary_key_field attribute set.

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

Warning

The imports are from pydantic_redis.asyncio NOT pydantic_redis

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


class Book(Model):
    _primary_key_field: str = "title"
    title: str
    author: str


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

    store.register_model(Book)

    await Book.insert(
        [
            Book(title="Oliver Twist", author="Charles Dickens"),
            Book(title="Jane Eyre", author="Emily Bronte"),
            Book(title="Pride and Prejudice", author="Jane Austen"),
        ]
    )
    await Book.update(_id="Oliver Twist", data={"author": "Charlie Ickens"})
    await Book.update(
        _id="Jane Eyre", data={"author": "Daniel McKenzie"}, life_span_seconds=1800
    )
    single_update_response = await Book.select()

    await Book.insert(
        [
            Book(title="Oliver Twist", author="Chuck Dickens"),
            Book(title="Jane Eyre", author="Emiliano Bronte"),
            Book(title="Pride and Prejudice", author="Janey Austen"),
        ],
        life_span_seconds=3600,
    )
    multi_update_response = await Book.select()

    print("single update:")
    pp.pprint(single_update_response)

    print("\nmulti update:")
    pp.pprint(multi_update_response)


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

Update One Record

To update a single record in redis, pass the primary key (_id) of that record and the new changes to the model's update method

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


class Book(Model):
    _primary_key_field: str = "title"
    title: str
    author: str


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

    store.register_model(Book)

    await Book.insert(
        [
            Book(title="Oliver Twist", author="Charles Dickens"),
            Book(title="Jane Eyre", author="Emily Bronte"),
            Book(title="Pride and Prejudice", author="Jane Austen"),
        ]
    )
    await Book.update(_id="Oliver Twist", data={"author": "Charlie Ickens"})
    await Book.update(
        _id="Jane Eyre", data={"author": "Daniel McKenzie"}, life_span_seconds=1800
    )
    single_update_response = await Book.select()

    await Book.insert(
        [
            Book(title="Oliver Twist", author="Chuck Dickens"),
            Book(title="Jane Eyre", author="Emiliano Bronte"),
            Book(title="Pride and Prejudice", author="Janey Austen"),
        ],
        life_span_seconds=3600,
    )
    multi_update_response = await Book.select()

    print("single update:")
    pp.pprint(single_update_response)

    print("\nmulti update:")
    pp.pprint(multi_update_response)


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

Update One Record With TTL

To update the record's time-to-live (TTL) also, pass the life_span_seconds argument to the model's update method.

Info

When the life_span_seconds argument is not specified, the life_span_in_seconds passed to the store during initialization is used.

The life_span_in_seconds in both cases is None by default. This means records never expire by default.

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


class Book(Model):
    _primary_key_field: str = "title"
    title: str
    author: str


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

    store.register_model(Book)

    await Book.insert(
        [
            Book(title="Oliver Twist", author="Charles Dickens"),
            Book(title="Jane Eyre", author="Emily Bronte"),
            Book(title="Pride and Prejudice", author="Jane Austen"),
        ]
    )
    await Book.update(_id="Oliver Twist", data={"author": "Charlie Ickens"})
    await Book.update(
        _id="Jane Eyre", data={"author": "Daniel McKenzie"}, life_span_seconds=1800
    )
    single_update_response = await Book.select()

    await Book.insert(
        [
            Book(title="Oliver Twist", author="Chuck Dickens"),
            Book(title="Jane Eyre", author="Emiliano Bronte"),
            Book(title="Pride and Prejudice", author="Janey Austen"),
        ],
        life_span_seconds=3600,
    )
    multi_update_response = await Book.select()

    print("single update:")
    pp.pprint(single_update_response)

    print("\nmulti update:")
    pp.pprint(multi_update_response)


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

Update/Upsert Many Records

To update many records in redis, pass a list of that model's instances as first argument to the model's insert method.

Technically, this will insert any records that don't exist and overwrite any that exist already.

Info

Updating many records at once is more performant than adding one record at a time repeatedly because less network requests are made in the former.

Warning

Calling insert always overwrites the time-to-live of the records updated.

When the life_span_seconds argument is not specified, the life_span_in_seconds passed to the store during initialization is used.

By default life_span_seconds is None i.e. the time-to-live is removed and the updated records never expire.

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


class Book(Model):
    _primary_key_field: str = "title"
    title: str
    author: str


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

    store.register_model(Book)

    await Book.insert(
        [
            Book(title="Oliver Twist", author="Charles Dickens"),
            Book(title="Jane Eyre", author="Emily Bronte"),
            Book(title="Pride and Prejudice", author="Jane Austen"),
        ]
    )
    await Book.update(_id="Oliver Twist", data={"author": "Charlie Ickens"})
    await Book.update(
        _id="Jane Eyre", data={"author": "Daniel McKenzie"}, life_span_seconds=1800
    )
    single_update_response = await Book.select()

    await Book.insert(
        [
            Book(title="Oliver Twist", author="Chuck Dickens"),
            Book(title="Jane Eyre", author="Emiliano Bronte"),
            Book(title="Pride and Prejudice", author="Janey Austen"),
        ],
        life_span_seconds=3600,
    )
    multi_update_response = await Book.select()

    print("single update:")
    pp.pprint(single_update_response)

    print("\nmulti update:")
    pp.pprint(multi_update_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
single update:
[   Book(title='Jane Eyre', author='Daniel McKenzie'),
    Book(title='Oliver Twist', author='Charlie Ickens'),
    Book(title='Pride and Prejudice', author='Jane Austen')]

multi update:
[   Book(title='Jane Eyre', author='Emiliano Bronte'),
    Book(title='Oliver Twist', author='Chuck Dickens'),
    Book(title='Pride and Prejudice', author='Janey Austen')]