from sqlalchemy.orm import Mapped, mapped_column, relationship
from sqlalchemy import ForeignKey, Integer, BigInteger, Float
from sqlalchemy import Text, ARRAY
from typing import Optional, List
from app.database import Base


class Brand(Base):
    __tablename__ = 'brands'

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(Text, nullable=False)
    description: Mapped[Optional[str]] = mapped_column(Text, nullable=True,
                                                       default='Без описания',
                                                       server_default='Описание не добавили')

    # Связь 1 бренд -> много товаров
    products: Mapped[List["Product"]] = relationship("Product", back_populates="brand")


class Product(Base):
    __tablename__ = 'products'

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    article: Mapped[int] = mapped_column(BigInteger, unique=True, nullable=False)
    title: Mapped[Optional[str]]
    description: Mapped[Optional[str]] = mapped_column(Text, nullable=True,
                                                       default='Без описания',
                                                       server_default='Описание не добавили')
    barcodes: Mapped[Optional[List[str]]] = mapped_column(ARRAY(Text), nullable=True)

    # Внешний ключ для связи с брендом
    brand_id: Mapped[int] = mapped_column(ForeignKey('brands.id'))
    brand: Mapped["Brand"] = relationship("Brand", back_populates="products")

    # Связь 1 товар -> 1 цена
    price: Mapped["Price"] = relationship("Price", uselist=False, back_populates="product")


class Price(Base):
    __tablename__ = 'prices'

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    value: Mapped[float] = mapped_column(Float, nullable=False)

    # Внешний ключ для связи с товаром
    product_id: Mapped[int] = mapped_column(ForeignKey('products.id'))
    product: Mapped["Product"] = relationship("Product", back_populates="price")

Готовлю для вас пилотную статью, которая станет настоящим путеводителем в мире SQLAlchemy. Выше представлен фрагмент кода, в котором я описал модели таблиц в трех базовых связках

Вот что я планирую охватить:

1️⃣ Создание базового конфигурационного файла (Base) для работы с базами данных на SQLAlchemy. Мы рассмотрим, как подключаться к PostgreSQL и SQLite — двум самым популярным базам данных в проектах. 🛠

2️⃣ Описание моделей таблиц. Я подробно объясню, как создавать обычные колонки, а также покажу, как работать с более экзотическими типами данных, такими как Array и JSON. 📊

3️⃣ Связи между таблицами. Мы разберем непростую тему связей «один к одному», «один ко многим» и «многие к одному». Новички часто теряются в этой теме, поэтому я проведу вас за руку через все нюансы. 🤝

4️⃣ Работа с Alembic. Мы создадим наши модели таблиц со всеми необходимыми связями и миграциями. 🔄

Цикл публикаций будет ориентирован на использование ORM стиля. Я постараюсь максимально игнорировать Core, так как считаю, что при наличии высокоуровневого взаимодействия на Python нет смысла опускаться на более низкие уровни SQL.

Описывать буду только асинхронные методы, как для работы с Alembic, так и для взаимодействия с таблицами в дальнейшем. Подозреваю, что вы из-за этого не расстроитесь. 😉

Будет глобально и фундаментально 🚀

Как вам такой план?

class Product1S(Base):
    # временные ссылки
    link_1s: Mapped[str_or_none]
    link_vendor: Mapped[str_or_none]
    link_unit: Mapped[str_or_none]
    link_nds: Mapped[str_or_none]

    code: Mapped[str_or_none]
    find_code: Mapped[str_or_none]
    name: Mapped[str_or_none]
    full_name: Mapped[str_or_none]
    article: Mapped[str_or_none]
    description: Mapped[description_an]
    barcodes: Mapped[list[str] | None]

    # связки по первичным ключам
    brand_id: Mapped[int_or_none] = mapped_column(ForeignKey('brands.id'))
    nds_id: Mapped[int_or_none] = mapped_column(ForeignKey('ndss.id'))
    unit_id: Mapped[int_or_none] = mapped_column(ForeignKey('units.id'))
    manufacture_id: Mapped[int_or_none] = mapped_column(ForeignKey('manufactures.id'))

Современный синтаксис SQLAlchemy в деле. Минимум телодвижений)

Недавно столкнулся с интересной ситуацией в работе с Telegram ботом, и думаю, что такой кейс может быть полезен 💡

История такая:

Я разработал бота, который работает в системе автообновлений. Когда появляются изменения в основной ветке на GitHub, они автоматически подтягиваются и применяются к боту 🤖. Бот использует переменные окружения, хранящиеся в файле .env.

Бота писал пару месяцев назад и после над проектом работали другие разработчики. Со временем переменные изменялись, и появилась необходимость внести обновления. Проблема в том, что DevOps, который мог бы дать доступ к новым переменным, как назло, был в отпуске 🌴.

📌 Сложность: на руках был исходный код бота, но локально я его запустить не мог. Бот уже был запущен на сервере, и переиспользовать один и тот же токен нельзя. Никто, включая заказчика, не знал всех переменных наизусть 🤷‍♂️.

Как я вышел из ситуации:

1. Временно вынес токен бота прямо в основной файл, чтобы он подтягивался не из .env, а из кода (так делать не рекомендуется, но выхода не было) 🚨.

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

3. Подставил битый токен для инициализации бота 🛠.

4. Запушил изменения на сервер (GitHub) 🚀.

Что получилось?

🤖 Бот на сервере упал, потому что использовал нерабочий токен. Зато все рабочие переменные остались на месте. Затем я поставил боевой токен локально и успешно запустил бота. Новый функционал не затронул старые переменные — всё продолжило работать!

💡 Вывод: даже если доступа к важным данным нет, можно найти решение. Надеюсь, этот кейс окажется полезным для вас!

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

Решил подготовить для вас шпаргалку с самыми популярными командами GIT 🧑‍💻. Это 20 самых часто используемых команд, которые пригодятся как новичкам, так и опытным разработчикам.

1. Инициализация пустого репозитория 👇

git init

💡 Инициализирует новый Git-репозиторий в текущей директории.

2. Клонирование репозитория 👇

git clone <url>

💡 Клонирует существующий репозиторий на локальный компьютер.

3. Добавление удаленного репозитория👇

git remote add origin <url>

💡Добавляет удаленный репозиторий с именем origin.

4. Добавление файлов в отслеживание👇

git add <файл/директория>
git add .

💡Добавляет изменения в указанные файлы или директорию в область подготовки (staging area).
git add . — добавляет все измененные файлы.

5. Создание коммита👇

git commit -m "Сообщение коммита"

💡Создает коммит с указанным сообщением.

6. Отправка изменений на удаленный репозиторий👇

git push origin <branch>

💡 Отправляет локальные изменения на указанный удаленный репозиторий в указанную ветку.

7. Получение изменений с удаленного репозитория👇

git pull

💡Забирает последние изменения с удаленного репозитория и объединяет их с локальными.

8. Просмотр состояния репозитория👇

git status

💡Показывает текущее состояние репозитория, например, какие файлы были изменены или подготовлены к коммиту.

9. Просмотр истории коммитов👇

git log

💡Показывает историю коммитов в репозитории.

10. Создание новой ветки👇

git checkout -b <branch-name>

💡Создает новую ветку и сразу переключается на неё.

11. Переключение между ветками👇

git checkout <branch-name>

💡 Переключается на указанную ветку.

12. Слияние веток👇

git merge <branch-name>

💡 Сливает изменения из указанной ветки в текущую.

13. Отмена последнего коммита (оставляя изменения в файлах)👇

git reset --soft HEAD^

💡Отменяет последний коммит, сохраняя изменения в рабочих файлах.

14. Удаление локальной ветки👇

git branch -d <branch-name>

💡Удаляет локальную ветку.

15. Просмотр различий между файлами👇

git diff

💡Показывает изменения в файлах по сравнению с последним коммитом.

16. Просмотр всех веток👇

git branch

💡 Показывает список всех локальных веток.

17. Переименование текущей ветки👇

git branch -m <new-branch-name>

18. Переименование другой ветки👇

git branch -m <old-branch-name> <new-branch-name>

💡 Переименовывает указанную ветку.

19. Создание новой ветки и переключение на неё👇

git checkout -b <branch-name>

💡 Создает новую ветку и сразу переключается на неё.

20. Настройка ветки для автоматического пуша 👇

git push --set-upstream origin <branch-name>

💡 Связывает текущую ветку с удаленной, чтобы в дальнейшем можно было просто использовать git push без указания ветки.

_____

Сохраняйте пост, чтобы не потерять, и возвращайтесь к нему, когда нужно! Если у вас есть вопросы или дополнения, оставляйте их в комментариях👇

Не забудьте поставить лайк, если шпаргалка оказалась полезной 👍

Это был нелегкий путь... 😅 Публикация планируется уже в эту субботу, 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).
- Если оба метода не дали результата, функция выбрасывает ошибку.