Select

Pydantic-redis can be used to retrieve model instances from 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"),
            Book(title="Utah Blaine", author="Louis L'Amour"),
        ]
    )

    select_all_response = await Book.select()
    select_by_id_response = await Book.select(
        ids=["Oliver Twist", "Pride and Prejudice"]
    )

    select_some_fields_response = await Book.select(columns=["author"])
    select_some_fields_for_ids_response = await Book.select(
        ids=["Oliver Twist", "Pride and Prejudice"], columns=["author"]
    )

    paginated_select_all_response = await Book.select(skip=0, limit=2)
    paginated_select_some_fields_response = await Book.select(
        columns=["author"], skip=2, limit=2
    )

    print("all:")
    pp.pprint(select_all_response)
    print("\nby id:")
    pp.pprint(select_by_id_response)
    print("\nsome fields for all:")
    pp.pprint(select_some_fields_response)
    print("\nsome fields for given ids:")
    pp.pprint(select_some_fields_for_ids_response)
    print("\npaginated; skip: 0, limit: 2:")
    pp.pprint(paginated_select_all_response)
    print("\npaginated returning some fields for each; skip: 2, limit: 2:")
    pp.pprint(paginated_select_some_fields_response)


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

Select All Records

To select all records for the given model in redis, call the model's select method without any arguments.

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"),
            Book(title="Utah Blaine", author="Louis L'Amour"),
        ]
    )

    select_all_response = await Book.select()
    select_by_id_response = await Book.select(
        ids=["Oliver Twist", "Pride and Prejudice"]
    )

    select_some_fields_response = await Book.select(columns=["author"])
    select_some_fields_for_ids_response = await Book.select(
        ids=["Oliver Twist", "Pride and Prejudice"], columns=["author"]
    )

    paginated_select_all_response = await Book.select(skip=0, limit=2)
    paginated_select_some_fields_response = await Book.select(
        columns=["author"], skip=2, limit=2
    )

    print("all:")
    pp.pprint(select_all_response)
    print("\nby id:")
    pp.pprint(select_by_id_response)
    print("\nsome fields for all:")
    pp.pprint(select_some_fields_response)
    print("\nsome fields for given ids:")
    pp.pprint(select_some_fields_for_ids_response)
    print("\npaginated; skip: 0, limit: 2:")
    pp.pprint(paginated_select_all_response)
    print("\npaginated returning some fields for each; skip: 2, limit: 2:")
    pp.pprint(paginated_select_some_fields_response)


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

Select Some Fields for All Records

To select some fields for all records for the given model in redis, pass the desired fields (columns) to the model's select method.

Info

This returns dictionaries instead of Model instances.

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"),
            Book(title="Utah Blaine", author="Louis L'Amour"),
        ]
    )

    select_all_response = await Book.select()
    select_by_id_response = await Book.select(
        ids=["Oliver Twist", "Pride and Prejudice"]
    )

    select_some_fields_response = await Book.select(columns=["author"])
    select_some_fields_for_ids_response = await Book.select(
        ids=["Oliver Twist", "Pride and Prejudice"], columns=["author"]
    )

    paginated_select_all_response = await Book.select(skip=0, limit=2)
    paginated_select_some_fields_response = await Book.select(
        columns=["author"], skip=2, limit=2
    )

    print("all:")
    pp.pprint(select_all_response)
    print("\nby id:")
    pp.pprint(select_by_id_response)
    print("\nsome fields for all:")
    pp.pprint(select_some_fields_response)
    print("\nsome fields for given ids:")
    pp.pprint(select_some_fields_for_ids_response)
    print("\npaginated; skip: 0, limit: 2:")
    pp.pprint(paginated_select_all_response)
    print("\npaginated returning some fields for each; skip: 2, limit: 2:")
    pp.pprint(paginated_select_some_fields_response)


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

Select Some Records

To select some records for the given model in redis, pass a list of the primary keys (ids) of the desired records to the model's select 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"),
            Book(title="Utah Blaine", author="Louis L'Amour"),
        ]
    )

    select_all_response = await Book.select()
    select_by_id_response = await Book.select(
        ids=["Oliver Twist", "Pride and Prejudice"]
    )

    select_some_fields_response = await Book.select(columns=["author"])
    select_some_fields_for_ids_response = await Book.select(
        ids=["Oliver Twist", "Pride and Prejudice"], columns=["author"]
    )

    paginated_select_all_response = await Book.select(skip=0, limit=2)
    paginated_select_some_fields_response = await Book.select(
        columns=["author"], skip=2, limit=2
    )

    print("all:")
    pp.pprint(select_all_response)
    print("\nby id:")
    pp.pprint(select_by_id_response)
    print("\nsome fields for all:")
    pp.pprint(select_some_fields_response)
    print("\nsome fields for given ids:")
    pp.pprint(select_some_fields_for_ids_response)
    print("\npaginated; skip: 0, limit: 2:")
    pp.pprint(paginated_select_all_response)
    print("\npaginated returning some fields for each; skip: 2, limit: 2:")
    pp.pprint(paginated_select_some_fields_response)


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

Select Some Fields for Some Records

We can go further and limit the fields returned for the desired records.

We pass the desired fields (columns) to the model's select method, together with the list of the primary keys (ids) of the desired records.

Info

This returns dictionaries instead of Model instances.

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"),
            Book(title="Utah Blaine", author="Louis L'Amour"),
        ]
    )

    select_all_response = await Book.select()
    select_by_id_response = await Book.select(
        ids=["Oliver Twist", "Pride and Prejudice"]
    )

    select_some_fields_response = await Book.select(columns=["author"])
    select_some_fields_for_ids_response = await Book.select(
        ids=["Oliver Twist", "Pride and Prejudice"], columns=["author"]
    )

    paginated_select_all_response = await Book.select(skip=0, limit=2)
    paginated_select_some_fields_response = await Book.select(
        columns=["author"], skip=2, limit=2
    )

    print("all:")
    pp.pprint(select_all_response)
    print("\nby id:")
    pp.pprint(select_by_id_response)
    print("\nsome fields for all:")
    pp.pprint(select_some_fields_response)
    print("\nsome fields for given ids:")
    pp.pprint(select_some_fields_for_ids_response)
    print("\npaginated; skip: 0, limit: 2:")
    pp.pprint(paginated_select_all_response)
    print("\npaginated returning some fields for each; skip: 2, limit: 2:")
    pp.pprint(paginated_select_some_fields_response)


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

Select Records Page by Page

In order to avoid overwhelming the server's memory resources, we can get the records one page at a time i.e. pagination.

We do this by specifying the number of records per page (limit) and the number of records to skip (skip) when calling the model's select method

Info

Records are ordered by timestamp of their insert into redis.

For batch inserts, the time difference is quite small but consistent.

Tip

You don't have to pass the skip if you wish to get the first records. skip defaults to 0.

limit, however is mandatory.

Warning

When both ids and limit are supplied, pagination is ignored.

It wouldn't make any sense otherwise.

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"),
            Book(title="Utah Blaine", author="Louis L'Amour"),
        ]
    )

    select_all_response = await Book.select()
    select_by_id_response = await Book.select(
        ids=["Oliver Twist", "Pride and Prejudice"]
    )

    select_some_fields_response = await Book.select(columns=["author"])
    select_some_fields_for_ids_response = await Book.select(
        ids=["Oliver Twist", "Pride and Prejudice"], columns=["author"]
    )

    paginated_select_all_response = await Book.select(skip=0, limit=2)
    paginated_select_some_fields_response = await Book.select(
        columns=["author"], skip=2, limit=2
    )

    print("all:")
    pp.pprint(select_all_response)
    print("\nby id:")
    pp.pprint(select_by_id_response)
    print("\nsome fields for all:")
    pp.pprint(select_some_fields_response)
    print("\nsome fields for given ids:")
    pp.pprint(select_some_fields_for_ids_response)
    print("\npaginated; skip: 0, limit: 2:")
    pp.pprint(paginated_select_all_response)
    print("\npaginated returning some fields for each; skip: 2, limit: 2:")
    pp.pprint(paginated_select_some_fields_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
all:
[   Book(title='Oliver Twist', author='Charles Dickens'),
    Book(title='Utah Blaine', author="Louis L'Amour"),
    Book(title='Jane Eyre', author='Emily Bronte'),
    Book(title='Pride and Prejudice', author='Jane Austen')]

by id:
[   Book(title='Oliver Twist', author='Charles Dickens'),
    Book(title='Pride and Prejudice', author='Jane Austen')]

some fields for all:
[   {'author': 'Charles Dickens'},
    {'author': "Louis L'Amour"},
    {'author': 'Emily Bronte'},
    {'author': 'Jane Austen'}]

some fields for given ids:
[{'author': 'Charles Dickens'}, {'author': 'Jane Austen'}]

paginated; skip: 0, limit: 2:
[   Book(title='Oliver Twist', author='Charles Dickens'),
    Book(title='Jane Eyre', author='Emily Bronte')]

paginated returning some fields for each; skip: 2, limit: 2:
[{'author': 'Jane Austen'}, {'author': "Louis L'Amour"}]