Flaskのログ設定(dictConfig)をカスタマイズして特定ロガー名のログ出力やログレベル毎のログの出し分け

はじめに〜ロギングはロガーオブジェクトを使用する〜

pythonでログ処理を実装しようとするときはloggingの基本チュートリアルを参考に構築すると思います。

src/logging/logging_test.py

import logging

def main():
    logging.info("it is logging.")

しかし、この記事でも指摘されているように、
この方法は不適切です。基本的にはロガーオブジェクトを使用することが推奨されています。

src/logging/logger_test.py

from logging import getLogger
logger = getLogger(__name__)

def main():
    logger.info("it is logger.")

Flaskのログ処理(logging)

一方、Flaskのログ処理（Logging）では、以下のようなdictConfigによる記法が紹介されています。

基本設定(Basic Configuration)

app.py

from flask import Flask
from logging.config import dictConfig
from src.logging import logging_test, logger_test

dictConfig({
    'version': 1,
    'formatters': {'default': {
        'format': '[%(asctime)s] %(levelname)s in %(module)s: %(message)s',
    }},
    'handlers': {'wsgi': {
        'class': 'logging.StreamHandler',
        'stream': 'ext://flask.logging.wsgi_errors_stream',
        'formatter': 'default'
    }},
    'root': {
        'level': 'INFO',
        'handlers': ['wsgi']
    }
})

app = Flask(__name__)

@app.route('/')
def hello_world():
    logging_test.main()
    logger_test.main()
    return 'Hello, World!'

if __name__ == "__main__":
    app.run(debug=True)

問題点

しかしこうすると、ロガーオブジェクトを設定したログが出力されません。何故？

[2020-11-12 09:47:05,506] INFO in logging_test: it is logging.
[2020-11-12 09:47:05,506] INFO in _internal: 127.0.0.1 - - [12/Nov/2020 09:47:05] "GET / HTTP/1.1" 200 -

これは、上記の記法ではルートロガー以外のロガーが出力されないためです。

解決方法 ~loggersの導入~

ルートロガー以外のロガーをdictConfigのloggersに追加すると解決できます。

    'loggers': {'src': {}},

こうするとsrcロガーを出力できるようになります。
logger_test.pyの__name__はsrc.logging.logger_testのため、出力されています。

[2020-11-12 10:02:17,987] INFO in logging_test: it is logging.
[2020-11-12 10:02:17,987] INFO in logger_test: it is logger.
[2020-11-12 10:02:17,988] INFO in _internal: 127.0.0.1 - - [12/Nov/2020 10:02:17] "GET / HTTP/1.1" 200 -

またログを出力するハンドラを個別に指定することもできます。

    'loggers': {'src': {
        'handlers': ['wsgi']
    }},

しかしハンドラがルートロガーのハンドラと被っている場合、ログが二重に出力されます。
これはロガーのログが上位(ルート)ロガーに伝播しているためです。
propagateをFalseに設定することでログの伝播を防止できます。

    'loggers': {'src': {
        'handlers': ['wsgi'],
        'propagate': False
    }},

dictConfigについて

ロギングの環境設定に記載されているように、dictConfigはロギングの環境設定を行いための関数です。
dictConfigに渡せる辞書は以下の通りです。

キー	必須	意味
version	○	スキーマのバージョン。ここは1でOK
formatters		ログのフォーマットを定義
filters		出力するログなどをフィルタリングする設定
handlers		ログをどこにどう出力するのかといったハンドラの設定
loggers		各ロガーに対してどのレベルのログをどのハンドラーに出力するのかといったことを定義する
root		ルートロガーの設定。propagateが設定できない点を除けばロガーと基本的に同じ

参考：辞書スキーマの詳細

各キーが実際にどのように処理されていくのかについてはソースコードのこの辺りを読むのをお勧めします。

dictConfigClass = DictConfigurator

def dictConfig(config):
    """Configure logging using a dictionary."""
    dictConfigClass(config).configure()

formatters

これはログのフォーマットを設定するスキーマになります。

                formatters = config.get('formatters', EMPTY_DICT)
                for name in formatters:
                    try:
                        formatters[name] = self.configure_formatter(
                                                            formatters[name])

内部ではFormatterインスタンスの生成を行っています。

    def configure_formatter(self, config):
        """Configure a formatter from a dictionary."""
        if '()' in config:
            factory = config['()'] # for use in exception handler
            try:
                result = self.configure_custom(config)
            except TypeError as te:
                if "'format'" not in str(te):
                    raise
                config['fmt'] = config.pop('format')
                config['()'] = factory
                result = self.configure_custom(config)
        else:
            fmt = config.get('format', None)
            dfmt = config.get('datefmt', None)
            style = config.get('style', '%')
            cname = config.get('class', None)
            if not cname:
                c = logging.Formatter
            else:
                c = _resolve(cname)
            result = c(fmt, dfmt, style)
        return result

以下はdefaultフォーマッタの定義です。

    'default': {
        'format': '[%(asctime)s] %(levelname)s in %(module)s:%(lineno)d: %(message)s',
    }

formatではログの属性を以下の形式で出力すると定義しています。
[時刻] ログレベル in モジュール:行番号: メッセージ

使用できるフォーマットの属性名：LogRecord 属性

カスタムフォーマッタ

リクエストの情報の注入で記載されているフォーマッタのサブクラスを定義することができます。

class RequestFormatter(Formatter):
    def format(self, record):
        if has_request_context():
            record.remote_addr = request.remote_addr
            record.method = request.method
            record.url = request.url
        else:
            record.remote_addr = None
            record.method = None
            record.url = None

        return super().format(record)

指定できる属性: flask.Request
ソースコード: werkzeug/src/werkzeug/wrappers/base_request.py

以下のようにdictconfigに記載することで設定完了です。

    'request': {
        '()': RequestFormatter,
        'fmt': '[%(asctime)s] %(remote_addr)s requested %(url)s\n'
               '%(levelname)s in %(module)s: %(message)s'
    }

filters

フィルターはログのフィルターに関する設定を行うことができます。

                filters = config.get('filters', EMPTY_DICT)
                for name in filters:
                    try:
                        filters[name] = self.configure_filter(filters[name])

Filter インスタンスを精製しています。

    def configure_filter(self, config):
        """Configure a filter from a dictionary."""
        if '()' in config:
            result = self.configure_custom(config)
        else:
            name = config.get('name', '')
            result = logging.Filter(name)
        return result

カスタムフィルター

指定したログレベル以下のログを出力するフィルターのサブクラスを定義します。

class LogLevelFilter(Filter):
    def __init__(self, level):
        super().__init__()
        self.__level = level

    def filter(self, record):
        return record.levelno <= self.__level

参考：Pythonのロギング標準ライブラリでファイル出力をログレベル毎にする

以下はle_infoフィルターの定義です。

    'filters': {
        'le_info': {
            '()': LogLevelFilter,
            'level': logging.INFO
        }
    },

()でサブクラスを指定します。
level引数に制限するログレベルを指定します。
これでINFOレベル以下のログのみ出力するようになります。

これはfilters['le_info']=LogLevelFilter(level=logging.INFO)と同等です。

ユーザー定義オブジェクト

カスタムフォーマッタでも記載したように、作成したサブクラスは()で指定できます。
()をキーにconfigure_custom関数が呼び出されます。

        if '()' in config:
            result = self.configure_custom(config)

        c = config.pop('()')
        if not callable(c):
            c = self.resolve(c)
        props = config.pop('.', None)
        # Check for valid identifiers
        kwargs = {k: config[k] for k in config if valid_ident(k)}
        result = c(**kwargs)

詳細についてはユーザー定義オブジェクトを参照ください。

handlers

ハンドラでは、ログの出力先に関する設定を行なっています。

                handlers = config.get('handlers', EMPTY_DICT)
                deferred = []
                for name in sorted(handlers):
                    try:
                        handler = self.configure_handler(handlers[name])
                        handler.name = name
                        handlers[name] = handler

    def configure_handler(self, config):
        """Configure a handler from a dictionary."""
        config_copy = dict(config)  # for restoring in case of error
        formatter = config.pop('formatter', None)
        if formatter:
            try:
                formatter = self.config['formatters'][formatter]
            except Exception as e:
                raise ValueError('Unable to set formatter '
                                 '%r' % formatter) from e
        level = config.pop('level', None)
        filters = config.pop('filters', None)
        if '()' in config:
            c = config.pop('()')
            if not callable(c):
                c = self.resolve(c)
            factory = c
        else:
            cname = config.pop('class')
            klass = self.resolve(cname)
            #Special case for handler which refers to another handler
            if issubclass(klass, logging.handlers.MemoryHandler) and\
                'target' in config:
                try:
                    th = self.config['handlers'][config['target']]
                    if not isinstance(th, logging.Handler):
                        config.update(config_copy)  # restore for deferred cfg
                        raise TypeError('target not configured yet')
                    config['target'] = th
                except Exception as e:
                    raise ValueError('Unable to set target handler '
                                     '%r' % config['target']) from e
            elif issubclass(klass, logging.handlers.SMTPHandler) and\
                'mailhost' in config:
                config['mailhost'] = self.as_tuple(config['mailhost'])
            elif issubclass(klass, logging.handlers.SysLogHandler) and\
                'address' in config:
                config['address'] = self.as_tuple(config['address'])
            factory = klass
        props = config.pop('.', None)
        kwargs = {k: config[k] for k in config if valid_ident(k)}
        try:
            result = factory(**kwargs)
        except TypeError as te:
            if "'stream'" not in str(te):
                raise
            kwargs['strm'] = kwargs.pop('stream')
            result = factory(**kwargs)
        if formatter:
            result.setFormatter(formatter)
        if level is not None:
            result.setLevel(logging._checkLevel(level))
        if filters:
            self.add_filters(result, filters)
        if props:
            for name, value in props.items():
                setattr(result, name, value)
        return result

以下はwsgiハンドラの定義です。

    'wsgi': {
        'class': 'logging.StreamHandler',
        'stream': 'ext://flask.logging.wsgi_errors_stream',
        'formatter': 'default'
    },

classでハンドラのクラスを指定します。
StreamHandlerはストリームに送信するハンドラです。
streamで出力先をflask.logging.wsgi_errors_streamに指定します。
これは、WSGI サーバのエラーストリームまたは標準エラー出力に出力するストリームです。
フォーマッタはformattersで作成したdefaultを使用します。

出力先を標準出力に

StreamHandlerの出力先をext://sys.stdoutにすることで標準出力させることが可能です。

    'console': {
        'class': 'logging.StreamHandler',
        'stream': 'ext://sys.stdout',
        'formatter': 'default'
    },

loggers

こちらは各種ロガーに対する設定を記載する箇所になります。

                loggers = config.get('loggers', EMPTY_DICT)
                for name in loggers:
                    try:
                        self.configure_logger(name, loggers[name], True)
                    except Exception as e:
                        raise ValueError('Unable to configure logger '
                                         '%r' % name) from e

    def configure_logger(self, name, config, incremental=False):
        """Configure a non-root logger from a dictionary."""
        logger = logging.getLogger(name)
        self.common_logger_config(logger, config, incremental)
        propagate = config.get('propagate', None)
        if propagate is not None:
            logger.propagate = propagate

以下はsrcロガーの定義です。

    'src': {
        'level': 'INFO',
        'handlers': ['wsgi'],
        'propagate': False
    }

ハンドラにwsgiを指定しています。
上位のハンドラに伝播させない=ルートロガーへの出力を無効に設定するために、
propagateをFalseにしています。

インポートしたパッケージのログ出力

Flask-OAuthlibをインポートしており、
そのログを出力したい場合はflask_oauthlibをloggersに追加します。

    'flask_oauthlib': {
        'level': 'DEBUG'
    }

flask-oauthlib/flask_oauthlib/client.pyで設定されているようにロガー名がflask_oauthlibなので、ログを出力することができるようになります。

log = logging.getLogger('flask_oauthlib')

root

こちらはルートロガーに対する設定です。

                root = config.get('root', None)
                if root:
                    try:
                        self.configure_root(root, True)
                    except Exception as e:
                        raise ValueError('Unable to configure root '
                                         'logger') from e

    def configure_root(self, config, incremental=False):
        """Configure a root logger from a dictionary."""
        root = logging.getLogger()
        self.common_logger_config(root, config, incremental)

ルートは最上位のロガーなため、上位ロガーへの伝播の有無を決定するpropagateがないこと以外はルートロガーと違いはありません。
以下の例の場合は、DEBUGレベル以上のログをconsoleとwsgiハンドラに出力しています。

    {
        'level': 'DEBUG',
        'handlers': ['console', 'wsgi']
    }

ログレベル毎に標準出力と標準エラー出力に出し分け

以下は指定したログレベルより上のレベルのログを除外するフィルタークラスです。

class LogLevelFilter(Filter):
    def __init__(self, level):
        super().__init__()
        self.__level = level

    def filter(self, record):
        return record.levelno <= self.__level

INFOレベル以下のログのみ取得するフィルターを作成します。

    'filters': {
        'le_filter': {
            '()': LogLevelFilter,
            'level': logging.INFO
        }
    },

ハンドラにフィルターを追加することで適用されます。
また標準エラー出力用のハンドラにWARNING以上のログのみを出力するためにログレベルを設定します

    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'stream': 'ext://sys.stdout',
            'formatter': 'default',
            'filters': ['le_filter']
        },
        'wsgi': {
            'class': 'logging.StreamHandler',
            'stream': 'ext://flask.logging.wsgi_errors_stream',
            'level': 'WARNING',
            'formatter': 'default'
        },
    },
    'root': {
        'level': 'DEBUG',
        'handlers': ['console', 'wsgi']
    }

これで標準出力にはINFOログレベル以下のログが出力され、標準エラー出力にはWARNINGログレベル以上のログが出力されるようになります。