始めに
※ 自宅で検証した際には実装できなかったので詳細は不明です。
テスト実装中に次のエラーが発生しました。
E sqlalchemy.exc.NoInspectionAvailable: No inspection system is available for object of type <class 'models.User'>
発生原因が不明ですが、発生しないように対応できたのでその件を記事にします。
環境
対応
インスタンスをcopyして対応しました。または、commitでも対応はできそうです。
もともとやりたかったことはDBの値をHTTPリクエストで更新できるかどうかをチェックすることでしたが、項目数が多かったのでDBにinsertしたインスタンスを元に加工していました。その加工方法が誤っており、エラーが発生した模様です。
import copy
def test_01(session: AsyncSession)
user = User(name="test")
session.add(user)
await session.flush()
NOTE
copied_user = copy.copy(user.__dict__)
_ = copied_user.pop("_sa_instance_state")
copied_user['name'] = "testUpd"
query: select = select(User).where(User.name == "test")
actual = (await db.execute(query)).scalars().first()
原因
自宅で再現できなかったので想像です。
次のエラーはmodels.Userを操作したときにSQLAlchemyのメタデータが正しく取得できず、インスタンスとDBを紐づける操作ができなかった時に発生します。
E sqlalchemy.exc.NoInspectionAvailable: No inspection system is available for object of type <class 'models.User'>
今回、DBに登録したインスタンスを元に入力値を加工しようとしたので、HTTPリクエスト時のJSONBodyに不要だと判断したSQLAlchemyのMetadataの_sq_instance_stateを削除しています。
copied_user = user.__dict__
_ = copied_user.pop("_sa_instance_state")
SQLAlchemyを使用して同一トランザクション内で処理する場合、処理結果をキャッシュに保持していて、処理結果に紐づいたインスタンスを返却します。今回の例でいうと、userとactualが別名ですが同一インスタンスを指しています。
user = User(name="test")
session.add(user)
...
query: select = select(User).where(User.name == "test")
actual = (await db.execute(query)).scalars().first()
そして、__dict__を使用したらインスタンスをDeepCopyできていると勘違いしたので、このcopied_userも同じインスタンスです。
copied_user = user.__dict__
_ = copied_user.pop("_sa_instance_state")
そのため、次のタイミングでインスタンスにマッピングしようとしたところ、本来マッピングできるはずのuserインスタンスが存在せずにメタデータのエラーが発生したと思われます。
query: select = select(User).where(User.name == "test")
actual = (await db.execute(query)).scalars().first()
だからこそ、copyでインスタンスをDeepCopyしたことにより、今回の事象を回避できたと推測してます。
copied_user = copy.copy(user.__dict__)
ソースコード
再現できなかったのでなし。
終わりに
起こった事象自体はかなりニッチな内容で、一般的に役に立たない記事かもしれませんが、NoInspectionAvailableで調べた際に私の記事が参考になって解決できると幸いです。
参考情報
類似情報
