risclog.logging 1.2.1

创建记录器

要创建记录器，请使用get_logger函数。此函数确保您获得一个正确配置的RiscLogger实例。

from risclog.logging import get_logger

# create logger
logger = get_logger(name='my_logger')

记录器配置

记录器配置在创建记录器实例时自动进行。_configure_logger方法设置structlog和logging以提供带时间戳、上下文变量和格式的日志。您可以根据需要自定义配置。

该模块配置为自动从环境变量中读取日志级别。默认情况下，级别设置为INFO。要调整此设置，请设置LOG_LEVEL环境变量

export LOG_LEVEL=DEBUG

使用以下方法以不同的日志级别记录消息

• 调试消息：logger.debug("这是一条调试消息")

• 信息消息：logger.info("这是一条信息消息")

• 警告消息：logger.warning("这是一条警告消息")

• 错误消息：logger.error("这是一条错误消息")

• 关键消息：logger.critical("这是一条关键消息")

• 致命消息：logger.fatal("这是一条致命消息")

• 异常消息：logger.exception("这是一条异常消息")

异步和同步日志消息

risclog.logging包支持同步和异步日志消息。如果您在异步环境中工作，请使用日志方法的异步版本

• 异步调试消息：await logger.debug("异步调试消息")

• 异步信息消息：await logger.info("异步信息消息")

• 等等…

日志装饰器

装饰器装饰器可用于为方法提供自动日志记录和可选的异常电子邮件通知

from risclog.logging import get_logger

logger = get_logger(name='my_logger')

@logger.decorator(send_email=True)
async def some_async_function(x, y):
    return x + y

from risclog.logging import get_logger

logger = get_logger(name='my_logger')

@logger.decorator
def some_sync_function(x, y):
    return x + y

错误处理和电子邮件通知

如果将send_email参数设置为True，则在发生异常时将自动发送电子邮件通知。电子邮件将通过ThreadPoolExecutor异步发送，并包含异常详细信息。

要能够发送电子邮件，必须设置以下环境变量！

• ‘logging_email_smtp_user’

• ‘logging_email_smtp_password’

• ‘logging_email_to’

• ‘logging_email_smtp_server’

示例

以下是一个完整示例，展示如何在应用程序中使用risclog.logging包

import os
import asyncio
from risclog.logging import get_logger


os.environ["LOG_LEVEL"] = "DEBUG"

logger = get_logger("async_debug_example")


@logger.decorator(send_email=True)
async def fetch_data(url: str):
    await logger.debug(f"Start retrieving data from  {url}")
    await asyncio.sleep(2)  # Simulates a delay, such as a network request
    await logger.debug(f"Successfully retrieved data from {url}")
    return {"data": f"Sample data from {url}"}


@logger.decorator
async def main():
    url = "https://example.com"
    await logger.debug(f"Start main function with URL: {url}")
    data = await fetch_data(url)
    await logger.debug(f"Data received: {data}")


if __name__ == "__main__":
    logger.info("Start main function")
    asyncio.run(main())

输出

2024-08-05 11:38:51 [info     ] [async_debug_example] __id=4378622064 __sender=inline message=Start main function
2024-08-05 11:38:51 [info     ] [async_debug_example] __id=4384943584 __sender=async_logging_decorator _function=main _script=example.py message=Method "main" called with no arguments.
2024-08-05 11:38:51 [debug    ] [async_debug_example] __id=4378552584 __sender=inline _function=main _script=example.py message=Start main function with URL: https://example.com
2024-08-05 11:38:51 [info     ] [async_debug_example] __id=4384943744 __sender=async_logging_decorator _function=fetch_data _script=example.py message=Method called: "fetch_data" with: "{'arg_0': 'https://example.com'}"
2024-08-05 11:38:51 [debug    ] [async_debug_example] __id=4366292144 __sender=inline _function=fetch_data _script=example.py message=Start retrieving data from  https://example.com
2024-08-05 11:38:53 [debug    ] [async_debug_example] __id=4366292144 __sender=inline _function=fetch_data _script=example.py message=Successfully retrieved data from https://example.com
2024-08-05 11:38:53 [info     ] [async_debug_example] __id=4384943744 __sender=async_logging_decorator _function=fetch_data _script=example.py message=Method "fetch_data" returned: "{'data': 'Sample data from https://example.com'}"
2024-08-05 11:38:53 [debug    ] [async_debug_example] __id=4378552584 __sender=inline message=Data received: {'data': 'Sample data from https://example.com'}
2024-08-05 11:38:53 [info     ] [async_debug_example] __id=4384943584 __sender=async_logging_decorator message=Method "main" returned: "None"

运行测试

$ ./pytest