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
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(title="Utah Blaine", author="Louis L'Amour"),
]
)
select_all_response = Book.select()
select_by_id_response = Book.select(ids=["Oliver Twist", "Pride and Prejudice"])
select_some_fields_response = Book.select(columns=["author"])
select_some_fields_for_ids_response = Book.select(
ids=["Oliver Twist", "Pride and Prejudice"], columns=["author"]
)
paginated_select_all_response = Book.select(skip=0, limit=2)
paginated_select_some_fields_response = 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)
Select All Records
To select all records for the given model in redis, call the model's select
method without any arguments.
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(title="Utah Blaine", author="Louis L'Amour"),
]
)
select_all_response = Book.select()
select_by_id_response = Book.select(ids=["Oliver Twist", "Pride and Prejudice"])
select_some_fields_response = Book.select(columns=["author"])
select_some_fields_for_ids_response = Book.select(
ids=["Oliver Twist", "Pride and Prejudice"], columns=["author"]
)
paginated_select_all_response = Book.select(skip=0, limit=2)
paginated_select_some_fields_response = 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)
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 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(title="Utah Blaine", author="Louis L'Amour"),
]
)
select_all_response = Book.select()
select_by_id_response = Book.select(ids=["Oliver Twist", "Pride and Prejudice"])
select_some_fields_response = Book.select(columns=["author"])
select_some_fields_for_ids_response = Book.select(
ids=["Oliver Twist", "Pride and Prejudice"], columns=["author"]
)
paginated_select_all_response = Book.select(skip=0, limit=2)
paginated_select_some_fields_response = 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)
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 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(title="Utah Blaine", author="Louis L'Amour"),
]
)
select_all_response = Book.select()
select_by_id_response = Book.select(ids=["Oliver Twist", "Pride and Prejudice"])
select_some_fields_response = Book.select(columns=["author"])
select_some_fields_for_ids_response = Book.select(
ids=["Oliver Twist", "Pride and Prejudice"], columns=["author"]
)
paginated_select_all_response = Book.select(skip=0, limit=2)
paginated_select_some_fields_response = 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)
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 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(title="Utah Blaine", author="Louis L'Amour"),
]
)
select_all_response = Book.select()
select_by_id_response = Book.select(ids=["Oliver Twist", "Pride and Prejudice"])
select_some_fields_response = Book.select(columns=["author"])
select_some_fields_for_ids_response = Book.select(
ids=["Oliver Twist", "Pride and Prejudice"], columns=["author"]
)
paginated_select_all_response = Book.select(skip=0, limit=2)
paginated_select_some_fields_response = 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)
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 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(title="Utah Blaine", author="Louis L'Amour"),
]
)
select_all_response = Book.select()
select_by_id_response = Book.select(ids=["Oliver Twist", "Pride and Prejudice"])
select_some_fields_response = Book.select(columns=["author"])
select_some_fields_for_ids_response = Book.select(
ids=["Oliver Twist", "Pride and Prejudice"], columns=["author"]
)
paginated_select_all_response = Book.select(skip=0, limit=2)
paginated_select_some_fields_response = 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)
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"}]