Настройка логирования в FastAPI проекте

Создание утилит для логирования и управления жизненным циклом

---

Зачем нужно логирование

Логирование — это запись всех событий, происходящих в вашем приложении. Это критически важный инструмент, который помогает:

• Отлаживать ошибки — когда приложение падает, логи показывают, что произошло
• Отслеживать работу — вы видите, какие запросы приходят, сколько времени они выполняются
• Анализировать поведение пользователей — кто, когда и что делал на сайте
• Выявлять проблемы до того, как о них сообщат пользователи

В этом проекте мы создадим гибкую систему логирования, которая будет работать как в разработке (вывод в консоль), так и в продакшене (запись в файлы).

---

Структура файлов логирования

Мы создадим две утилиты:

• app/utils/log.py — основная утилита для логирования
• app/utils/lifespan.py — управление жизненным циклом приложения (запуск и остановка)

Лог-файлы будут храниться в папке log/ со структурой по годам и месяцам:

                log/
                └── 2026/
                    └── 08/
                        ├── 04.log
                        ├── 05.log
                        └── 06.log

Каждый день создаётся отдельный файл — это упрощает поиск и архивацию.

---

Утилита log.py

Начнём с создания файла app/utils/log.py. Эта утилита будет отвечать за запись логов как в файлы, так и в консоль.

Библиотека aiologger уже добавлена в requirements.txt в прошлых статьях. Все зависимости уже установлены в виртуальном окружении.

Полный код log.py

Создайте файл app/utils/log.py и добавьте следующий код:

                import os
import datetime
from aiologger import Logger
from aiologger.handlers.files import AsyncFileHandler
import logging

class Log:
    def __init__(self):
        self.log_dir = "log"
        os.makedirs(self.log_dir, exist_ok=True)
        self.handlers = {}
        self.log_print = os.getenv("LOG_PRINT", "1").lower() in ("1", "true", "yes")

    def build_log_path(self, target: str, now: datetime.datetime) -> str:
        """Формируем путь к лог-файлу: log/2025/10/04.log"""
        year = f"{now.year}"
        month = f"{now:%m}"
        day = f"{now:%d}"

        base_dir = os.path.join(self.log_dir, year, month)
        os.makedirs(base_dir, exist_ok=True)

        filename = f"{day}.log"
        return os.path.join(base_dir, filename)

    async def get_logger(self, target: str, now: datetime.datetime) -> Logger:
        """Асинхронный логгер для target."""
        log_path = self.build_log_path(target, now)

        if target not in self.handlers or self.handlers[target]["path"] != log_path:
            handler = AsyncFileHandler(filename=log_path, mode="a", encoding="utf-8")
            target_logger = Logger(name=f"logger_{target}")
            target_logger.add_handler(handler)

            if target in self.handlers:
                try:
                    await self.handlers[target]["logger"].shutdown()
                except Exception:
                    pass

            self.handlers[target] = {
                "path": log_path,
                "logger": target_logger,
                "handler": handler,
            }

        return self.handlers[target]["logger"]

    async def unescape_newlines(self, obj):
        """Рекурсивно заменяет '\\n' на реальные переносы."""
        if isinstance(obj, dict):
            return {k: await self.unescape_newlines(v) for k, v in obj.items()}
        elif isinstance(obj, list):
            return [await self.unescape_newlines(x) for x in obj]
        elif isinstance(obj, str):
            return obj.replace("\\n", "\n")
        else:
            return obj

    async def log_info(
        self,
        target: str = "",
        message: str = "",
        data: dict | None = None,
        is_console: bool = None,
    ):
        """Асинхронная запись информационного сообщения."""
        if data is None:
            data = {}

        now = datetime.datetime.now()

        log_data_str = f"{now:%d.%m.%Y %H:%M:%S} {target}: {message}"
        if data:
            log_data_str += f": {self.safe_serialize(data)}"

        target_logger = await self.get_logger(target, now)
        await target_logger.info(log_data_str)

        should_print = self.log_print if is_console is None else is_console
        if should_print:
            print(log_data_str)

    async def log_error(
        self, target: str = "", message: str = "", data: dict | None = None, is_console: bool = True
    ):
        """Асинхронная запись ошибки."""
        await self.log_info(target, f"ERROR: {message}", data, is_console)

    def log_info_sync(
        self, target: str = "", message: str = "", data: dict | None = None, is_console: bool = None
    ):
        """Синхронная запись информационного сообщения."""
        if data is None:
            data = {}

        now = datetime.datetime.now()
        log_path = self.build_log_path(target, now)
        os.makedirs(os.path.dirname(log_path), exist_ok=True)

        log_data_str = f"{now:%d.%m.%Y %H:%M:%S} {target}: {message}"
        if data:
            log_data_str += f": {self.safe_serialize(data)}"

        logger = logging.getLogger(f"sync_logger_{target}")
        logger.setLevel(logging.INFO)

        if not logger.handlers:
            handler = logging.FileHandler(log_path, mode="a", encoding="utf-8")
            formatter = logging.Formatter("%(message)s")
            handler.setFormatter(formatter)
            logger.addHandler(handler)

        logger.info(log_data_str)

        should_print = self.log_print if is_console is None else is_console
        if should_print:
            print(log_data_str)

    def log_error_sync(
        self, target: str = "", message: str = "", data: dict | None = None, is_console: bool = None
    ):
        """Синхронная запись ошибки."""
        self.log_info_sync(target, f"ERROR: {message}", data, is_console)

    async def log_warning(
        self, target: str = "", message: str = "", data: dict | None = None, is_console: bool = None
    ):
        """Асинхронная запись предупреждения."""
        await self.log_info(target, f"WARNING: {message}", data, is_console)

    def log_warning_sync(
        self, target: str = "", message: str = "", data: dict | None = None, is_console: bool = None
    ):
        """Синхронная запись предупреждения."""
        self.log_info_sync(target, f"WARNING: {message}", data, is_console)

    def safe_serialize(self, obj):
        """Преобразует объект в сериализуемый вид для JSON/log."""
        if obj is None:
            return None
        elif isinstance(obj, (str, int, float, bool)):
            return obj
        elif isinstance(obj, dict):
            return {k: self.safe_serialize(v) for k, v in obj.items()}
        elif isinstance(obj, (list, tuple, set)):
            return [self.safe_serialize(v) for v in obj]
        elif hasattr(obj, "model_dump"):  # Pydantic
            return self.safe_serialize(obj.model_dump())
        elif hasattr(obj, "__dict__"):
            return {k: self.safe_serialize(v) for k, v in vars(obj).items() if not k.startswith("_")}
        else:
            return f"<{type(obj).__name__}>"

    async def shutdown(self):
        """Корректно закрывает все логгеры."""
        for h in list(self.handlers.values()):
            try:
                await h["logger"].shutdown()
            except Exception:
                pass

Разбор кода log.py

Рассмотрим ключевые моменты:

1. Класс Log

Основной класс, который управляет логированием. При создании создаёт папку log/ и определяет, нужно ли выводить логи в консоль через переменную окружения LOG_PRINT.

2. build_log_path()

Формирует путь к файлу лога на основе даты. Структура: log/ГОД/МЕСЯЦ/ДЕНЬ.log

3. Асинхронные методы (log_info, log_error, log_warning)

Используют aiologger для неблокирующей записи в файлы. Это особенно важно в продакшене, где логирование не должно замедлять ответ на запрос.

4. Синхронные методы (log_info_sync, log_error_sync, log_warning_sync)

Используют стандартный logging модуль. Удобны для быстрого логирования в простых скриптах или для обратной совместимости.

5. safe_serialize()

Преобразует сложные объекты (Pydantic модели, словари, списки) в строковое представление для записи в лог. Это предотвращает ошибки при попытке записать несериализуемые объекты.

6. shutdown()

Корректно закрывает все асинхронные логгеры при остановке приложения.

---

Утилита lifespan.py

Теперь создадим файл app/utils/lifespan.py, который будет управлять жизненным циклом нашего приложения: что делать при старте и при остановке.

                from fastapi import FastAPI
from contextlib import asynccontextmanager
from ..utils.log import Log
from ..config import settings


def get_lifespan():
    """Возвращает lifespan контекстный менеджер"""
    
    @asynccontextmanager
    async def lifespan(app: FastAPI):
        """Управление жизненным циклом приложения"""
        
        # Создание объекта логгера
        app.state.log = Log()
        
        # Лог запуска приложения
        await app.state.log.log_info(target="startup", message=f"Запуск приложения на порту {settings.APP_PORT}")
        
        # Контекст приложения
        yield
        
        # Лог остановки
        await app.state.log.log_info(target="shutdown", message="Остановка приложения")
        
        # Shutdown для корректного завершения работы асинхронных логгеров
        await app.state.log.shutdown()
    
    return lifespan

Разбор кода lifespan.py

• @asynccontextmanager — декоратор, который создаёт асинхронный контекстный менеджер
• app.state.log — сохраняем объект логгера в состоянии приложения, чтобы он был доступен во всех модулях
• yield — точка, где приложение работает. Всё, что до yield — выполняется при старте, всё после — при остановке
• app.state.log.shutdown() — корректно закрываем логгеры перед завершением

---

Обновление main.py

Теперь обновим наш app/main.py, чтобы использовать созданные утилиты. Откройте файл и замените его содержимое на:

                from fastapi import FastAPI
from .utils.lifespan import get_lifespan
from .config import settings

# Получаем lifespan
lifespan = get_lifespan()

app = FastAPI(
    title=settings.APP_NAME,
    description=settings.APP_DESCRIPTION,
    version=settings.APP_VERSION,
    lifespan=lifespan
)

@app.get("/")
async def home():
    return {"message": "Hello World! Мой первый сайт на FastAPI работает!"}

@app.get("/about")
async def about():
    return {"message": "Это страница 'О проекте'."}

Теперь при запуске приложения вы увидите в консоли лог о запуске:

04.08.2026 14:30:00 startup: Запуск приложения на порту 8000

А в папке log/2026/08/ появится файл 04.log с этой записью.

---

Итог

Сегодня мы создали:

• ✅ app/utils/log.py — гибкую систему логирования с асинхронной и синхронной записью
• ✅ app/utils/lifespan.py — управление жизненным циклом приложения
• ✅ Настроили автоматическую ротацию лог-файлов по датам
• ✅ Подключили логирование к основному приложению

После проверки закоммитьте и запушьте изменения в GitHub:

git add .
git commit -m "Добавлено логирование и lifespan"
git push