Обзор библиотек Python для работы с данными в SQLite и PostgreSQL

Расширенные примеры и нераспространенные сценарии

Транзакции с явным управлением в SQLAlchemy Core

from sqlalchemy import create_engine, text

engine = create_engine('sqlite:///transactions.db')
with engine.begin() as conn:  # автоматический commit или rollback при ошибке
    conn.execute(text('CREATE TABLE IF NOT EXISTS accounts (id INTEGER PRIMARY KEY, balance REAL)'))
    conn.execute(text('INSERT INTO accounts (balance) VALUES (:bal)'), {'bal': 1000.0})
    # если ниже возникнет исключение, транзакция откатится
    conn.execute(text('UPDATE accounts SET balance = balance + 500 WHERE id = 1'))
# после выхода из блока conn, если не было ошибок, происходит commit

Результат: таблица accounts создана, баланс 1500.0.

Метод begin() гарантирует, что соединение вернется в пул после завершения транзакции.

Bulk insert с помощью executemany

import sqlite3

conn = sqlite3.connect(':memory:')
c = conn.cursor()
c.execute('CREATE TABLE numbers (val INTEGER)')
data = [(i,) for i in range(100000)]
c.executemany('INSERT INTO numbers VALUES (?)', data)
conn.commit()
print('Вставлено', c.rowcount, 'строк')
conn.close()

Вставлено 100000 строк

Использование executemany значительно быстрее, чем множественные вызовы execute. Для больших объемов стоит увеличить параметр batch_size в ORM.

Кастомные типы в SQLAlchemy ORM

from sqlalchemy import create_engine, Column, Integer, String, TypeDecorator
from sqlalchemy.orm import declarative_base, Session
import json

class JSONType(TypeDecorator):
    impl = String
    def process_bind_param(self, value, dialect):
        return json.dumps(value) if value else None
    def process_result_value(self, value, dialect):
        return json.loads(value) if value else None

Base = declarative_base()

class Article(Base):
    __tablename__ = 'articles'
    id = Column(Integer, primary_key=True)
    tags = Column(JSONType)

engine = create_engine('sqlite:///custom_type.db')
Base.metadata.create_all(engine)

with Session(engine) as session:
    article = Article(tags=['python', 'sqlalchemy'])
    session.add(article)
    session.commit()
    print(article.tags)

['python', 'sqlalchemy']

Кастомный тип JSONType автоматически сериализует список в JSON при сохранении и десериализует при загрузке.

Пагинация с использованием LIMIT и OFFSET

from sqlalchemy import create_engine, text

engine = create_engine('sqlite:///pagination.db')
with engine.begin() as conn:
    conn.execute(text('CREATE TABLE IF NOT EXISTS posts (id INTEGER PRIMARY KEY, title TEXT)'))
    for i in range(1, 101):
        conn.execute(text('INSERT INTO posts (title) VALUES (:t)'), {'t': f'Post {i}'})

# пагинация: страница 3, по 10 записей
page = 3
per_page = 10
with engine.connect() as conn:
    result = conn.execute(
        text('SELECT * FROM posts ORDER BY id LIMIT :limit OFFSET :offset'),
        {'limit': per_page, 'offset': (page-1)*per_page}
    )
    for row in result:
        print(row.id, row.title)

21 Post 21
22 Post 22
...
30 Post 30

Важно: в SQLite OFFSET без LIMIT не работает, поэтому всегда указывайте LIMIT.

Использование подзапросов в SQLAlchemy Core

from sqlalchemy import create_engine, text, select, func, table, column

engine = create_engine('sqlite:///subquery_example.db')
with engine.begin() as conn:
    conn.execute(text('''CREATE TABLE orders(
        id INTEGER PRIMARY KEY,
        customer_id INTEGER,
        amount REAL
    )'''))
    conn.execute(text('''INSERT INTO orders VALUES
        (1,1,100), (2,1,200), (3,2,150), (4,2,300)'''))

orders = table('orders', column('id'), column('customer_id'), column('amount'))
subq = select(func.avg(orders.c.amount)).where(orders.c.customer_id == 1).scalar_subquery()
stmt = select(orders).where(orders.c.amount > subq)
with engine.connect() as conn:
    result = conn.execute(stmt)
    for row in result:
        print(row)

(1, 1, 100.0)
(2, 1, 200.0)

Подзапрос scalar_subquery() вычисляет среднюю сумму для клиента 1 (150), затем выбираются все заказы этого клиента с суммой больше 150 (только 200).

Асинхронное соединение с пулом в asyncpg

import asyncio
import asyncpg

async def main():
    pool = await asyncpg.create_pool(
        user='myuser', password='mypass', database='mydb',
        host='localhost', min_size=2, max_size=10
    )
    async with pool.acquire() as conn:
        await conn.execute('''CREATE TABLE IF NOT EXISTS metrics (
            id serial PRIMARY KEY, value real, ts timestamptz DEFAULT now()
        )''')
        await conn.execute('INSERT INTO metrics (value) VALUES ($1)', 3.14)
        row = await conn.fetchrow('SELECT * FROM metrics')
        print(dict(row))
    await pool.close()

asyncio.run(main())

{'id': 1, 'value': 3.14, 'ts': datetime.datetime(2025, 4, 2, 12, 0, 0, tzinfo=datetime.timezone.utc)}

Пул соединений позволяет эффективно работать с большим количеством запросов, переиспользуя открытые соединения.

Миграции схемы с использованием Alembic (SQLAlchemy)

# Установка: pip install alembic
# Инициализация: alembic init alembic
# Настройка env.py для подключения к БД
# Создание миграции: alembic revision --autogenerate -m "add email field"
# Применение: alembic upgrade head
# Пример файла миграции (автоматически генерируется):
"""
Revision ID: abc123
Revises: 
Create Date: 2025-04-02
"""
from alembic import op
import sqlalchemy as sa

def upgrade():
    op.add_column('users', sa.Column('email', sa.String(120), nullable=True))

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

Миграции позволяют версионировать изменения схемы БД и выполнять их последовательно.