Это был нелегкий путь... 😅 Публикация планируется уже в эту субботу, 12 октября. 🗓 Статья получилась объёмная — около 30 минут чтения, но оно того стоит! 🔥

​​Друзья, доброе утро! 🌅

Опубликовал обещанную пилотную статью из мини-курса по асинхронной SQLAlchemy в стиле ORM! 🎉 В ней подробно раскрыл следующие темы:

📌 Основы SQLAlchemy: что это такое и как с ним работать
📌 Настройка проекта для работы с этим инструментом
📌 Детали моделей таблиц, как их описывать в современном стиле с использованием Mapped и mapped-column
📌 Связи между таблицами (один к одному, один ко многим, многие к одному) с примерами и объяснениями
📌 Alembic — создание и управление таблицами в асинхронном стиле и особенности миграций

Материала получилось много, но уверен, что после прочтения вопросов не останется. Статья рассчитана на разработчиков, которые планируют работать с табличными базами данных, такими как SQLite, PostgreSQL, MySQL и другие. Это сэкономит вам кучу времени при интеграции баз данных в проекты! 🧠⏳

Статья написана на примере работы с PostgreSQL, но описанный код будет работать с любой базой данных. Кстати, я упомянул три способа развернуть PostgreSQL.

Буду рад вашей поддержке и обратной связи, так как написание таких материалов требует

Вопросы и обратную связь по статье "Асинхронный SQLAlchemy 2: простой пошаговый гайд по настройке, моделям, связям и миграциям с использованием Alembic" можно оставить ниже👇

Друзья, добрый вечер!

Для всех, кто пользуется или только планирует воспользоваться моей "болванкой" для написания ботов на Aiogram 3 с использованием асинхронной SQLAlchemy, у меня отличная новость — важное обновление!

🚀 Я улучшил базовый класс для работы с таблицами. Теперь базовый класс больше не создаёт сессию в каждом методе. Этот подход оказался не самым оптимальным, и в следующей статье я объясню, почему именно. Но спойлер: это напрямую влияет на производительность и управляемость транзакциями.

Теперь, вместо создания сессии в каждом методе, я предлагаю использовать декоратор connection, который импортируется из базового config-файла базы данных (database.py). Декоратор выглядит следующим образом:

def connection(method):
    async def wrapper(*args, **kwargs):
        async with async_session_maker() as session:
            try:
                return await method(*args, session=session, **kwargs)
            except Exception as e:
                await session.rollback()  # Откат транзакции в случае ошибки
                raise e
            finally:
                await session.close()  # Закрытие сессии
    return wrapper

Пример использования в обработчике команды:

@user_router.message(CommandStart())
@connection
async def cmd_start(message: Message, command: CommandObject, session, **kwargs):
    try:
        user_id = message.from_user.id
        user_info = await UserDAO.find_one_or_none(session=session, telegram_id=user_id)

        if user_info:
            await message.answer(f"👋 Привет, {message.from_user.full_name}! Выберите необходимое действие")
            return

        ref_id = get_refer_id_or_none(command_args=command.args, user_id=user_id)

        # Добавление нового пользователя
        await UserDAO.add(
            session=session,
            telegram_id=user_id,
            username=message.from_user.username,
            first_name=message.from_user.first_name,
            last_name=message.from_user.last_name,
            referral_id=ref_id
        )

        ref_message = f" Вы успешно закреплены за пользователем с ID {ref_id}" if ref_id else ""
        msg = f"🎉 <b>Благодарим за регистрацию!{ref_message}</b>."

        await message.answer(msg)

    except Exception as e:
        logger.error(f"Ошибка при выполнении команды /start для пользователя {message.from_user.id}: {e}")
        await message.answer("Произошла ошибка при обработке вашего запроса. Пожалуйста, попробуйте снова позже.")

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

Обновлённую версию проекта можно уже найти на моём GitHub. 📂

Этот подход, так же, будет хорошо работать и в обычных асинхронных функциях, например для ваших парсеров.

Что касается FastApi, что, думаю, многих интересует, тут подход будет немного другой.

Функция, которая будет выполнять роль декоратора в FastApi:

async def get_db_session() -> AsyncSession:
    async with async_session_maker() as session:
        try:
            yield session  # Сессия открыта на всё время работы эндпоинта
        except Exception as e:
            await session.rollback()  # Откат транзакции в случае ошибки
            raise e  # Повторно поднимаем исключение, чтобы оно обработалось дальше
        finally:
            await session.close()  # Явное закрытие сессии

Далее нам необходимо использовать технологию зависимостей в FastApi (dependenses)

from fastapi import Depends, FastAPI
from sqlalchemy.ext.asyncio import AsyncSession

app = FastAPI()

@app.get("/users/{user_id}")
async def get_user(user_id: int, session: AsyncSession = Depends(get_db_session)):
    user = await UserDAO.find_one_or_none(session=session, id=user_id)
    if user:
        return user
    return {"error": "User not found"}

Всем привет! 🚀 Не стал затягивать и спешу обрадовать вас: уже завтра выходит вторая часть статьи по асинхронной SQLAlchemy!

Что вас ждет в новой статье:

✅ Сессии и фабрики сессий: Поговорим о том, как управлять сессиями для взаимодействия с базой данных.
✅ Добавление данных в таблицы: Узнаем, как правильно и безопасно добавлять новые записи, обсудим метод flush и его отличие от commit.
✅ Извлечение данных из таблиц: Обретем уверенность в извлечении данных через select, используя фильтры (where, filter, filter_by).
✅ Разберем методы SQLAlchemy, такие как scalar, scalars, scalar_one_or_none, all, и научимся преобразовывать объекты SQLAlchemy в удобные словари Python с помощью Pydantic.

С исходным кодом второй части можно ознакомиться по ссылке: GitHub.

Не пропустите! 😉

​​Друзья, доброе утро! ☀️ Обещанное продолжение в цикле публикаций про асинхронную SQLAlchemy уже вышло на Хабре и его можно почитать прямо сейчас. Статья получила название "Асинхронный SQLAlchemy 2: пошаговый гайд по управлению сессиями, добавлению и извлечению данных с Pydantic".

В ней я:

✅ Рассматриваю сессии и фабрики сессий — как эффективно управлять сессиями для взаимодействия с базой данных.
✅ Объясняю добавление данных в таблицы — безопасные методы добавления новых записей с использованием ORM-методов.
✅ Погружаюсь в извлечение данных из таблиц — от простых запросов до более сложных фильтров.

После прочтения этой статьи вы сможете уверенно работать с данными в любой табличной базе данных с помощью SQLAlchemy.

Надеюсь на вашу поддержку и комментарии, так как написание таких статей — задача не из простых. 😅 Ссылка на статью и исходный код, использованный в примерах, закреплены под постом. 📌

Приятного чтения и продуктивного дня! 😊

Вопросы и обратную связь по статье "Асинхронный SQLAlchemy 2: пошаговый гайд по управлению сессиями, добавлению и извлечению данных с Pydantic" можно оставить ниже в комментариях👇

Друзья, рад представить большое обновление шаблона для разработки ботов на Aiogram 3 в связке с асинхронной SQLAlchemy.

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

Ключевые изменения:

1) Отказ от использования **kwargs в базовом классе для работы с SQLAlchemy:

— Теперь методы принимают объекты Pydantic моделей, что делает код более структурированным и безопасным. Вместо передачи множества непредсказуемых аргументов через kwargs, данные фильтров передаются через строго типизированные модели.
— Для этого введен новый файл schemas.py, где описаны схемы Pydantic, которые используются для фильтрации и передачи данных.

Пример:

from pydantic import BaseModel, ConfigDict

class TelegramIDModel(BaseModel):
    telegram_id: int
    model_config = ConfigDict(from_attributes=True)

class UserModel(TelegramIDModel):
    username: str | None
    first_name: str | None
    last_name: str | None
    referral_id: str | None

2) Изменения в базовом и дочерних классах для работы с данными SQLAlchemy:

— Теперь для работы с данными используются дженерики, что повышает типобезопасность и делает код более гибким.
— Это особенно полезно при работе с DAO-классами, обеспечивая более точную типизацию и предсказуемость в использовании моделей данных.

3) Декоратор @connection стал фабрикой декораторов:

— Теперь декоратор @connection принимает аргументы, что добавляет гибкости при настройке поведения сессии. В частности, появилась возможность задавать уровень изоляции транзакций через параметр isolation_level.
— Пример использования декоратора:

def connection(isolation_level=None):
    def decorator(method):
        @wraps(method)
        async def wrapper(*args, **kwargs):
            async with async_session_maker() as session:
                try:
                    if isolation_level:
                        await session.execute(text(f"SET TRANSACTION ISOLATION LEVEL {isolation_level}"))
                    return await method(*args, session=session, **kwargs)
                except Exception as e:
                    await session.rollback()
                    raise e
                finally:
                    await session.close()
        return wrapper
    return decorator

4) Пример использования новых методов:

— В методе обработки команды /start теперь используются Pydantic модели для фильтрации данных пользователя. Это делает код более понятным и устойчивым к ошибкам:

@user_router.message(CommandStart())
@connection()
async def cmd_start(message: Message, command: CommandObject, session, state: FSMContext, **kwargs):
    await state.clear()
    user_id = message.from_user.id

    try:
        user_info = await UserDAO.find_one_or_none(session=session, filters=TelegramIDModel(telegram_id=user_id))

        if user_info:
            await message.answer(GREETING_TEXT, reply_markup=start_keyboard())
            return

        ref_id = get_refer_id_or_none(command_args=command.args, user_id=user_id)
        values = UserModel(telegram_id=user_id, username=message.from_user.username, 
                           first_name=message.from_user.first_name, last_name=message.from_user.last_name, 
                           referral_id=ref_id)
        await UserDAO.add(session=session, values=values)
        await message.answer(GREETING_TEXT, reply_markup=start_keyboard())

    except Exception as e:
        logger.error(f"Ошибка при выполнении команды /start для пользователя {user_id}: {e}")
        await message.answer("Произошла ошибка при обработке вашего запроса. Пожалуйста, попробуйте снова позже.")

Зачем это было сделано:

— Безопасность: Использование Pydantic-моделей вместо kwargs повышает типобезопасность, помогает избежать ошибок и упрощает валидацию данных.
— Удобство: Дженерики и строгая типизация делают код более предсказуемым и удобным для работы, особенно при масштабировании проекта.
— Гибкость: Декоратор теперь может управлять сессией более детально, что позволяет гибко адаптировать уровень изоляции транзакци под различные сценарии.

Ссылка на GitHub:

Все изменения уже доступны в репозитории: Ссылка на репозиторий. Жду ваших отзывов и предложений!

Друзья, обещанная подробная статья про Pydantic уже опубликована на Хабре! 🎉

В ней я подробно рассмотрел Pydantic 2 - мощный инструмент для валидации данных в Python, который используется в более чем 30% всех Python-проектов.

Основные темы статьи:

• Что такое Pydantic и зачем он нужен
• Модели и поля в Pydantic
• Встроенные и кастомные механизмы валидации
• Автогенерируемые поля
• Настройка моделей через ConfigDict
• Наследование моделей
• Интеграция с ORM (на примере SQLAlchemy)
• Преобразование данных в словари и JSON

Статья подойдет как новичкам, так и опытным разработчикам. Вы узнаете, как применять Pydantic 2 в своих проектах и почему он стал неотъемлемой частью экосистемы Python.

Ссылка на статью: Pydantic 2: Полное руководство для Python-разработчиков — от основ до продвинутых техник

Приятного чтения! Буду рад вашим комментариям и вопросам. 👨‍💻📚

Упрощенное тестирование Pydantic с имитацией ORM

При изучении Pydantic по моей статье, вы можете столкнуться с необходимостью тестирования метода model_validate(my_obj, from_attributes=True). Чтобы избежать развертывания отдельной базы данных с описанием модели таблицы SQLAlchemy, можно использовать следующий элегантный подход:

1. Создание имитации ORM-класса

Создайте простой класс, имитирующий ORM-модель. Например:

from datetime import date

class MyUser:
    id = 3
    name = "Алексей"
    surname = "Яковенко"
    birthday_date = date(1993, 2, 19)

2. Определение Pydantic-модели

Создайте соответствующую Pydantic-модель. Например:

from pydantic import BaseModel, computed_field
from dateutil.relativedelta import relativedelta

class User(BaseModel):
    id: int
    name: str
    surname: str
    birthday_date: date

    @computed_field
    def full_name(self) -> str:
        return f"{self.name} {self.surname}"

    @computed_field
    def age(self) -> str:
        today = date.today()
        delta = relativedelta(today, self.birthday_date)
        return f"{delta.years} лет, {delta.months} месяцев и {delta.days} дней"

3. Использование модели

Теперь вы можете использовать вашу "ORM-модель" с Pydantic:

my_obj = MyUser()
alex = User.model_validate(my_obj, from_attributes=True)
print(alex.model_dump())

Результат

{
    'id': 3,
    'name': 'Алексей',
    'surname': 'Яковенко',
    'birthday_date': datetime.date(1993, 2, 19),
    'full_name': 'Алексей Яковенко',
    'age': '31 лет, 7 месяцев и 29 дней'
}

Почему это работает

Этот подход эффективен благодаря мощи объектно-ориентированного программирования в Python. Создавая простой класс MyUser, вы фактически имитируете поведение ORM-модели. Pydantic способен работать с этим классом так же, как он работал бы с реальной ORM-моделью, извлекая атрибуты и применяя свою логику валидации и преобразования.

Такой метод позволяет легко тестировать и демонстрировать функциональность Pydantic без необходимости настройки полноценной базы данных, что особенно удобно для обучения и быстрого прототипирования.

Друзья, решил поделиться с вами универсальным методом получения UserID пользователя в приложениях Telegram-ботов с MiniApp. Этот подход позволит получить UserID из командного меню, текстовых и инлайн кнопок.

1. Выполняем импорт JS TelegramWEBApp

<script src="https://telegram.org/js/telegram-web-app.js"></script>

2. Пишем функцию

async function getUserID() {
    const tg = window.Telegram.WebApp;
    if (tg.initDataUnsafe?.user?.id) {
        console.log("User data from Telegram:", JSON.stringify(tg.initDataUnsafe.user));
        return tg.initDataUnsafe.user.id;
    }

    const urlParams = new URLSearchParams(window.location.search);
    const userID = urlParams.get('user_id');
    if (userID) {
        console.log("User ID from URL:", userID);
        return userID;
    }

    throw new Error("User ID not available");
}

- Сначала она пытается получить ID из объекта Telegram Web App, что работает для командного меню и инлайн-кнопок.
- Если первый метод не сработал, функция ищет параметр 'user_id' в URL, что позволяет передавать ID через текстовые кнопки (не забудьте только подставить параметр запроса, чтоб получилась ссылка такого вида: https://your_site.ru?user_id=12345).
- Если оба метода не дали результата, функция выбрасывает ошибку.

Проект QRCodeMaster наконец-то готов! В ходе разработки пришлось пересмотреть несколько подходов, но теперь я полностью доволен результатом.

Бот представляет собой MiniApp + Telegram-бота с функциями:

📷 Сканирование QR-кодов в реальном времени через камеру.
📤 Распознавание QR-кодов на загруженных изображениях.
✨ Создание новых QR-кодов с возможностью кастомизации.

Как и в предыдущем проекте, всё работает в одном приложении на FastAPI, которое использует вебхуки для высокой скорости обработки запросов. В боте я использовал две JavaScript библиотеки:

jsQR — для сканирования QR-кодов через камеру и загрузку изображений.
EasyQRCodeJS — для генерации QR-кодов.

Проект уже доступен на GitHub, а бота можно протестировать здесь: @qr_code_masterBOT.

К среде постараюсь выложить на Хабре подробную статью о разработке с детальным описанием всех этапов.

Новая статья готова. Сегодня ее вечером вычитаю и завтра к утру, если без форс-мажоров, выполню публикацию. Будете ждать?