FastAPIのログ設定と全体のエラーハンドリング

DockerでFastAPI環境構築(HelloWorldまで)
https://qiita.com/rom0323/items/5d338d3fe60644c196c4
FastAPIのDB接続とマイグレーション(DIコンテナも準備)
https://qiita.com/rom0323/items/565c7b9c125824a0fbe9

まずは前回からのディレクトリ構成に今回、追加するディレクトリやファイルを示します。

project_root
├── _docker
│   ├── nginx
│   │   └── nginx.conf
│   └── python
│       └── Dockerfile
├── src
│   └── main.py
│   └── init.py // 追加(mainから初期化処理を切り離し)
│   └── core
│        └── dependency.py
│        └── logging.py // ログ
│   └── middleware //ディレクトリ追加
│        └── exception_handler.py // エラーハンドラー
│   └── schema //ディレクトリ追加
│        └── response //ディレクトリ追加
│　　　        └── error_response.py // エラーレスポンス
│   └── database
│        └── database.py
├── .env
├── pyproject.toml
├── poetry.lock
├── makefile
├── makefile.local
├── makefile.container
├── .gitignore
└── docker-compose.yml

ログには標準パッケージのloggingを使用します。

ログ設定と関数作成

ログの種別は下記の3つとします。(15日ローテーション)

access_log
- uvicornのログでアクセスログを出力する
error_log
- uvicornのログでuvicornのエラーを出力する。
exception_log
- アプリケーションレベルのエラーやログを出力する

命名はもっとわかりやすい方がいいですね。。。

ログの出力場所は/logとしました。日付のついていないログファイルが現在使用しているログファイルで、ローテーションされると日付が付与されて別ファイルとなります。

logs/
├── access.log
├── access.log.2024-05-17
├── error.log
├── error.log.2024-05-17
├── exception.log
└── exception.log.2024-05-17

設定と関数を記載する

src/core/loggin.py

import logging
from logging.config import dictConfig
import os
import traceback

# ログファイルのディレクトリを作成（存在しない場合）
log_directory = "logs"
if not os.path.exists(log_directory):
    os.makedirs(log_directory)

# ロギング設定
LOGGING_CONFIG = {
    'version': 1,
    'disable_existing_loggers': False,
    'formatters': {
        'default': {
            'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s',
        },
    },
    'handlers': { // 各ログの設定
        'access_log': {
            'level': 'INFO',
            'class': 'logging.handlers.TimedRotatingFileHandler',
            'when': 'midnight',
            'interval': 1,
            'backupCount': 15,
            'filename': os.path.join(log_directory, 'access.log'),
            'formatter': 'default',
        },
        'error_log': {
            'level': 'ERROR',
            'class': 'logging.handlers.TimedRotatingFileHandler',
            'when': 'midnight',
            'interval': 1,
            'backupCount': 15,
            'filename': os.path.join(log_directory, 'error.log'),
            'formatter': 'default',
        },
        'exception_log': {
            'level': 'ERROR',
            'class': 'logging.handlers.TimedRotatingFileHandler',
            'when': 'midnight',
            'interval': 1,
            'backupCount': 15,
            'filename': os.path.join(log_directory, 'exception.log'),
            'formatter': 'default',
        },
    },
    'loggers': {
        'uvicorn.access': {
            'handlers': ['access_log'],
            'level': 'INFO',
            'propagate': False,
        },
        'uvicorn.error': {
            'handlers': ['error_log'],
            'level': 'ERROR',
            'propagate': False,
        },
        'app.exception': {
            'handlers': ['exception_log'],
            'level': 'ERROR',
            'propagate': False,
        },
    },
}

dictConfig(LOGGING_CONFIG)

def log_error(e: Exception) -> None:
    error_logger = logging.getLogger("app.exception")
    detailed_tb = get_error_message(e)
    error_logger.error(f"An error occurred: {e}\n{detailed_tb}")
    
def get_error_message(e: Exception) -> str:
    tb = traceback.format_exception(type(e), e, e.__traceback__)
    return "".join(tb[::-1]) # 逆順にしてスタックトレースを取得

これでログを出力する準備は整いました。

めっちゃお手軽ですね！

仮のエンドポイントを作成

アクセスして例外を出すようにしておきます。この後、エラーハンドリングができているか確かめます。

src/main.py

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    try:
        # 何らかの処理
        raise Exception("Internal Server Error")
    except Exception as e:
        raise HTTPException(status_code=500)

エラーのレスポンスモデルも作成しておきましょう

基本的にはエラーは全て下記のレスポンススキーマに従います。pydanticのバリデーションエラーの形を模倣して作成してあるため、フロントでエラーのデータを受け取る際には、エラーの受け取り管理が楽になります。

src/schema/response/error_response.py

from typing import List

from pydantic import BaseModel, Field

class ErrorDetail(BaseModel):
    loc: List[str] = Field(..., description="エラー箇所")
    msg: str = Field(..., description="エラーメッセージ")
    type: str = Field(..., description="エラータイプ")

    class ConfigDict:
        json_schema_extra = {
            "example": {
                "loc": [],
                "msg": "1文字以上である必要があります。",
                "type": "value_error",
            }
        }

class ErrorJsonResponse(BaseModel):
    detail: List[ErrorDetail] = Field(..., description="エラーメッセージ")

エラーハンドラーの作成

.envのAPP_ENVがdevelopmentになっているか確認します。

/.env

APP_ENV=development

APP_ENVがproduction以外は基本的にレスポンスにスタックトレースを含みます。

http_exception_handlerはHTTPExceptionクラスの例外を補足した場合に使用されます。

EnhancedTracebackMiddlewareクラスはその他の例外が投げられた際に補足してresponseをハンドリングしています。

src/middleware/exception_handler.py

import os

from dotenv import load_dotenv
from fastapi import HTTPException, Request
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware

from src.main import app
from src.schema.response.error_response import ErrorJsonResponse
from src.core.logging import log_error, get_error_message

load_dotenv()
# 環境設定を取得
app_env = os.getenv("APP_ENV", "development")

# カスタムHTTPエラーハンドラの追加
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    log_error(f"An error occurred: {exc}")
    # 本番環境ではスタックトレースをレスポンスに含めない
    if app_env == "production":
        error_detail = "Internal Server Error"
    else:
        error_detail = exc.detail
    error = ErrorJsonResponse(
        detail=[
            {
                "loc": [f"{request.method} {request.url.path}"],
                "msg": error_detail,
                "type": "http_error",
            }
        ]
    )
    return JSONResponse(status_code=exc.status_code, content=error.dict())

class EnhancedTracebackMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        try:
            response = await call_next(request)
            return response
        except Exception as exc:
            log_error(exc)

            # 本番環境ではスタックトレースをレスポンスに含めない
            if app_env == "production":
                error_detail = "Internal Server Error"
            else:
                error_detail = get_error_message(exc)

            error = ErrorJsonResponse(
                detail=[
                    {
                        "loc": [f"{request.method} {request.url.path}"],
                        "msg": error_detail,
                        "type": "server_error",
                    }
                ]
            )
            return JSONResponse(status_code=500, content=error.dict())

HTTPエラーハンドラーで@appを使用して、src/mainのappを参照しています。

またsrc/mainでミドルウェア登録を行うため、互いのファイルを参照し合うことになってしまいます。

それを防ぐために、appのインスタンス化をsrc/init.pyに切り出します。

src/init.pyではその他の初期化も行っても良いと思います。

src/init.py

from fastapi import FastAPI

app = FastAPI(
    title="app_name",
    description="API for app_name",
    verion="0.1.0",
    servers=[
        {
            "url": "http://localhost:8000",
            "description": "Local server"
        }
    ]
)

src/main.py

from fastapi import HTTPException
from src.middleware.exception_handler import EnhancedTracebackMiddleware
from src.init import app

app = app
// ミドルウェアの登録
app.add_middleware(EnhancedTracebackMiddleware)

@app.get("/")
async def root():
    try:
        # 何らかの処理
        raise Exception("Internal Server Error")
    except Exception as e:
        raise HTTPException(status_code=500)

ハンドラーのapp参照も修正します。

src/middleware/exception_handler.py

import os

from dotenv import load_dotenv
from fastapi import HTTPException, Request
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware

from src.init import app // 修正 main->init
from src.schema.response.error_response import ErrorJsonResponse
from src.core.logging import log_error, get_error_message

load_dotenv()
# 環境設定を取得
app_env = os.getenv("APP_ENV", "development")

# カスタムエラーハンドラの追加
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    log_error(f"An error occurred: {exc}")
    # 本番環境ではスタックトレースをレスポンスに含めない
    if app_env == "production":
        error_detail = "Internal Server Error"
    else:
        error_detail = exc.detail
    error = ErrorJsonResponse(
        detail=[
            {
                "loc": [f"{request.method} {request.url.path}"],
                "msg": error_detail,
                "type": "http_error",
            }
        ]
    )
    return JSONResponse(status_code=exc.status_code, content=error.dict())

省略

これで、エラーハンドリングとログの設定は完了です。

にアクセスして、実際にログが出るか確認できると思います。

access_logは常に記録されます。

error_logはimportなどあえて内部でエラーを起こしてみてください

exception_logは例外を投げれば補足されると思います。

翌日に試せば、ファイルが新たに作成されることも確認できるかと思います。

ログとエラーハンドリングは以上とします。

次は、エンドポイントからサービス、リポジトリと一連のデータ取得の流れを書いてみようと思います。