aiohttp Logging: From Basic Setup to Advanced Methods

In this tutorial, you’ll learn how to set up and customize logging in your aiohttp projects.

You’ll learn how to configure loggers, customize log formats, track HTTP requests and responses, measure performance, implement log rotation, and more.

 

 

Configuring aiohttp Logger

To configure the aiohttp logger, use the Python built-in logging module.

Here’s how to set up basic logging:

import logging
import aiohttp
import asyncio
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger('aiohttp')
async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get('https://api.github.com') as resp:
            print(f"Status: {resp.status}")
asyncio.run(main())

Output:

DEBUG:asyncio:Using proactor: IocpProactor
Status: 200

You can customize the log format to include specific information:

import logging
import aiohttp
import asyncio
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger('aiohttp')
async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get('https://api.github.com') as resp:
            print(f"Status: {resp.status}")
asyncio.run(main())

Output:

2024-09-09 10:01:34,136 - asyncio - DEBUG - Using proactor: IocpProactor
Status: 200

This output includes timestamps, logger names, and log levels in addition to the log messages.

 

Logging HTTP Requests and Responses

Log Request details

You can also create a custom logger class to log request details.

import logging
from aiohttp import web
from datetime import datetime
class RequestLogger:
  def __init__(self):
    self.logger = logging.getLogger(__name__)
    logging.basicConfig(level=logging.INFO)

  async def log_request(self, request):
    self.start_time = datetime.now()
    self.logger.info(f"Request started: {request.method} {request.path}")
    self.logger.info(f"Headers: {dict(request.headers)}")
app = web.Application()
request_logger = RequestLogger()
async def hello(request):
  await request_logger.log_request(request)
  response = web.Response(text="Hello, World!")
  return response
app.add_routes([web.get('/', hello)])
if __name__ == '__main__':
  web.run_app(app)

Output:

INFO:__main__:Request started: GET /
INFO:__main__:Headers: {'Host': '127.0.0.1:8080', 'Connection': 'keep-alive', 'sec-ch-ua': '"Chromium";v="128", "Not;A=Brand";v="24", "Microsoft Edge";v="128"', 'sec-ch-ua-mobile': '?0', 'sec-ch-ua-platform': '"Windows"', 'Upgrade-Insecure-Requests': '1', 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36 Edg/128.0.0.0', 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7', 'Sec-Fetch-Site': 'none', 'Sec-Fetch-Mode': 'navigate', 'Sec-Fetch-User': '?1', 'Sec-Fetch-Dest': 'document', 'Accept-Encoding': 'gzip, deflate, br, zstd', 'Accept-Language': 'en-US,en;q=0.9,ar;q=0.8'}
INFO:aiohttp.access:127.0.0.1 [09/Sep/2024:09:10:39 +0200] "GET / HTTP/1.1" 200 166 "-" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36 Edg/128.0.0.0"

This code logs the request method, URL, and headers before sending the request.

Response details (status code, body size)

To log response details, modify the custom ClientSession:

import logging
from aiohttp import web
from datetime import datetime
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@web.middleware
async def response_logger_middleware(request, handler):
  response = await handler(request) 
  logger.info(f"Response: {request.method} {request.path} - Status: {response.status}")
  logger.info(f"Response Headers: {dict(response.headers)}")
  
  # Log 200 characters of the body
  logger.info(f"Response Body: {str(response.body)[:200]}...")  
  return response
app = web.Application(middlewares=[response_logger_middleware])
async def hello(request):
  return web.Response(text="Hello, World!")
app.add_routes([web.get('/', hello)])
if __name__ == '__main__':
  web.run_app(app)

Output:

INFO:__main__:Response: GET / - Status: 200
INFO:__main__:Response Headers: {'Content-Type': 'text/plain; charset=utf-8'}
INFO:__main__:Response Body: b'Hello, World!'...
INFO:aiohttp.access:127.0.0.1 [09/Sep/2024:09:14:09 +0200] "GET / HTTP/1.1" 200 166 "-" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36 Edg/128.0.0.0"

This code logs response details including the response status code and body content.

Timing information

To log timing information, you can set up a timing middleware that logs the start of each request and the time taken to process it:

from aiohttp import web
import logging
import time
import asyncio
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@web.middleware
async def timing_middleware(request, handler):
  start_time = time.time()
  logger.info(f"Request started: {request.method} {request.path}")  
  response = await handler(request)  
  elapsed_time = time.time() - start_time
  logger.info(f"Response: Status {response.status}, Time {elapsed_time:.4f}s")  
  return response

async def hello(request):
  return web.Response(text="Hello, World!")

async def slow(request):
  await asyncio.sleep(2)
  return web.Response(text="This was a slow response")

app = web.Application(middlewares=[timing_middleware])
app.router.add_get('/', hello)
app.router.add_get('/slow', slow)
if __name__ == '__main__':
  web.run_app(app, host='localhost', port=8080)

Output:

INFO:__main__:Request started: GET /
INFO:__main__:Response: Status 200, Time 0.0000s
INFO:aiohttp.access:127.0.0.1 [09/Sep/2024:09:26:43 +0200] "GET / HTTP/1.1" 200 166 "-" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36 Edg/128.0.0.0"
INFO:__main__:Request started: GET /slow
INFO:__main__:Response: Status 200, Time 2.0014s
INFO:aiohttp.access:127.0.0.1 [09/Sep/2024:09:26:46 +0200] "GET /slow HTTP/1.1" 200 177 "-" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36 Edg/128.0.0.0"

http://localhost:8080. This will trigger the fast route.

Then navigate to http://localhost:8080/slow. This will trigger the slow route.

 

Log Resource Usage (CPU, Memory)

To log resource usage, use the psutil library:

import logging
import aiohttp
from aiohttp import web
import psutil
import os
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('aiohttp.server')

@web.middleware
async def resource_usage_middleware(request, handler):
    process = psutil.Process(os.getpid())
    start_cpu = process.cpu_percent()
    start_memory = process.memory_info().rss / 1024 / 1024  # MB
    response = await handler(request)
    end_cpu = process.cpu_percent()
    end_memory = process.memory_info().rss / 1024 / 1024  # MB
    logger.info(f"Request to {request.path}: CPU {end_cpu - start_cpu:.2f}%, Memory {end_memory - start_memory:.2f}MB")
    return response
async def hello(request):
    return web.Response(text="Hello, World!")
app = web.Application(middlewares=[resource_usage_middleware])
app.router.add_get('/', hello)
if __name__ == '__main__':
    web.run_app(app)

To test this, run the server and go to http://localhost:8080:

Output:

INFO:aiohttp.server:Request to /: CPU 0.00%, Memory 0.04MB
INFO:aiohttp.access:127.0.0.1 [09/Sep/2024:09:36:22 +0200] "GET / HTTP/1.1" 200 166 "-" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36 Edg/128.0.0.0"

This middleware logs the CPU and memory usage for each request so you can identify which route consumes more resources.

 

Rotate Log Files

Time-based rotation

To implement time-based log rotation, use the TimedRotatingFileHandler:

import logging
from logging.handlers import TimedRotatingFileHandler
import aiohttp
from aiohttp import web
import traceback
import time
logger = logging.getLogger('aiohttp.server')
logger.setLevel(logging.DEBUG)
handler = TimedRotatingFileHandler('aiohttp_server.log', when='midnight', interval=1, backupCount=7)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
handler.setLevel(logging.DEBUG)
logger.addHandler(handler)
logging.getLogger('aiohttp').setLevel(logging.DEBUG)

async def hello(request):
    logger.debug("Received request to /")
    return web.Response(text="Hello, World!")

@web.middleware
async def logging_middleware(request, handler):
    start_time = time.time()
    logger.info(f"Request: {request.method} {request.path} - Headers: {dict(request.headers)}")
    
    try:
        response = await handler(request)
        return response
    except web.HTTPException as ex:
        response = ex
        return response
    except Exception as ex:
        logger.error(f"Unexpected error occurred: {ex}")
        logger.error(traceback.format_exc())
        response = web.Response(status=500, text="Internal Server Error")
        return response
    finally:
        duration = time.time() - start_time
        status = getattr(response, 'status', None)
        logger.info(f"Response: Status {status} - Duration: {duration:.4f}s - Headers: {dict(response.headers)}")
app = web.Application(middlewares=[logging_middleware])
app.router.add_get('/', hello)
if __name__ == '__main__':
    web.run_app(app)

This code sets up a TimedRotatingFileHandler that rotates the log file daily at midnight and keeps 7 backup files.

Size-based rotation

For size-based rotation, use the RotatingFileHandler:

import logging
from logging.handlers import RotatingFileHandler
import aiohttp
from aiohttp import web
import traceback
import time
logger = logging.getLogger('aiohttp.server')
logger.setLevel(logging.DEBUG)
handler = RotatingFileHandler('aiohttp_server.log', maxBytes=10*1024*1024, backupCount=5)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
handler.setLevel(logging.DEBUG)
logger.addHandler(handler)
logging.getLogger('aiohttp').setLevel(logging.DEBUG)

async def hello(request):
  logger.debug("Received request to /")
  return web.Response(text="Hello, World!")

@web.middleware
async def logging_middleware(request, handler):
  start_time = time.time()  
  logger.info(f"Request: {request.method} {request.path} - Headers: {dict(request.headers)}")
  
  try:
      response = await handler(request)
      return response
  except web.HTTPException as ex:
      response = ex
      return response
  except Exception as ex:
      logger.error(f"Unexpected error occurred: {ex}")
      logger.error(traceback.format_exc())
      response = web.Response(status=500, text="Internal Server Error")
      return response
  finally:
      duration = time.time() - start_time
      status = getattr(response, 'status', None)
      logger.info(f"Response: Status {status} - Duration: {duration:.4f}s - Headers: {dict(response.headers)}")
app = web.Application(middlewares=[logging_middleware])
app.router.add_get('/', hello)
if __name__ == '__main__':
  web.run_app(app)

This code sets up a RotatingFileHandler that rotates the log file.

it creates a main log file named aiohttp_server.log.

When this file reaches 10 MB, it will be renamed to aiohttp_server.log.1, a new aiohttp_server.log file will be created for new log entries.

This process continues, creating aiohttp_server.log.2, aiohttp_server.log.3, etc., up to aiohttp_server.log.5.

When all backup files are full, the oldest one (aiohttp_server.log.5) will be deleted to make room for a new backup.