np.nanをNoneに変換しないとDBアクセス時にエラーになることがある

始めに

※ 自宅で再現しなかったので、そういう事象が発生したということをメモするだけの記事。

FastAPIでアップロードされたCSVをもとに登録する処理を作っていました。そして、特定の条件でエラーになることに気づきました。しかし、エラーが発生行を確認しているとINSERT や UPDATEではなく、SELECTしたタイミングでエラーになっていることがわかりました。

今回の記事では、回避した方法を残します。

環境

• Python
• SQLAlchemy
• MySQL

※ 再現しなかったのでバージョン不明

実装

特定の条件で発生していたエラーはnp.nanのパラメータを検索で使用していた時に発生していました。

query = select(User).where(User.id.in_([np.nan]))
# この検索時に発生
actual = (await db.execute(query)).scalars().all()

そのため、最も簡単な方法としてはnp.nanをNoneに変換することで簡単に回避できます。都度、np.nanのチェックを入れる方法もあるのですが、メモリに余裕がある場合はこちらで一気に置換してしまったほうがコードの可読性も上がるし楽だと思います。

import pandas as pd
import numpy as np

# Sample DataFrameを作成します.
df = pd.DataFrame({'Column1': [1, 2, np.nan],
                   'Column2': [3, np.nan, 6],
                   'Column3': [7, 8, 9]})

df.replace({np.nan: None}, inplace=True)

ソースコード

なし

終わりに

発生後でブログにしようと考えていたのですが、その後にライブラリアップデートやPython本体のバージョン等のアップデートをかけてしまったせいで発生条件を見失ってしまいました。

更新時に変換できないパターンならともかく、検索時に変換できずにエラーになってしまうパターンを面白いと思ったタイミングですぐにブログにしてしまうべきでしたね。内容自体は薄いですが、そういうこともあるという周知ができればいいかなと思ってます。

Pythonのコンテキストマネジャーをまとめる

始めに

Pythonのコンテキストマネージャは、with文と組み合わせてリソース（ファイル、DB接続、ロックなど）の確保と解放を自動化する仕組みです。enterメソッドでセットアップ、exitメソッドでクリーンアップを行うことで、例外発生時も安全にリソース管理できます。

知っておけば便利なことが多いので、ブログに便利な箇所をまとめます。

環境

• Python
  • 3.13.3

実装

基本的には明示的にリソースの開放が必要な時に使用します。ただ、DB自体のリソース開放はライブラリ側が自動的に行ってくれますので、インスタンス生成時にコンテキストマネジャーを使用しましょう。

async def get_reader_db():
    async with async_session() as session:
        yield session
        
SessionReaderDep = Annotated[AsyncSession, Depends(get_reader_db)]

@router.get("/sample")
async def sample(
    db: SessionReaderDep,
) -> Any:
    try:
        // 適当な処理
        await sample2(db);
        await session.commit()
    except HTTPException:
        await session.rollback()
        raise
    except Exception as e:
        await session.rollback()
        raise e

ただし、通常時のコミット処理や、エラーが発生したときのロールバックやログの出力等々はアプリケーション側の責任となります。どの箇所でも同じ記載をすることになりますので、それもまたコンテキストマネジャーで共通化しましょう。

from contextlib import asynccontextmanager

from fastapi import HTTPException
from sqlalchemy.ext.asyncio import AsyncSession

# コンテキストマネジャーの定義
@asynccontextmanager
async def session_context(session: AsyncSession) -> None:
    try:
        yield session
        await session.commit()
    except HTTPException:
        await session.rollback()
        raise
    except Exception as e:
        await session.rollback()
        raise e
 
 
@router.get("/sample")
async def sample(
    param_db: SessionReaderDep,
) -> Any:
    async with session_context(param_db) as db:
        // 適当な処理
        await sample2(db);

さらに重要なのが__exit__または__aexit__処理でFalseかNoneを返却することです。もし定義しない場合、エラー発生行がコンテキストマネジャーで発生したことになり、何行目でエラーが発生したかがわからなくなります。ライブラリとしては意図的にエラーをまとめていると思いますが、アプリケーションの場合にはエラー行を特定させない行為にメリットは薄いので、必ず追加で定義したほうが良いです。

• https://docs.python.org/ja/3.8/reference/compound_stmts.html#id1

例外は、別の例外を送出するような finally 節が無い場合にのみ呼び出しスタックへ伝わります。新しい例外によって、古い例外は失われます。 `

@asynccontextmanager
async def session_context(session: AsyncSession) -> None:
    try:
        yield session
        await session.commit()
    except HTTPException:
        await session.rollback()
        raise
    except Exception as e:
        await session.rollback()
        raise e
        
    async def __aexit__(self, exc_type, exc_val, exc_tb):  # type: ignore
        """
        呼出元にエラーが発生したときに伝播させたいので、Falseを常に返却する。
        発生しないと、全てのエラーがcontext_managerの行で発生したことになる。
        """
        return False

@router.get("/sample")
async def sample(
    param_db: SessionReaderDep,
) -> Any:
    async with session_context(param_db) as db: // この行でエラーになったことになる
        // 適当な処理
        await sample2(db);
        
async def sample2(db):
    raise ValueError("") // 本当はこの行でエラーになったのに。

また、コンテキストマネジャーはデコレータとしても使用できます。例えば、バッチ処理で処理開始前と処理開始後にログを出力して、処理時間を計りたい等々があったとします。それも簡単に作成できます。

@asynccontextmanager
async def timeit(operation_name: str = "Operation"):
    """
    処理時間を計測し、開始と終了時にログを出力するコンテキストマネージャー

    Args:
        operation_name: 操作名（ログに表示される）
    """
    start_time = time.time()
    logger.info(f"[{operation_name}] 処理開始")

    try:
        yield
    finally:
        end_time = time.time()
        elapsed_time = end_time - start_time
        logger.info(f"[{operation_name}] 処理完了 - 実行時間: {elapsed_time:.4f}秒")
        
async def test_decolator():
    @timeit("TEST DECOLATOR")
    async def test_func():
        await asyncio.sleep(0.1)

    # WHEN
    await test_func()
    # ログとして、次のログが出力される
    # [TEST DECOLATOR] 処理開始
    # [TEST DECOLATOR] 処理完了 - 実行時間: 0.1022秒

このようにリソース管理を安全にするだけでなく、簡単にデコレータ作成までできてしまうので、意外と便利な機能です。

ソースコード

• https://github.com/hirotoKirimaru/fastapi-practice/blob/7e49ee151f90db1626fe2a87983f5001b6578496/src/helper/helper.py#L11
• https://github.com/hirotoKirimaru/fastapi-practice/blob/df87a8aa21612ca8134c98282b3a27e97bd7e0e7/tests/unit/helper/test_helper.py

終わりに

正直なところでいえば、コンテキストマネジャーに絞った記事というよりは、__exit__による挙動に少々苦しめられたのでそれを共有するための記事でした。ただ、内容が薄くなってしまったので、コンテキストマネジャーをまとめるという形で記事を記載しています。

参考情報

• https://docs.python.org/ja/3.8/reference/compound_stmts.html#id1

Pythonの起動時のメモリ消費を何とかしたかった

始めに

※ 結果的に言えばどうにもできなかった話です。

---

検証環境へのデプロイが通常なら10分程度で終わるはずなのに、最悪のパターンだと30分くらいかかるようになりました。原因を追っていくと、どうやらFastAPIが起動時にメモリを大量消費していることがわかりました。特にライブラリのimport時に非常にメモリを消費しており、メモリ使用量が100%になって張り付いた結果、応答待ちになっていることがわかりました。取り急ぎ、検証環境のメモリを増やすことで通常通りの10分程度のデプロイ時間に戻せたのですが、通常利用時のメモリ使用量が多くないのに、起動時の問題だけでメモリを増やすのは根本解決にはならないと考えていたので、それを何とかしようとした試みについて書いたブログです。

環境

• Python
  • 3.13
• tuna
  • 0.5.11

実装

import時間を知りたい

まずは、import時間を可視化することを目指しました。import時間が長いモジュールはメモリ消費量が多い傾向があるためです。

Python 3.7以降であれば、標準機能としてimportにかかる時間を表示できる機能があります。 機能をオンにすると次のようにどのライブラリでimporttimeがかかっているか出力されます。

import time: self [us] | cumulative | imported package
import time:        88 |         88 | django
import time:       646 |        646 |       billiard.reduction
import time:      1221 |       1867 |     billiard.connection
import time:       662 |       2528 |   billiard.dummy
import time:      1798 |       4326 | billiard.pool
import time:       111 |        111 | redis
import time:        68 |         68 | uwsgi

起動オプションに-X importtimeを付与してください。または環境変数のPYTHONPROFILEIMPORTTIMEを設定してください。

# オプション
python3 -X importtime -m uvicorn src.main:app --host 0.0.0.0 --port 9997

# 環境変数
PYTHONPROFILEIMPORTTIME=1 python3 -m uvicorn src.main:app --host 0.0.0.0 --port 9997

ただし、これだけ見ても理解が難しいので、別のツールで可視化します。ファイルとして出力しましょう。

python3 -X importtime -m uvicorn src.main:app --host 0.0.0.0 --port 9997 2>&1 | grep "^import time:" > importtime.log

# `2>&1`  標準エラー出力を標準出力にリダイレクトします（importtimeの結果は標準エラーに出力されるため）

もし、Dockerで起動しているアプリケーションであれば、次のコマンドでログを取得しましょう。

# compose.yml に定義された fastapi サービスをログに出力する
## --no-log-prefix が付与されているとどのサービスから出力されたか、というログを削ります。この文言があると構造が変わるので可視化できません。

docker compose logs --no-log-prefix fastapi > importtime.log

import時間の可視化

今回、import時間の可視化にtunaを使用しました。使用方法は簡単で、先ほど出力したファイルを読み込ませるだけです。

uv add --dev tuna
uv run tuna importtime.log

私のソースコードではsrc.api.mainに依存しているgoogle.generativeaiライブラリが重そうであることがわかりました。

なお、もし可視化できない場合、内部的に起動が終わっておらず、可視化に必要な情報がそろいきっていない場合があるので、起動を待ってから再度ファイルにしましょう。

対策しようとした方法

基本的にはlazy loadで対応することになります。もしくは、先に重たいライブラリを読み込んでおくことで、メモリ消費タイミングをずらそうとしていました。

import sqlalchemy # noqa

起動時

次のように重たいライブラリを先に読み込んでおくことで対応しようとしていました。しかし、起動時にメモリ消費が多いのに、読み込むタイミングをずらしても意味はほとんどありませんでした。

lifespan時

FastAPIではアプリケーション起動後からリクエスト受信前に一度だけ実行されるロジックを記載できます。このタイミングでライブラリを読み込むことは一定の効果を得られました。ただし、一定といってもかなりごくわずかだったので意味が薄かったです。

ヘルスチェック時

ECSではヘルスチェックが通って問題ないときに旧アプリと新アプリを入れ替えるような設定を入れていました。そのため、ヘルスチェックのエンドポイントでライブラリを読み込むようにしました。ただ、このタイミングで読み込んでも起動時のメモリが消費されたままであることが多く、import timeには大きい影響を与えませんでした。

エンドポイントアクセス時

エンドポイントにアクセスされたタイミングでimportすることで、メモリ消費タイミングをずらそうとしました。

これもメモリに対しては一定の効果を得られたのですが、アプリケーションの挙動を安定させることができず断念しました。

唯一効果があった処理

ファイルの上部でimportするとファイルを読み込んだタイミングでライブラリを読み込んでしまうため、メソッド内でimport処理をしました。かなり重めな機械学習のライブラリをメソッド内に入れることで、メソッド実行時にライブラリが読み込まれるため改善が見込めました。ただ、多用するとソースコードの見通しも悪くなるため、部分的に適用したものの、基本的には対応していないです。

# メソッド内でimportする例（効果的な方法）
def process_data():
   # 重いライブラリはここでimportする
   import google.generativeai

学んだこと

アプリケーション起動時には、次のことを意識すると起動時のメモリ消費量を減らせます。

• APIのroutingのファイルは最低限のファイルしか読み込ませない
• routingの以外の処理はすべて別ファイルに追い出し、メソッド内で読み込ませる

ソースコード

なし

終わりに

最終的には何もできていないのですが、いろいろと学べました。あとでこういう問題に当たらないように、最初から意識して設計するべきですが、手遅れになってから気づくのでなかなか難しいですね…。

参考情報

• https://docs.python.org/3/using/cmdline.html#cmdoption-X