Advanced Python Logging: StreamHandler, FileHandler, HTTPHandler and Asynchronous Remote Logging with aiohttp and Thread Pools

In Python, the most common way to log is to output to the console and a file using the logging module; however, sometimes logs need to be sent to a remote server or a database.

StreamHandler and FileHandler

The following script creates a logger, sets its level to DEBUG , defines a formatter, and adds both a StreamHandler (to stdout) and a FileHandler (to debug.log ) before emitting a test message:

# -*- coding: utf-8 -*-
"""
-------------------------------------------------
 File Name：   loger
 Description :
 Author :    yangyanxing
 date：     2020/9/23
-------------------------------------------------
"""
import logging
import sys
import os
# 初始化logger
logger = logging.getLogger("yyx")
logger.setLevel(logging.DEBUG)
# 设置日志格式
fmt = logging.Formatter('[%(asctime)s] [%(levelname)s] %(message)s', '%Y-%m-%d\n%H:%M:%S')
# 添加cmd handler
cmd_handler = logging.StreamHandler(sys.stdout)
cmd_handler.setLevel(logging.DEBUG)
cmd_handler.setFormatter(fmt)
# 添加文件的handler
logpath = os.path.join(os.getcwd(), 'debug.log')
file_handler = logging.FileHandler(logpath)
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(fmt)
# 将cmd和file handler添加到logger中
logger.addHandler(cmd_handler)
logger.addHandler(file_handler)
logger.debug("今天天气不错")

The message appears on the console and is written to debug.log .

Adding an HTTPHandler

To send logs to a remote server, an HTTPHandler can be attached. The example below posts the formatted log to 127.0.0.1:1987/api/log/get :

import logging.handlers
http_handler = logging.handlers.HTTPHandler(r"127.0.0.1:1987", '/api/log/get')
http_handler.setLevel(logging.DEBUG)
http_handler.setFormatter(fmt)
logger.addHandler(http_handler)
logger.debug("今天天气不错")

The server receives a JSON payload containing many fields, but often only the formatted message is needed. The desired output looks like:

[2020-09-23 10:45:56][DEBUG] 今天天气不错

To control the payload, a custom handler can be created by subclassing logging.Handler and overriding emit .

class CustomHandler(logging.Handler):
    def __init__(self, host, uri, method="POST"):
        logging.Handler.__init__(self)
        self.url = "%s/%s" % (host, uri)
        method = method.upper()
        if method not in ["GET", "POST"]:
            raise ValueError("method must be GET or POST")
        self.method = method
    def emit(self, record):
        '''Re‑format the log record before sending.'''
        msg = self.format(record)
        # send using requests (synchronous) or later with aiohttp (asynchronous)
        ...

Asynchronous Remote Logging

When the remote endpoint is slow, synchronous requests blocks the script. Converting the handler to use asyncio and aiohttp removes the bottleneck. The emit method becomes asynchronous and registers its coroutine with the event loop.

class CustomHandler(logging.Handler):
    def __init__(self, host, uri, method="POST"):
        ...
    async def emit(self, record):
        msg = self.format(record)
        timeout = aiohttp.ClientTimeout(total=6)
        if self.method == "GET":
            sep = '&' if '?' in self.url else '?'
            url = self.url + "%c%s" % (sep, urllib.parse.urlencode({"log": msg}))
            async with aiohttp.ClientSession(timeout=timeout) as session:
                async with session.get(url) as resp:
                    print(await resp.text())
        else:
            headers = {"Content-type": "application/x-www-form-urlencoded",
                       "Content-length": str(len(msg))}
            async with aiohttp.ClientSession(timeout=timeout, headers=headers) as session:
                async with session.post(self.url, data={"log": msg}) as resp:
                    print(await resp.text())
        return True

Because emit now returns a coroutine, it must be scheduled on the loop, e.g. loop.create_task(self.submit(msg)) . The script creates a global loop with loop = asyncio.get_event_loop() and finally calls loop.run_forever() , stopping it from a task callback when all logs are sent.

Thread‑Based Solutions

Before async, a common approach is to offload the HTTP request to a separate thread. The handler can start a new threading.Thread for each log, or use a ThreadPoolExecutor to limit the number of concurrent threads and preserve order by using a single‑worker pool.

exector = ThreadPoolExecutor(max_workers=1)
def emit(self, record):
    msg = self.format(record)
    if self.method == "GET":
        url = ...
        exector.submit(requests.get, url, timeout=6)
    else:
        headers = {...}
        exector.submit(requests.post, self.url, data={"log": msg}, headers=headers, timeout=6)

Using a single‑worker pool guarantees that logs are sent in the order they were generated.

Putting It All Together

The final example combines the async submit helper, the custom handler, and the event loop to achieve non‑blocking remote logging:

loop = asyncio.get_event_loop()
class CustomHandler(logging.Handler):
    def __init__(self, host, uri, method="POST"):
        ...
    async def submit(self, data):
        timeout = aiohttp.ClientTimeout(total=6)
        if self.method == "GET":
            ...
        else:
            ...
        return True
    def emit(self, record):
        msg = self.format(record)
        loop.create_task(self.submit(msg))

http_handler = CustomHandler(r"http://127.0.0.1:1987", 'api/log/get')
http_handler.setLevel(logging.DEBUG)
http_handler.setFormatter(fmt)
logger.addHandler(http_handler)
logger.debug("今天天气不错")
logger.debug("是风和日丽的")
loop.run_forever()

With this setup, log messages are formatted, queued, and sent asynchronously without slowing down the main application.