Python: обзор интерфейсов баз данных

Расширенные примеры работы с интерфейсами баз данных

Пример 1: SQLAlchemy с отношениями и транзакциями

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, sessionmaker

engine = create_engine('postgresql://user:pass@localhost/testdb')
Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    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)
Session = sessionmaker(bind=engine)
session = Session()

# Транзакция с добавлением пользователя и адреса
try:
    with session.begin():
        user = User(name='Eve')
        session.add(user)
        session.flush()  # чтобы получить id пользователя
        address = Address(email='eve@example.com', user_id=user.id)
        session.add(address)
except Exception as e:
    session.rollback()
    print(f'Ошибка: {e}')

# Запрос с жадной загрузкой
users = session.query(User).options(joinedload(User.addresses)).all()
for u in users:
    print(f'User: {u.name}, Emails: {[a.email for a in u.addresses]}')

User: Eve, Emails: ['eve@example.com']

Пояснение: Используется relationship для связи таблиц, block=True в session.begin() управляет транзакцией. joinedload уменьшает количество запросов.

Пример 2: Асинхронная работа с PostgreSQL через asyncpg

import asyncio
import asyncpg

async def main():
    conn = await asyncpg.connect(user='postgres', password='secret', database='testdb', host='localhost')
    await conn.execute('''
        CREATE TABLE IF NOT EXISTS tasks (
            id SERIAL PRIMARY KEY,
            title TEXT,
            done BOOLEAN DEFAULT FALSE
        )
    ''')
    await conn.execute('INSERT INTO tasks(title) VALUES($1)', 'Learn asyncpg')
    rows = await conn.fetch('SELECT * FROM tasks')
    for row in rows:
        print(row['title'], row['done'])
    await conn.close()

asyncio.run(main())

Learn asyncpg False

Пояснение: asyncpg использует асинхронные вызовы, подходит для высоконагруженных приложений. $1 для параметров. Требуется Python 3.7+.

Пример 3: Миграции с Alembic (совместно с SQLAlchemy)

# После установки alembic и инициализации (alembic init migrations)
# Создаётся миграция: alembic revision --autogenerate -m "add age column"

# В сгенерированном файле revision.py:
"""add age column

Revision ID: 1234...
"""
from alembic import op
import sqlalchemy as sa

# revision identifiers
revision = '1234...'
down_revision = 'previous_revision'

def upgrade():
    op.add_column('users', sa.Column('age', sa.Integer(), nullable=True))

def downgrade():
    op.drop_column('users', 'age')

# После выполнения alembic upgrade head таблица users получает колонку age

Пояснение: Alembic автоматически генерирует код миграции на основе изменений моделей (если используется --autogenerate). Позволяет откатывать изменения.

Пример 4: Подключение к MySQL через mysql-connector-python

import mysql.connector

cnx = mysql.connector.connect(
    host='localhost',
    user='root',
    password='pass',
    database='test'
)
cursor = cnx.cursor()
cursor.execute('CREATE TABLE IF NOT EXISTS products (id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(100), price DECIMAL(10,2))')
cursor.execute('INSERT INTO products (name, price) VALUES (%s, %s)', ('Laptop', 1500.00))
cnx.commit()
cursor.execute('SELECT * FROM products')
for (id, name, price) in cursor:
    print(f'{id}: {name} - {price}')
cursor.close()
cnx.close()

1: Laptop - 1500.00

Пояснение: mysql-connector-python - официальный драйвер MySQL, использует %s для параметров. Требует установки: pip install mysql-connector-python.