Models
The very first thing you need to create for pydantic-redis are the models (or schemas) that the data you are to save in redis is to be based on.
These models are derived from pydantic's BaseModel
.
Import Pydantic-redis' Model
First, import pydantic-redis' Model
import pprint
from datetime import date
from typing import List
from pydantic_redis import Model, Store, RedisConfig
class Book(Model):
_primary_key_field: str = "title"
title: str
author: str
rating: float
published_on: date
tags: List[str] = []
in_stock: bool = True
if __name__ == "__main__":
pp = pprint.PrettyPrinter(indent=4)
store = Store(
name="some_name",
redis_config=RedisConfig(db=5, host="localhost", port=6379),
life_span_in_seconds=3600,
)
store.register_model(Book)
Book.insert(
Book(
title="Oliver Twist",
author="Charles Dickens",
published_on=date(year=1215, month=4, day=4),
in_stock=False,
rating=2,
tags=["Classic"],
)
)
response = Book.select(ids=["Oliver Twist"])
pp.pprint(response)
Create the Model
Next, declare a new model as a class that inherits from Model
.
Use standard Python types for all attributes.
import pprint
from datetime import date
from typing import List
from pydantic_redis import Model, Store, RedisConfig
class Book(Model):
_primary_key_field: str = "title"
title: str
author: str
rating: float
published_on: date
tags: List[str] = []
in_stock: bool = True
if __name__ == "__main__":
pp = pprint.PrettyPrinter(indent=4)
store = Store(
name="some_name",
redis_config=RedisConfig(db=5, host="localhost", port=6379),
life_span_in_seconds=3600,
)
store.register_model(Book)
Book.insert(
Book(
title="Oliver Twist",
author="Charles Dickens",
published_on=date(year=1215, month=4, day=4),
in_stock=False,
rating=2,
tags=["Classic"],
)
)
response = Book.select(ids=["Oliver Twist"])
pp.pprint(response)
Specify the _primary_key_field
Attribute
Set the _primary_key_field
attribute to the name of the attribute
that is to act as a unique identifier for each instance of the Model.
Example
In this case, there can be no two books with the same title
.
import pprint
from datetime import date
from typing import List
from pydantic_redis import Model, Store, RedisConfig
class Book(Model):
_primary_key_field: str = "title"
title: str
author: str
rating: float
published_on: date
tags: List[str] = []
in_stock: bool = True
if __name__ == "__main__":
pp = pprint.PrettyPrinter(indent=4)
store = Store(
name="some_name",
redis_config=RedisConfig(db=5, host="localhost", port=6379),
life_span_in_seconds=3600,
)
store.register_model(Book)
Book.insert(
Book(
title="Oliver Twist",
author="Charles Dickens",
published_on=date(year=1215, month=4, day=4),
in_stock=False,
rating=2,
tags=["Classic"],
)
)
response = Book.select(ids=["Oliver Twist"])
pp.pprint(response)
Register the Model in the Store
Then, 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 datetime import date
from typing import List
from pydantic_redis import Model, Store, RedisConfig
class Book(Model):
_primary_key_field: str = "title"
title: str
author: str
rating: float
published_on: date
tags: List[str] = []
in_stock: bool = True
if __name__ == "__main__":
pp = pprint.PrettyPrinter(indent=4)
store = Store(
name="some_name",
redis_config=RedisConfig(db=5, host="localhost", port=6379),
life_span_in_seconds=3600,
)
store.register_model(Book)
Book.insert(
Book(
title="Oliver Twist",
author="Charles Dickens",
published_on=date(year=1215, month=4, day=4),
in_stock=False,
rating=2,
tags=["Classic"],
)
)
response = Book.select(ids=["Oliver Twist"])
pp.pprint(response)
Use the Model
Then you can use the model class to:
insert
into the storeupdate
an instance of the modeldelete
from storeselect
from store
Info
The store is connected to the Redis instance, so any changes you make will reflect in redis itself.
import pprint
from datetime import date
from typing import List
from pydantic_redis import Model, Store, RedisConfig
class Book(Model):
_primary_key_field: str = "title"
title: str
author: str
rating: float
published_on: date
tags: List[str] = []
in_stock: bool = True
if __name__ == "__main__":
pp = pprint.PrettyPrinter(indent=4)
store = Store(
name="some_name",
redis_config=RedisConfig(db=5, host="localhost", port=6379),
life_span_in_seconds=3600,
)
store.register_model(Book)
Book.insert(
Book(
title="Oliver Twist",
author="Charles Dickens",
published_on=date(year=1215, month=4, day=4),
in_stock=False,
rating=2,
tags=["Classic"],
)
)
response = Book.select(ids=["Oliver Twist"])
pp.pprint(response)
Run the App
Running the above code in a file main.py
would produce:
Tip
Probably FLUSHALL redis first