Aiogram 3 log messages and handlers

2.README-GIST.md

Aiogram 3 log messages and handlers

Warning

Tested on aiogram=3.3.0, code may failed with another version as it uses tricky things

Features

Automatically log every action from user

Each user action will be displayed in a log stream with the appropriate type (Message, CallbackQuery and so on).

Link to handler source code

If you use the standard aiogram syntax, then a link to the handler source code will appear in the logs. It's clickable in PyCharm! Jump to source in no second.

Answer user if event is not handled

Bot will send "dunno" message if the action was not processed in any way by any handler.

Colored log

-	color
Timestamp	black
Levelname	corresponding color
Clickable source link	cyan

Gist checklist

0. Add colorlog dependency to your project.
1. Insert logging.yaml andlogging_.py into your project and import it before routes definition.
2. Replace default Dispatcher class with CustomDispatcher from gist dispatcher.py.
3. Register LogAllEventsMiddleware from logging_middleware.py as message and callback_query middlewares.
4. Done!

Note

Use logging_.logger to log anything from your code

License

MIT License

Copyright (c) 2024 Ruslan Belkov

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

dispatcher.py

from typing import Any

from aiogram import Dispatcher, Bot
from aiogram.dispatcher.event.bases import UNHANDLED
from aiogram.types import Update, User, Message, CallbackQuery

from src.logging_ import logger


# noinspection PyMethodMayBeStatic
class CustomDispatcher(Dispatcher):
    async def _send_dunno_message(self, bot: Bot, chat_id: int):
        await bot.send_message(
            chat_id,
            "⚡️ I don't understand you. Please, use /start command.",
        )

    async def _listen_update(self, update: Update, **kwargs) -> Any:
        res = await super()._listen_update(update, **kwargs)
        if res is UNHANDLED:
            bot: Bot = kwargs.get("bot")
            event_from_user: User = kwargs.get("event_from_user")
            username = event_from_user.username
            user_string = f"User @{username}<{event_from_user.id}>" if username else f"User <{event_from_user.id}>"
            event = update.event
            event_type = type(event).__name__

            if isinstance(event, Message):
                message_text = f"{event.text[:50]}..." if len(event.text) > 50 else event.text
                msg = f"{user_string}: [{event_type}] `{message_text}`"
            elif isinstance(event, CallbackQuery):
                msg = f"{user_string}: [{event_type}] `{event.data}`"
            else:
                msg = f"{user_string}: [{event_type}]"

            logger.info(f"Unknown event from user. {msg}")
            await self._send_dunno_message(bot, event_from_user.id)
        return res

example.py

from aiogram import Bot
from aiogram.fsm.storage.memory import MemoryStorage

from logging_ import logger # noqa: F401
from logging_middleware import LogAllEventsMiddleware
from my_routes import router
from dispatcher import CustomDispatcher

async def main():
    bot = Bot("YOUR_BOT_TOKEN", parse_mode="HTML")
    dp = CustomDispatcher(storage=MemoryStorage())
    # Setup middleware
    log_all_middleware = LogAllEventsMiddleware()
    dp.message.middleware(log_all_middleware)
    dp.callback_query.middleware(log_all_middleware)
    # Setup routes
    dp.include_router(router)
    # Drop pending updates
    await bot.delete_webhook(drop_pending_updates=True)
    # Start long-polling
    await dp.start_polling(bot, allowed_updates=dp.resolve_used_update_types())

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        logger.info("Bot stopped!")

logging.yaml

version: 1
disable_existing_loggers: False
formatters:
  src:
    "()": colorlog.ColoredFormatter
    format: '[%(black)s%(asctime)s%(reset)s] [%(log_color)s%(levelname)s%(reset)s] [%(cyan)sFile "%(relativePath)s", line %(lineno)d%(reset)s] %(message)s'
  default:
    "()": colorlog.ColoredFormatter
    format: '[%(black)s%(asctime)s%(reset)s] [%(log_color)s%(levelname)s%(reset)s] [%(name)s] %(message)s'
handlers:
  src:
    formatter: src
    class: logging.StreamHandler
    stream: ext://sys.stdout
  default:
    formatter: default
    class: logging.StreamHandler
    stream: ext://sys.stdout
loggers:
  src:
    level: INFO
    handlers:
      - src
    propagate: no
  aiogram:
    level: INFO
    handlers:
      - default
    propagate: no
  aiogram.event:
    level: WARNING

logging_.py

__all__ = ["logger"]

import logging.config
import os

import yaml


class RelativePathFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.relativePath = os.path.relpath(record.pathname)
        return True


with open("logging.yaml", "r") as f:
    config = yaml.safe_load(f)
    logging.config.dictConfig(config)

logger = logging.getLogger("src")
logger.addFilter(RelativePathFilter())

logging_middleware.py

from logging_ import logger

# noinspection PyMethodMayBeStatic
class LogAllEventsMiddleware(BaseMiddleware):
    async def __call__(
        self,
        handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]],
        event: TelegramObject,
        data: Dict[str, Any],
    ) -> Any:
        loop = asyncio.get_running_loop()
        start_time = loop.time()
        r = await handler(event, data)
        finish_time = loop.time()
        duration = finish_time - start_time
        try:
            # get to `aiogram.dispatcher.event.TelegramEventObserver.trigger` method
            frame = inspect.currentframe()
            frame_info = inspect.getframeinfo(frame)
            while frame is not None:
                if frame_info.function == "trigger":
                    _handler = frame.f_locals.get("handler")
                    if _handler is not None:
                        _handler: HandlerObject
                        record = self._create_log_record(_handler, event, data, duration=duration)
                        logger.handle(record)
                    break
                frame = frame.f_back
                frame_info = inspect.getframeinfo(frame)
        finally:
            del frame
        return r

    def _create_log_record(
        self, handler: HandlerObject, event: TelegramObject, data: Dict[str, Any], *, duration: Optional[float] = None
    ) -> logging.LogRecord:
        callback = handler.callback
        func_name = callback.__name__
        pathname = inspect.getsourcefile(callback)
        lineno = inspect.getsourcelines(callback)[1]

        event_type = type(event).__name__
        username = event.from_user.username
        user_string = f"User @{username}<{event.from_user.id}>" if username else f"User <{event.from_user.id}>"

        if isinstance(event, Message):
            message_text = f"{event.text[:50]}..." if len(event.text) > 50 else event.text
            msg = f"{user_string}: [{event_type}] `{message_text}`"
        elif isinstance(event, CallbackQuery):
            msg = f"{user_string}: [{event_type}] `{event.data}`"
        else:
            msg = f"{user_string}: [{event_type}]"

        if duration is not None:
            msg = f"Handler `{func_name}` took {int(duration * 1000)} ms: {msg}"

        record = logging.LogRecord(
            name="src.bot.middlewares.LogAllEventsMiddleware",
            level=logging.INFO,
            pathname=pathname,
            lineno=lineno,
            msg=msg,
            args=(),
            exc_info=None,
            func=func_name,
        )
        record.relativePath = os.path.relpath(record.pathname)
        return record