Создание блога на FastAPI с нуля: JWT, Markdown и современный веб-дизайн

Код красиво написан, но есть замечания

Во-первых, эндпоинты.

/api/blogs/delete_blog
/api/blogs/create_blog

Это не соответствует naming-конвенциями REST

Follow the rest.

POST /api/blogs
GET /api/blogs
GET /api/blogs/{blog_id}
PATCH /api/blogs/{blog_id}

Так выглядит чище.

Во-вторых, бизнес-логика в контроллере

@router.post("/add_post/", summary="Добавление нового блога с тегами")
async def add_blog(
        add_data: BlogCreateSchemaBase,
        user_data: User = Depends(get_current_user),
        session: AsyncSession = TransactionSessionDep
):
    blog_dict = add_data.model_dump()
    blog_dict['author'] = user_data.id
    tags = blog_dict.pop('tags', [])

    try:
        blog = await BlogDAO.add(session=session, values=BlogCreateSchemaAdd.model_validate(blog_dict))
        blog_id = blog.id

        if tags:
            tags_ids = await TagDAO.add_tags(session=session, tag_names=tags)
            await BlogTagDAO.add_blog_tags(session=session,
                                           blog_tag_pairs=[{'blog_id': blog_id, 'tag_id': i} for i in tags_ids])

        return {'status': 'success', 'message': f'Блог с ID {blog_id} успешно добавлен с тегами.'}
    except IntegrityError as e:
        if "UNIQUE constraint failed" in str(e.orig):
            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
                                detail="Блог с таким заголовком уже существует.")
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Ошибка при добавлении блога.")

По идее, с CRUD надо взаимодействовать в Service, делая связку используя интерфейсы, а тут все смешано.

Кстати, в блоке try не принято делать больше одного действия, оберни создание в отдельный метод и вызывай его, это улучшит читабельность блока.

В третьих, CRUD-методы.

@classmethod
async def get_blog_list(cls, session: AsyncSession, author_id: Optional[int] = None, tag: Optional[str] = None,
                        page: int = 1, page_size: int = 10):
    """
    Получает список опубликованных блогов с возможностью фильтрации и пагинации.

    :param session: Асинхронная сессия SQLAlchemy
    :param author_id: ID автора для фильтрации (опционально)
    :param tag: Название тега для фильтрации (опционально)
    :param page: Номер страницы (начиная с 1)
    :param page_size: Количество записей на странице (от 3 до 100)
    :return: Словарь с ключами page, total_page, total_result, blogs
    """
    # Ограничение параметров
    page_size = max(3, min(page_size, 100))
    page = max(1, page)

    # Начальная сборка базового запроса
    base_query = select(cls.model).options(
        joinedload(cls.model.user),
        selectinload(cls.model.tags)
    ).filter_by(status='published')

    # Фильтрация по автору
    if author_id is not None:
        base_query = base_query.filter_by(author=author_id)

    # Фильтрация по тегу
    if tag:
        base_query = base_query.join(cls.model.tags).filter(cls.model.tags.any(Tag.name.ilike(f"%{tag.lower()}%")))

    # Подсчет общего количества записей
    count_query = select(func.count()).select_from(base_query.subquery())
    total_result = await session.scalar(count_query)

    # Если записей нет, возвращаем пустой результат
    if not total_result:
        return {
            "page": page,
            "total_page": 0,
            "total_result": 0,
            "blogs": []
        }

    # Расчет количества страниц
    total_page = (total_result + page_size - 1) // page_size

    # Применение пагинации
    offset = (page - 1) * page_size
    paginated_query = base_query.offset(offset).limit(page_size)

    # Выполнение запроса и получение результатов
    result = await session.execute(paginated_query)
    blogs = result.scalars().all()

    # Удаление дубликатов блогов по их ID
    unique_blogs = []
    seen_ids = set()
    for blog in blogs:
        if blog.id not in seen_ids:
            unique_blogs.append(BlogFullResponse.model_validate(blog))
            seen_ids.add(blog.id)

    # Логирование
    filters = []
    if author_id is not None:
        filters.append(f"author_id={author_id}")
    if tag:
        filters.append(f"tag={tag}")
    filter_str = " & ".join(filters) if filters else "no filters"

    logger.info(f"Page {page} fetched with {len(blogs)} blogs, filters: {filter_str}")
    # Формирование результата
    return {
        "page": page,
        "total_page": total_page,
        "total_result": total_result,
        "blogs": unique_blogs
    }

Пагинацию лучше вынести в отдельный метод, который будет принимать некий select

Для логгинга - создай отдельный логгер, кстати, f-strings не принято использовать в логгерах, в logger.info можно передавать аргументы, которые будут подставлены в финальную строку.

Это самое заметное, конкретно все - не читал.

Как я сделал Telegram-бота для студентов РТСУ

Вам спасибо!

Как я сделал Telegram-бота для студентов РТСУ

Спасибо за позитивный фидбэк, насчёт спойлеров обязательно учту!

Как я сделал Telegram-бота для студентов РТСУ

Рад что Вам нравится =)

Как я сделал Telegram-бота для студентов РТСУ

Вам спасибо =)