👋

PythonからRediSearchを触ってみた

2022/02/28に公開

Python

Redis

tech

RediSearchとは

RediSearchはRedis上にモジュールとして構築されたセカンダリインデックス、クエリエンジン、分散型全文検索および集約エンジンです。Elasticsearchと比較したベンチ―マーク結果が公開されており、高速なインデックス作成が特徴です。記事執筆時点(2022/02/24)のv2.2では日本語対応していないようです。

環境

Python 3.10.2
redis-py 4.1.3
redis-om 0.0.17
module:name=ReJSON,ver=999999
module:name=search,ver=20206

pip install redis-om
docker run -p 6379:6379 redislabs/redisearch:latest

redis-om

redis-omはRediSearchを扱う際にインデックス付与やクエリ周りを良い感じにしてくれるライブラリです。RedisがサポートしているHash型やJson型を用いてモデリングします。pydanticが内部で使われているのでバリデーションやシリアライズ・デシリアライズもしやすいです。

from redis_om import JsonModel


class Article(JsonModel):
    title: str


article = Article(title="First article")
article.save()
article_pk = article.pk

print(article_pk)
# 01FWKS7P4148TAKNED9DK26FYE

loaded_article = Article.get(article_pk)
print(loaded_article) 
# pk='01FWKS7P4148TAKNED9DK26FYE' title='First article'

print(loaded_article.json())
# {"pk": "01FWKS7P4148TAKNED9DK26FYE", "title": "First article"}

上記のように記述することでRedisにデータを格納できます。何もしないとプライマリキーはULIDになります。ここでは詳しく触れませんが独自のルールを定めて生成することもできます。データを取得する時はプライマリキーを引数に渡して取得します。
JsonModelを継承することでJson型、HashModelを継承することでHash型でRedisに格納されます。JsonModelやHashModelの基底クラスにpydanticのBaseModelが使われているのでjson・dict出力なんかもできます。

検索

基本

from redis_om import Field, JsonModel, Migrator


class Article(JsonModel):
    title: str = Field(index=True)
    writer: str = Field(index=True)


Article(title="1st article", writer="user1").save()
Article(title="1st article", writer="user2").save()
Article(title="2nd article", writer="user1").save()
Article(title="3rd article", writer="user1").save()
Migrator().run()

# 条件に一致する要素一つ
article = Article.find(Article.writer == "user1").first()
print(article)
# pk='01FWVCJ4E0JG2QEVKVPDQY49PN' title='2nd article' writer='user1'

# 条件に一致する全て
user1_articles = Article.find(Article.writer == "user1").all()
print(user1_articles)
# [Article(pk='01FWVCJ4E0JG2QEVKVPDQY49PN', title='2nd article', writer='user1'),
#  Article(pk='01FWVCJ4DYS7KCDP2TPHFH4K0T', title='1st article', writer='user1'),
#  Article(pk='01FWVCJ4E12QRX6Y8HX2694YDH', title='3rd article', writer='user1')]

# AND
user1_1st_articles = Article.find((Article.writer == "user1") & (Article.title == "1st article")).all()
print(user1_1st_articles)
# [Article(pk='01FWVCJ4DYS7KCDP2TPHFH4K0T', title='1st article', writer='user1')]

# OR
first_and_third_articles = Article.find((Article.title == "1st article") | (Article.title == "3rd article")).all()
print(first_and_third_articles)
# [Article(pk='01FWVCJ4DYS7KCDP2TPHFH4K0T', title='1st article', writer='user1'),
#  Article(pk='01FWVCJ4E12QRX6Y8HX2694YDH', title='3rd article', writer='user1'),
#  Article(pk='01FWVCJ4DZMGD5WXMVGRHVDTW8', title='1st article', writer='user2')]

# 上と同じ。OR検索で一つの項目にリストで渡すこともできる
first_or_third_articles = Article.find(Article.title << ["1st article", "3rd article"]).all()
print(first_or_third_articles)
# [Article(pk='01FWVCJ4DYS7KCDP2TPHFH4K0T', title='1st article', writer='user1'),
#  Article(pk='01FWVCJ4E12QRX6Y8HX2694YDH', title='3rd article', writer='user1'),
#  Article(pk='01FWVCJ4DZMGD5WXMVGRHVDTW8', title='1st article', writer='user2')]


# 一致しない全て
not_user1_articles = Article.find(Article.writer != "user1").all()
print(not_user1_articles)
# [Article(pk='01FWVCJ4DZMGD5WXMVGRHVDTW8', title='1st article', writer='user2')]

# 上と同じ。否定
not_user1_articles = Article.find(~(Article.writer == "user1")).all()
print(not_user1_articles)
# [Article(pk='01FWVCJ4DZMGD5WXMVGRHVDTW8', title='1st article', writer='user2')]

条件一致、否定、AND、ORの基本的な構文です。検索を有効にしたいフィールドでField(index=True)とします。

テキスト検索

from redis_om import Field, JsonModel, Migrator


class Article(JsonModel):
    title: str
    type_: str = Field(index=True, full_text_search=True)


Article(title="1st article", type_="tech").save()
Article(title="2nd article", type_="tech startup").save()
Article(title="3rd article", type_="technology").save()
Migrator().run()

tech_articles = Article.find(Article.type_ == "tech").all()
print(tech_articles)
# [Article(pk='01FWP8YGCFZA4F8D2FRWSYKWH6', title='1st article', type_='tech')]

include_tech_articles = Article.find(Article.type_ % "tech").all()
print(include_tech_articles)
# [Article(pk='01FWP8YGCG6JA0R7VNTBNDDNQ8', title='2nd article', type_='tech startup'), 
#  Article(pk='01FWP8YGCFZA4F8D2FRWSYKWH6', title='1st article', type_='tech')]

fuzzy_tech_articles = Article.find(Article.type_ % "tech*").all()
print(fuzzy_tech_articles)
# [Article(pk='01FWP8YGCJ0HP1YBKQ9V9DF8C7', title='3rd article', type_='technology'), 
#  Article(pk='01FWP8YGCG6JA0R7VNTBNDDNQ8', title='2nd article', type_='tech startup'), 
#  Article(pk='01FWP8YGCFZA4F8D2FRWSYKWH6', title='1st article', type_='tech')]

全文検索を有効にするフィールドでfull_text_search=Trueを指定します。Fuzzy検索もできます。

数値

from redis_om import Field, JsonModel, Migrator


class Article(JsonModel):
    title: str
    view_count: int = Field(index=True)


Article(title="1st article", view_count=10).save()
Article(title="2nd article", view_count=20).save()
Article(title="3rd article", view_count=30).save()
Article(title="4th article", view_count=40).save()
Migrator().run()

between_20_30_views_articles = Article.find((Article.view_count >= 20) & (Article.view_count <= 30)).all()
print(between_20_30_views_articles)
# [Article(pk='01FWTY12YWRFQ2Q7PBBT726C14', title='3rd article', view_count=30), 
#  Article(pk='01FWTY12YV4XZD679M4PCYAJW4', title='2nd article', view_count=20)]

数値型は不等号を使って範囲検索できます。

日付

import datetime
from redis_om import Field, JsonModel, Migrator


class Article(JsonModel):
    title: str
    created_at: datetime.date = Field(index=True)


Article(title="1st article", created_at="2022-01-01").save()
Article(title="2nd article", created_at="2022-02-01").save()
Article(title="3rd article", created_at="2022-03-01").save()
Article(title="4th article", created_at=datetime.date(2022, 3, 1)).save()
Migrator().run()

articles = Article.find(Article.created_at == datetime.date(2022, 3, 1)).all()
"""
Traceback (most recent call last):
  File "main.py", line 16, in <module>
    articles = Article.find(Article.created_at == datetime.date(2022, 3, 1)).all()
    ...

TypeError: argument of type 'datetime.date' is not iterable
zsh: exit 1     python main.py
"""

データを格納する時はdate型でもstr型でもどちらの値を渡してもいいのですが、検索する時はdate型のままだとダメでした。下のコードのようにstr型の値を使うとこで検索できます。

articles = Article.find(Article.created_at == "2022-03-01").all()
print(articles)
# [Article(pk='01FWV8AVNCR1B0BSGCXCZBM3FP', title='4th article', created_at=datetime.date(2022, 3, 1)), 
#  Article(pk='01FWV8AVNBA9W31MMTYJQAA24T', title='3rd article', created_at=datetime.date(2022, 3, 1))]

articles = Article.find((Article.created_at < "2022-03-01")).all()
print(articles)
# []

上のコードでは日付の範囲検索を試みたのですがダメでした。範囲検索をするには数値型で格納する必要があります。下記のように格納時にUnixTimeに変換すると範囲検索できるようになりました。

import datetime
import pandas as pd
from pydantic import validator
from redis_om import Field, JsonModel, Migrator


def date_to_unix_time(date_time: str | int) -> int:
    if isinstance(date_time, int):
        return date_time
    else:
        unix_time = int(pd.Timestamp(date_time, tz="UTC").timestamp())
        return unix_time


class Article(JsonModel):
    title: str
    created_at: int = Field(index=True)

    created_at_unix_time = validator("created_at", allow_reuse=True, pre=True)(date_to_unix_time)


Article(title="1st article", created_at="2022-01-01").save()
Article(title="2nd article", created_at="2022-02-01").save()
Article(title="3rd article", created_at="2022-03-01").save()
Article(title="4th article", created_at=datetime.date(2022, 3, 1)).save()
Migrator().run()

articles = Article.find((Article.created_at > 1643673600)).all()  # 2022-02-01 = 1643673600
print(articles)
# [Article(pk='01FWVATEXV77DN0RKRJ9KBK1EC', title='3rd article', created_at=1646092800), 
#  Article(pk='01FWVATEXWZA19D1N9CFSF8PQX', title='4th article', created_at=1646092800)]

pydanticのvalidator機能を使ってUnixTimeに変換して数値化しました。データを格納する時と取り出す時にpydanticのバリデーションがはしるのでif文でint型の時(すでにUnixTimeに変換済み)は何も処理をせず値を返すようにしています。

EmbeddedJsonModel

from redis_om import Field, JsonModel, Migrator, EmbeddedJsonModel


class User(EmbeddedJsonModel):
    first_name: str = Field(index=True)
    last_name: str = Field(index=True)


class Article(JsonModel):
    title: str = Field(index=True)
    writer: User


Article(title="1st article", writer={"first_name": "taro", "last_name": "suzuki"}).save()
Article(title="1st article", writer={"first_name": "jiro", "last_name": "sato"}).save()
Article(title="1st article", writer={"first_name": "taro", "last_name": "sato"}).save()
Article(title="2nd article", writer={"first_name": "taro", "last_name": "sato"}).save()
Migrator().run()

taro_articles = Article.find(Article.writer.first_name == "taro").all()
print(taro_articles)
# [Article(pk='01FWVE6DVECMPB302N54E355TF', title='1st article', writer=User(pk='01FWVE6DVEMVY5QZB8S8J99KZC', first_name='taro', last_name='sato')), 
#  Article(pk='01FWVE6DVD9883RFK3FJ4EPAFC', title='1st article', writer=User(pk='01FWVE6DVDBGS9K94GSJ2XVH53', first_name='taro', last_name='suzuki')), 
#  Article(pk='01FWVE6DVFGCQKYSXJJKAX303P', title='2nd article', writer=User(pk='01FWVE6DVFBQBSZV365BM99X1S', first_name='taro', last_name='sato'))]

query = Article.find(
    (Article.writer.first_name == "taro") & (Article.writer.last_name == "sato") & (Article.title == "1st article")
)
articles = query.all()
print(articles)
# [Article(pk='01FWVE6DVECMPB302N54E355TF', title='1st article', writer=User(pk='01FWVE6DVEMVY5QZB8S8J99KZC', first_name='taro', last_name='sato'))]

print(query.expression.tree)
"""
           ┌first_name
        ┌EQ┤
        |  └taro
    ┌AND┤
    |   |  ┌last_name
    |   └EQ┤
    |      └sato
 AND┤
    |  ┌title
    └EQ┤
       └1st article
"""

ネストされたモデルの検索とクエリの可視化例です。

さいごに

Pythonのredis-omからRediSearchを触ってみました。redis-omを使うことで生のRediSearchコマンドを書いたり、RediSearchモジュールのPythonAPIを使うよりも手軽に記述できました。

RediSearchとは

環境

redis-om

検索

基本

テキスト検索

数値

日付

EmbeddedJsonModel

さいごに

Discussion