Опыт работы с синхронной SQLAlchemy?

Да, я активно работал с синхронной версией SQLAlchemy, которая является мощным ORM (Object-Relational Mapper) для Python. Она позволяет взаимодействовать с базами данных, используя объекты Python, что упрощает разработку и поддержку кода.

Почему синхронная SQLAlchemy? Синхронная SQLAlchemy идеально подходит для традиционных веб-фреймворков, таких как Flask и Django (в случае использования SQLAlchemy вместо встроенного ORM), а также для скриптов и консольных приложений, где операции ввода-вывода могут быть блокирующими без существенного влияния на общую производительность или где асинхронность не требуется.

Ключевые компоненты:

• Engine: Точка входа для взаимодействия с базой данных, управляет пулом соединений.
• Session: Основной интерфейс для работы с базой данных, управляет транзакциями и состоянием объектов.
• Declarative Base: Базовый класс для определения моделей ORM, связывающий классы Python с таблицами базы данных.

Пример синхронного использования (CRUD):

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker, declarative_base

# 1. Определение Engine: подключение к базе данных
# 'sqlite:///example.db' - строка подключения для SQLite
engine = create_engine('sqlite:///example.db', echo=False) # echo=True для вывода SQL-запросов

# 2. Определение Declarative Base
Base = declarative_base()

# 3. Определение модели ORM
class User(Base):
    __tablename__ = 'users' # Имя таблицы в БД
    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False) # Ограничение на длину и обязательность

    def __repr__(self):
        return f"<User(id={self.id}, name='{self.name}')>"

# 4. Создание таблиц в базе данных (если их нет)
Base.metadata.create_all(engine)

# 5. Определение SessionMaker для создания сессий
Session = sessionmaker(bind=engine)

# Пример операций CRUD
# Создание и сохранение объекта
with Session() as session: # Использование контекстного менеджера для автоматического закрытия сессии
    new_user = User(name='Alice')
    session.add(new_user)
    session.commit() # Фиксация изменений в БД
    print(f"Добавлен пользователь: {new_user}")

# Чтение объектов
with Session() as session:
    users = session.query(User).all() # Получить всех пользователей
    print("Все пользователи:")
    for user in users:
        print(user)

    # Поиск по имени
    bob = session.query(User).filter_by(name='Bob').first()
    if not bob:
        bob = User(name='Bob')
        session.add(bob)
        session.commit()
        print(f"Добавлен пользователь: {bob}")

# Обновление объекта
with Session() as session:
    user_to_update = session.query(User).filter_by(name='Alice').first()
    if user_to_update:
        user_to_update.name = 'Alicia'
        session.commit()
        print(f"Обновлен пользователь: {user_to_update}")

# Удаление объекта
with Session() as session:
    user_to_delete = session.query(User).filter_by(name='Bob').first()
    if user_to_delete:
        session.delete(user_to_delete)
        session.commit()
        print(f"Удален пользователь: {user_to_delete}")

Лучшие практики:

• Всегда используйте контекстный менеджер with Session() для управления сессиями. Это гарантирует, что session.close() будет вызван автоматически, даже при возникновении ошибок, предотвращая утечки соединений.
• Вызывайте session.commit() для сохранения изменений в базе данных и session.rollback() в случае ошибок.
• Для асинхронных приложений (например, с FastAPI) рекомендуется использовать SQLAlchemy 2.0 с AsyncSession и асинхронными драйверами БД (например, asyncpg для PostgreSQL).