Как работать с базами данных (SQLAlchemy, ORM)?python-44

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# Создаем движок (подключение к SQLite)
engine = create_engine('sqlite:///database.db', echo=True)

# Базовый класс для моделей
Base = declarative_base()

# Фабрика сессий
Session = sessionmaker(bind=engine)
session = Session()

from sqlalchemy import Column, Integer, String, ForeignKey

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    email = Column(String(100), unique=True)

    # Связи (relationships)
    addresses = relationship("Address", back_populates="user")

class Address(Base):
    __tablename__ = 'addresses'

    id = Column(Integer, primary_key=True)
    email = Column(String(100))
    user_id = Column(Integer, ForeignKey('users.id'))

    user = relationship("User", back_populates="addresses")

# Создаем таблицы
Base.metadata.create_all(engine)

new_user = User(name="John Doe", email="john@example.com")
session.add(new_user)
session.commit()  # Не забываем коммитить!

# Получение всех пользователей
users = session.query(User).all()

# Фильтрация
john = session.query(User).filter_by(name="John Doe").first()

# Сложные запросы
from sqlalchemy import or_
users = session.query(User).filter(
    or_(
        User.name.like("%John%"),
        User.email.contains("example")
    )
).order_by(User.name).limit(10)

user = session.query(User).get(1)  # Получаем пользователя с id=1
user.email = "new_email@example.com"
session.commit()

user = session.query(User).get(1)
session.delete(user)
session.commit()

# Создание связанных объектов
user = User(name="Alice")
user.addresses = [
    Address(email="alice@home.com"),
    Address(email="alice@work.com")
]
session.add(user)
session.commit()

# Запрос с join
users_with_addresses = session.query(User).join(Address).filter(
    Address.email.like("%@work.com%")
).all()

try:
    user1 = User(name="Bob")
    session.add(user1)

    # Имитация ошибки
    raise Exception("Oops!")

    session.commit()
except:
    session.rollback()  # Откатываем изменения
    raise
finally:
    session.close()  # Всегда закрываем сессию

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

async def main():
    engine = create_async_engine("postgresql+asyncpg://user:pass@host/db")
    async with AsyncSession(engine) as session:
        result = await session.execute(
            select(User).where(User.name == "John")
        )
        user = result.scalars().first()

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50))

from sqlalchemy.orm import scoped_session
session = scoped_session(sessionmaker(bind=engine))