FastAPIを用いた複雑な条件によるフィルタリングの実装

2021年4月15日仕事

こんにちは、ものレボの橋本です。
今回は以前の記事でも軽く触れましたが、ものレボのバックエンドの実装で使っているFastAPIについて、実務上あった課題への対応の概略について、書きたいと思います。

いきなり本題に入りますが、以下のようなデータがあった場合に、optionsの中身をフィルタリングしてAPIから返したいようなケースでした。

{
  "items": [
    {
      "id": 1,
      "name": "name1",
      "options": {}
    },
    {
      "id": 2,
      "name": "name2",
      "options": {
        "int": 1,
        "float": 1.1,
        "str": "foobazbar",
        "date": "2020-01-01"
      }
    }
  ]
}

optionsの中のデータ構造が決まっていれば、よくあるRESTfulなAPIなら/items?options_int=1のようなクエリパラメータでフィルタリングを実現することができますが、今回のケースでは、optionsは、クライアント側が任意にPOSTしたデータを保持するためのフィールドとなっていて、サーバサイドではどのようなデータ構造となっているかわからない状態で、常にクライアントサイドのみで使われるものでした。

そこで以下のような実装を用いて、フィルタリングを実現しました。

import json
from enum import Enum, auto
from fastapi import FastAPI, Query, Body
from pydantic import BaseModel, Field, Json
from typing import Any, Optional

app = FastAPI()


class StrEnum(str, Enum):
    @staticmethod
    def _generate_next_value_(name: str, start: int, count: int, last_values: list[Any]) -> str:
        return name.lower()


class Condition(StrEnum):
    EQ = auto()  # equal ==
    NEQ = auto()  # not equal !=
    GT = auto()  # greater than <
    GE = auto()  # greater equal <=
    LT = auto()  # less than >
    LE = auto()  # less equal >=


class Filter(BaseModel):
    key: str = Field(title="key")
    val: Any = Field(title="value")
    cond: Condition = Field(title="condition")


class Filters(BaseModel):
    options: list[Filter] = Field([], title="optionsのフィルタ")


class Item(BaseModel):
    id: int = Field(title="ID")
    name: str = Field(title="名前")
    options: dict[str, Any] = Field(title="options")


def options_filter(f: Filter, item: Item) -> bool:
    if f.key not in item.options:
        return False

    val = item.options[f.key]

    if type(f.val) != type(val):
        return False

    if f.cond == Condition.EQ:
        return val == f.val
    if f.cond == Condition.NEQ:
        return val != f.val
    if f.cond == Condition.GT:
        return val > f.val
    if f.cond == Condition.GE:
        return val >= f.val
    if f.cond == Condition.LT:
        return val < f.val
    if f.cond == Condition.LE:
        return val <= f.val

    return False  # pragma: no cover


items = [
    Item(id=1, name="名前1", options={}),
    Item(id=2, name="名前2", options={"int": 1, "float": 1.1, "date": "2021-01-01", "str": "foobazbar"}),
]


@app.post("/items/search", response_model=list[Item])
def search_items(
    name: str = Query("", title="name", description="nameの部分一致"),
    filters: Filters = Body(Filters(), title="filter", description="options filter")
) -> Any:
    results = items

    if name:
        results = list(filter(lambda x: name in x.name, results))

    for o in filters.options:
        results = list(filter(lambda x: options_filter(o, x), results))

    return results

少し長くなってしまいましたが、ほぼ定数や型の定義部分で重要なのは、filters: Filters = Body(Filters(), title="filter", description="options filter")の部分です。
これが何を行っているかというと、本来取得系であれば、/itemsをエンドポイントにして、GETでリクエストするのが一般的ですが、あえてPOSTで受けてやることで、複雑な条件部分をjsonで受け取ることができるようにしています。

FastAPIとしては、GETリクエストでもリクエストボディを受け取ることができるので、POSTではなく、GETで受けてもよいのですが、公式のドキュメントにも下記のようにあり、稀なユースケースとなるため、/items/searchと検索用のエンドポイントを用意してやることで対処しています。

データを送るには、POST (もっともよく使われる)、PUT、DELETE または PATCH を使うべきです。

GET リクエストでボディを送信することは、仕様では未定義の動作ですが、FastAPI でサポートされており、非常に複雑な（極端な）ユースケースにのみ対応しています。

https://fastapi.tiangolo.com/ja/tutorial/body/

また、その他の部分の実装は、key, value, conditonを受け取って条件通りにフィルタリングするだけのもので特筆すべきことはないため、ここでは解決を省略します。

上記の実装により、以下のようにリクエストボディにフィルタリング用のjsonを指定することで、期待した動作をさせることができました。

bash-5.1$ http POST 'http://127.0.0.1:8000/items/search'
HTTP/1.1 200 OK
content-length: 136
content-type: application/json
date: Thu, 15 Apr 2021 05:11:58 GMT
server: uvicorn

[
    {
        "id": 1,
        "name": "名前1",
        "options": {}
    },
    {
        "id": 2,
        "name": "名前2",
        "options": {
            "date": "2021-01-01",
            "float": 1.1,
            "int": 1,
            "str": "foobazbar"
        }
    }
]

bash-5.1$ cat filters.json 
{
  "options": [
    {
      "key": "float",
      "val": 1.0,
      "cond": "ge"
    }
  ]
}
bash-5.1$ http POST 'http://127.0.0.1:8000/items/search' < filters.json 
HTTP/1.1 200 OK
content-length: 97
content-type: application/json
date: Thu, 15 Apr 2021 05:12:03 GMT
server: uvicorn

[
    {
        "id": 2,
        "name": "名前2",
        "options": {
            "date": "2021-01-01",
            "float": 1.1,
            "int": 1,
            "str": "foobazbar"
        }
    }
]

しかし、取得系の処理にPOSTを使うことや、クエリパラメータとリクエストボディを併用する気持ち悪さもあるため、filtersの部分をjsonに変更してみます。

@app.get("/items", response_model=list[Item])
def read_items(
    name: str = Query("", title="name", description="nameの部分一致"),
    filters: Optional[Json] = Query(None, title="filter", description="options filter")
) -> Any:
    try:
        validated_filters = Filters(**filters)
    except ValidationError as e:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail=e.errors(),
        )

    results = items

    if name:
        results = list(filter(lambda x: name in x.name, results))

    for o in validated_filters.options:
        results = list(filter(lambda x: options_filter(o, x), results))

    return results

先程のコードに上記のような実装を追加することで、filtersでNoneもしくは、jsonをデコードしたdictが受け取れるので、それを更にFiltersにしています。
バリデーションエラーとなった場合には、通常のFastAPIの実装と同様のレスポンスになるように、ステータスコード=422でエラーを返してやっています。
（ここはもう少しいいやり方がある気がしますが、執筆中に思いつかなかったためより良い方法があれば教えてほしいです。）

bash-5.1$ http GET 'http://127.0.0.1:8000/items?filters=%7B%22options%22%3A%20%5B%7B%22key%22%3A%20%22float%22%2C%20%22val%22%3A%201.0%2C%20%22cond%22%3A%20%22ge%22%7D%5D%7D'
HTTP/1.1 200 OK
content-length: 97
content-type: application/json
date: Thu, 15 Apr 2021 05:17:55 GMT
server: uvicorn

[
    {
        "id": 2,
        "name": "名前2",
        "options": {
            "date": "2021-01-01",
            "float": 1.1,
            "int": 1,
            "str": "foobazbar"
        }
    }
]

これで期待したような実装ができましたが、filtersのクエリパラメータの可読性が著しく低く、optionsに複雑なデータが入りフィルタリングが複雑になった場合URLの長さも問題になりそうなため、risonを導入したいと思います。

@app.get("/items", response_model=list[Item])
def read_items(
    name: str = Query("", title="name", description="nameの部分一致"),
    filters: Optional[str] = Query(None, title="filter", description="options filter")
) -> Any:
    if filters:
        try:
            decoded_filters = prison.loads(filters)
            validated_filters = Filters(**decoded_filters)
        except prison.decoder.ParserException as e:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail=str(e),
            )
        except ValidationError as e:
            raise HTTPException(
                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
                detail=e.errors(),
            )

    results = items

    if name:
        results = list(filter(lambda x: name in x.name, results))

    for o in validated_filters.options:
        results = list(filter(lambda x: options_filter(o, x), results))

    return results

filtersをpydantic.Jsonから、strで受け取るように変更し、それをprisonでデコードする処理を追加します。
理想を言うのであれば、pydantic.Jsonのように、Rison型を用意してやってクエリパラメータで受け取った時点で、バリデーションとデコードがされているとベストです。

bash-5.1$ http GET 'http://127.0.0.1:8000/items?filters=(options:!((cond:ge,key:float,val:1.0)))'
HTTP/1.1 200 OK
content-length: 97
content-type: application/json
date: Thu, 15 Apr 2021 05:34:17 GMT
server: uvicorn

[
    {
        "id": 2,
        "name": "名前2",
        "options": {
            "date": "2021-01-01",
            "float": 1.1,
            "int": 1,
            "str": "foobazbar"
        }
    }
]

URLもスッキリと短くなりました。
少し可読性は下がりますが、key, val, condとしているものをk, v, cなど1文字にしてより短いURLにするのも有りだと思います。

最終的なコードはこのリポジトリにありますので、眺めてみたり、より良い方法があればフォークしてPRをいただけるととても嬉しいです。