从模块外部异步动态更改日志级别

我需要动态地更改日志级别而不重新加载任何模块。为此,我将使用asyncio和多处理的组合(对于我正在编写的更大的程序都是这样)。

日志_配置.py

def setupLogging(log_level=None): # load custom logger
    with open(os.path.join(LOG_PATH,'log_config.yaml'), 'rt') as file_:
        config = yaml.safe_load(file_.read())
        logging.config.dictConfig(config)
        logging.Formatter.converter = time.gmtime

    if not log_level:
        log_level = 'preview'
    
    return logging.getLogger(log_level)

logger=setupLogging() # global var that needs dynamic updation
async def logger_changes(): # socket listener that changes the logger object
    global logger
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.bind(('localhost', 9999))
    sock.listen()
    sock.setblocking(0)
    while True:
        try:
            conn, addr = sock.accept()
            data = conn.recv(1024)
            conn.send(data)
            logger = setupLogging(data.decode('utf8'))
        except Exception as e:
            pass
        await asyncio.sleep(5)

async def logger_handler():
    t1 = asyncio.create_task(logger_changes())
    await t1

def start_logger():
    start_func = asyncio.run(logger_handler())

pLOGGER = Process(name="__startLOGGER__", target=start_logger, daemon=False)
pLOGGER.start()

主要功能

import log_config
logger=log_config.logger

async def core_func():
    for i in range(10):
        logger.debug("sample debug log")
        logger.info("sample info log")
        logger.warning("Watch out!")
        logger.error("Heading for trouble!")
        logger.critical("Seriously, do something!")
        print("#"*80)

        await asyncio.sleep(5)

async def core_func_parent():
    t1 = asyncio.create_task(core_func())
    await t1

def core_func_handler():
    start_func = asyncio.run(core_func_parent())

if __name__=='__main__':
    pMODULE = Process(name="__startMODULE__", target=core_func_handler, daemon=False)
    pMODULE.start()

触发日志级别更改

import socket

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.connect(('localhost', 9999))
request = "test"

try:
    server.send(request.encode('utf8'))
    response = server.recv(255).decode('utf8')
    print(response)
except Exception as e:
    print(e)
    server.close()

步骤

1. 运行python脚本的主要功能。从内部自动调用log\u config,启动套接字侦听器并实例化 logger 对象
2. 运行trigger log level change,将消息发送到具有更新的 记录器 设置

挑战

一旦我运行步骤1,日志记录级别是 INFO (这正是我想要的)

但是一旦我运行了第2步,我就希望日志级别变为 DEBUG 这是不可能的。

感谢您的帮助。