fastapi-cacheでUnion型をキャッシュした際にdictが返却される問題

fastapi-cacheとは

fastapiでのキャッシュ機構をよしなに提供してくれるライブラリ。
fastapiとの使用のためにjsonのシリアライズを考慮したキャッシュの仕組みや、asyncな関数のサポートを提供してくれている。
バックエンド=キャッシュ機構の実態としてredis、memcache、dynamodb、in memoryをサポートしているので自分の用途に合わせて選べる。

使用方法

まずはアプリケーションのスタートアップ時にキャッシュを初期化する必要がある。
このときにバックエンドの設定を行う。

from fastapi import FastAPI
from fastapi_cache import FastAPICache
from fastapi_cache.backends.inmemory import InMemoryBackend

@asynccontextmanager
async def lifespan(_app: FastAPI):
    # in-memoryなキャッシュの初期化
    FastAPICache.init(InMemoryBackend(), prefix="fastapi-cache")

    yield

app = FastAPI(lifespan=lifespan)

あとは好きな関数にcacheデコレータをつければ、その関数の結果がキャッシュされて、有効期限内には関数は走らずにキャッシュされた値がレスポンスされる。

@app.get("/")
@cache(expire=60) # この関数は60秒間は同じキャッシュが使用される。
async def index():
    return dict(hello="world")

キャッシュできる関数は別にrouteの関数だけでなく、好きなasync/syncな関数をキャッシュできる。
ので例えば外部APIを叩くときの結果のキャッシュとかにも使用できる。

キャッシュ結果のencode/decode

キャッシュしたいデータをどのようなデータ型で保持するかという問題がある。
redisや他のキャッシュ機構間で同じようにデータを保持できなければならないので何らかのencode/decodeの仕組みが必要になる。

fastapi-cacheではこの仕組みをCoderというクラス群で提供している。
デフォルトではJsonCoderが使用され、これはfastapiがjsonに変換可能なpydanticのmodelやdataclassなどを勝手にキャッシュしてくれる。
更に適切に関数に対して戻り値の型がついていれば、その型のクラスに変換してキャッシュの結果を返してくれる。

from .models import SomeModel, create_some_model

@app.get("/foo")
@cache(expire=60)
async def foo() -> SomeModel: # 2度目以降でもSomeModelが返却される
    return create_some_model()

注意しなければならないのは、関数の戻り値が型付けされていないとこの関数はdictやlistなどのpythonの標準データ型を返却する。

from .models import SomeModel, create_some_model

@app.get("/foo")
@cache(expire=60)
async def foo(): # 2度目以降はdictやlistで返却される
    return create_some_model()

この挙動は関数呼び出しの際に結果のデータ型が変わるなかなか危険なものだが、おそらくcacheの使用想定箇所のメインがroute関数であるため大きな問題にならない、という思想なのだと思う。
なぜならdictだろうが自前のクラスだろうがjsonに変換できて結果のjson値という意味では挙動は変わらないので。

Union型の値をキャッシュしたときの問題

pythonのtype hintではUnion型という物があって、ある型付の項に対して複数のデータ型の可能性を表現できる。
TypeScriptとかでよく使うやつ。

どういうときに使いたいかと言うと、エラーを起こす可能性のある関数の戻り値をUnion型で表現すると、そのエラーのハンドルを強制できて便利だったりする。

from pydantic import BaseModel

class Result(BaseModel):
    value: str

def may_raise() -> Result | SomeError:
    ...

def handle_incorrectly():
    result = may_raise()

    print(result.value) # resultはSomeErrorの可能性があるので、型チェックでエラーになる

def handle_correctly():
    result = may_raise()

    if isinstance(result, SomeError)
        raise RuntimeException()

    print(result.value) # resultがSomeErrorの場合は前段でハンドルされているので、型チェックに通る

こういう関数にも当然cacheをつけられる。

@cache(expire=60)
def may_raise() -> Result | SomeError:
    ...

ここでデフォルトのJsonCoderを使用していると問題が起きる。
なにかというと、Union型はJsonCoderが解釈encode元、decode先として解釈しないので二度目以降のこの関数の結果はdictやlistとして返却されてしまう。
更にたちが悪いのは一度目の呼び出し時にはすんなり元のクラスのインスタンスを返却してくるので、気付きづらい。
一度呼び出した結果のテストのみ書いているとこの変化に気付けない。

この挙動の解決策は簡単で、JsonCoderじゃなく、PickleCoderを使えば良い。

from fastapi_cache.coder import PickleCoder

@cache(expire=60, coder=PickleCoder)
def may_raise() -> Result | SomeError:
    ...

こうすればUnion型が戻り値に指定されていようと、何度呼んでもResultクラスのインスタンスが返却されるようになる。

参考

• https://github.com/long2ice/fastapi-cache