5
6

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

More than 1 year has passed since last update.

[FastAPI] 鬼の爆速実装!!!デコレータを使用したロギング

Posted at

この記事を読んでできること

  • Pythonのloggingモジュールを使用して、ログ出力ができる
  • デコレータの実装

概要

Pythonでのロギングの実践

記述しないこと

  • FastAPIについて
  • 基礎的なPythonの記述方法
  • ロギングの基本的な設定(log_config.yaml)
  • 詳細な引数の設定

使用技術

  • Python 3.10.4
  • fastapi 0.78.0

実装例

main.py
from fastapi import FastAPI

from routers import router
from log import read_logger

read_logger()

app = FastAPI()
app.include_router(router.router)

api/address.py
from logging import getLogger getlogger

from log import (start_and_end_logging, ApiName)
from schemas.api.base_address import BaseAddress

@start_and_end_logging(ApiName.ADDRESS)
def fetch_address(address_base: BaseAddress, request_id):
# 関数の処理(以下略)
log_config.yaml
version: 1
formatters:
  simple_fmt:
    format: '%(asctime)s:%(levelname)s:%(message)s'
    datefmt: '%Y/%m/%d %I:%M:%S'
handlers:
  console:
    class: logging.StreamHandler
    level: DEBUG
    formatter: simple_fmt
    stream: ext://sys.stdout
loggers:
  ApplicationLogger:
    level: DEBUG
    handlers: [console]
    propagate: no
root:
  level: INFO
  handlers: []
log.py
import requests
import yaml
import socket

from fastapi.responses import JSONResponse
from requests.models import Response
from enum import Enum
from functools import wraps
from logging import getLogger, config
from inspect import signature


# ロガーの名前
APPLICATION_LOGGER = "ApplicationLogger"

logger = getLogger(APPLICATION_LOGGER)


class ApiName(Enum):
    """APIの種類を明示するclass"""

    # 申し込みAPI
    APPLICATION = "Application"
    # 住所検索API
    ADDRESS = "Address"


def read_logger():
    """ロガーの設定を読み込む"""
    with open("log_config.yaml") as file:
        read_data = file.read()
    yaml_data = yaml.safe_load(read_data)
    config.dictConfig(yaml_data)


def get_arg_value(func, args: tuple, name: str):
    """関数の引数から指定した名前の値を取得

    Args:
        func: 検索対象の関数
        args (tuple): 取得したい引数の値
        name (str): 取得したい引数名

    Raises:
        Exception: 関数の中に取得したい引数名が存在しない場合

    Returns:
        Any: 引数の値
    """
    sig = signature(func)
    keys = tuple(sig.parameters.keys())
    if name not in keys:
        raise Exception(f"関数の中に引数{name}が存在しません。")
    return args[keys.index(name)]


def start_and_end_logging(api_name: ApiName, id_name: str = "request_id"):
    """デコレータの引数を受け取る部分

    Args:
        api_name (ApiName): 出力したいAPIの引数名
        id_name (str, optional): 出力したいIDの引数名
    """

    def _start_and_end_logging(func):
        """APIの開始と終了のロギングをするデコレータ
        Args:
            func: デコレータの関数を受け取る部分

        Raises:
            Exception: 戻り値の型がintと異なる場合
        """

        @wraps(func)
        def wrapper(*args, **kwargs):
            """APIの処理前後で開始と終了のロギングをするラッパー

            Raises:
                Exception: 戻り値の型がstrと異なる場合

            Returns:
                result (int): API処理結果のclassオブジェクト
            """
            if not isinstance(api_name, ApiName):
                e = Exception(f"api_nameの型はApiName型を想定していますが、{type(api_name)}です。")
                logger.warning(e)
                raise e
            id = get_arg_value(func, args, id_name)
            logger.info(f"[{id}][START] POST {api_name.value} API is called.")
            result = func(*args, **kwargs)
            logger.info(f"[{id}][END] POST {api_name.value} API is called.")
            # resultに入るオブジェクトはそれぞれ親子関係ではないため、tupleで複数の型を与え、型チェックをおこなっている
            if not isinstance(result, (Response, JSONResponse)):
                e = Exception(f"{type(result)}: 戻り値の型が仕様と異なります。")
                logger.warning(e)
                raise e
            if result.status_code == requests.codes.ok:
                logger.info(f"[{id}]POST {api_name.value} API Completed 200 OK.")

            return result

        return wrapper

    return _start_and_end_logging
routers/router.py
import socket
import uuid

from fastapi import APIRouter

@router.post("/api/address", tags=["address"])
def fetch_address(base_address: BaseAddress):
    return address.fetch_address(base_address, str(uuid.uuid4()))
  • @router.post("/api/address")で返されるfetch_address関数にstart_and_end_loggingデコレータをつける
  • するとlog.pyのデコレータの処理が実行される
    • 関数の前と後ろで処理が実行される
      • result = func(*args, **kwargs)が関数の処理が呼ばれている部分
  • デコレータ内で型チェックを万全に行い、動的言語ながらお硬いコードを記述している

なぜデコレータを使用するのか

APIごとに共通するロギングを行う場合(今回はAPIの開始と終了)はデコレータを使用することで共通するロギングコードをそれぞれのAPIで書かなくて済む(可読性・保守性の向上)

ロギング結果

こんな感じで
image.png (21.4 kB)
image.png (32.9 kB)

参考

5
6
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
5
6

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?