Insert

Pydantic-redis can be used to insert new model instances into 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"))
    await Book.insert(
        Book(title="Great Expectations", author="Charles Dickens"),
        life_span_seconds=1800,
    )
    await Book.insert(
        [
            Book(title="Jane Eyre", author="Emily Bronte"),
            Book(title="Pride and Prejudice", author="Jane Austen"),
        ]
    )
    await Book.insert(
        [
            Book(title="Jane Eyre", author="Emily Bronte"),
            Book(title="Pride and Prejudice", author="Jane Austen"),
        ],
        life_span_seconds=3600,
    )

    response = await Book.select()
    pp.pprint(response)


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

Insert One Record

To add a single record to the redis instance, pass that model's instance as first argument to the model's insert 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"))
    await Book.insert(
        Book(title="Great Expectations", author="Charles Dickens"),
        life_span_seconds=1800,
    )
    await Book.insert(
        [
            Book(title="Jane Eyre", author="Emily Bronte"),
            Book(title="Pride and Prejudice", author="Jane Austen"),
        ]
    )
    await Book.insert(
        [
            Book(title="Jane Eyre", author="Emily Bronte"),
            Book(title="Pride and Prejudice", author="Jane Austen"),
        ],
        life_span_seconds=3600,
    )

    response = await Book.select()
    pp.pprint(response)


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

Insert One Record With TTL

To make the record added to redis temporary, add a life_span_seconds (Time To Live i.e. TTL) key-word argument when calling the model's insert 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"))
    await Book.insert(
        Book(title="Great Expectations", author="Charles Dickens"),
        life_span_seconds=1800,
    )
    await Book.insert(
        [
            Book(title="Jane Eyre", author="Emily Bronte"),
            Book(title="Pride and Prejudice", author="Jane Austen"),
        ]
    )
    await Book.insert(
        [
            Book(title="Jane Eyre", author="Emily Bronte"),
            Book(title="Pride and Prejudice", author="Jane Austen"),
        ],
        life_span_seconds=3600,
    )

    response = await Book.select()
    pp.pprint(response)


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

Insert Many Records

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

Info

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

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"))
    await Book.insert(
        Book(title="Great Expectations", author="Charles Dickens"),
        life_span_seconds=1800,
    )
    await Book.insert(
        [
            Book(title="Jane Eyre", author="Emily Bronte"),
            Book(title="Pride and Prejudice", author="Jane Austen"),
        ]
    )
    await Book.insert(
        [
            Book(title="Jane Eyre", author="Emily Bronte"),
            Book(title="Pride and Prejudice", author="Jane Austen"),
        ],
        life_span_seconds=3600,
    )

    response = await Book.select()
    pp.pprint(response)


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

Insert Many Records With TTL

To add temporary records to redis, add a life_span_seconds (Time To Live i.e. TTL) argument when calling the model's insert 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"))
    await Book.insert(
        Book(title="Great Expectations", author="Charles Dickens"),
        life_span_seconds=1800,
    )
    await Book.insert(
        [
            Book(title="Jane Eyre", author="Emily Bronte"),
            Book(title="Pride and Prejudice", author="Jane Austen"),
        ]
    )
    await Book.insert(
        [
            Book(title="Jane Eyre", author="Emily Bronte"),
            Book(title="Pride and Prejudice", author="Jane Austen"),
        ],
        life_span_seconds=3600,
    )

    response = await Book.select()
    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='Jane Eyre', author='Emily Bronte'),
    Book(title='Great Expectations', author='Charles Dickens'),
    Book(title='Oliver Twist', author='Charles Dickens'),
    Book(title='Pride and Prejudice', author='Jane Austen')]