使用 Watchtower 进行 Python 快速 API 日志记录

问题描述 投票:0回答:1

我编写了下面的代码,使用瞭望塔将日志写入cloudwatch。

import os
import sys
import time
import boto3
import watchtower
import logging.handlers
from scripts.config import app_configurations

def fast_logging():
    try:
        boto3_session = boto3.session.Session()
        LOG_GROUP = "Fast-Logging"
        log_level = DEBUG
        stream_name = os.path.join("fast_logging"+ "_" + time.strftime("%Y%m%d") + '.log')

        logger = logging.getLogger("Fast-Logger")
        logger.setLevel(log_level)

        formatter = logging.Formatter('%(name)s - %(levelname)s - %(filename)s - %(module)s: %(funcName)s: '
                                      '%(lineno)d - %(message)s')
        log_handler = watchtower.CloudWatchLogHandler(log_group=LOG_GROUP, boto3_session=boto3_session,
                                                      stream_name=stream_name)
        log_handler.setFormatter(formatter)
        logger.addHandler(log_handler)
        return logger
    except Exception as e:
        raise e


logger = fast_logging()

上面的代码适用于普通的Python代码,但无法将Fast API服务中的日志转储到cloudwatch流中。

python fastapi amazon-cloudwatchlogs docker-watchtower watchtower
1个回答
0
投票

我发现这段代码对我有用

import os
import time
import boto3
import watchtower
import logging.handlers

def fast_logging():
    try:
        LOG_GROUP = "Fast-Logging"
        log_level = "INFO"
        stream_name = os.path.join("fast_logging"+ "_" + time.strftime("%Y%m%d") + '.log')

        logger = logging.getLogger("Fast-Logger")
        logger.setLevel(log_level)

        formatter = logging.Formatter('%(name)s - %(levelname)s - %(filename)s - %(module)s: %(funcName)s: '
                                      '%(lineno)d - %(message)s')
        log_handler = watchtower.CloudWatchLogHandler(log_group=LOG_GROUP,
                                                      stream_name=stream_name)
        log_handler.setFormatter(formatter)
        logger.addHandler(log_handler)
        return logger
    except Exception as e:
        raise e


logger = fast_logging()
logger.info("test this")
© www.soinside.com 2019 - 2024. All rights reserved.