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

import pprint
from pydantic_redis import Model, Store, RedisConfig


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


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

    store.register_model(Book)

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

    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 = Book.select()

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

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

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 pprint
from pydantic_redis import Model, Store, RedisConfig


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


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

    store.register_model(Book)

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

    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 = Book.select()

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

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

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 pprint
from pydantic_redis import Model, Store, RedisConfig


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


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

    store.register_model(Book)

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

    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 = Book.select()

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

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

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 pprint
from pydantic_redis import Model, Store, RedisConfig


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


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

    store.register_model(Book)

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

    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 = Book.select()

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

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

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