Пример использования SQLAlchemy для работы с PostgreSQL

Сегодня мы покорим вершины PostgreSQL вместе с SQLAlchemy! Будем работать с этой мощной реляционной базой данных, научимся подключаться, выполнять операции и настраивать индексы для повышения производительности.

PostgreSQL — один из самых популярных выборов для реляционных баз данных. Он мощный, надёжный и отлично подходит для высоконагруженных систем. Но что делает PostgreSQL таким замечательным?

• Мощные возможности: поддерживает JSONB (для работы с полуструктурированными данными), Full-Text Search, CTE и Window Functions.
• Расширяемость: можно добавлять свои типы данных и функции.
• Поддержка транзакций: ACID-свойства и оптимизированная обработка запросов.
• Совместимость с SQLAlchemy: наша любимая ORM обладает полной поддержкой возможностей PostgreSQL.

Итак, переходим от теории к практике!

---

Настройка подключения к PostgreSQL

SQLAlchemy взаимодействует с PostgreSQL через драйверы. Самый популярный драйвер для этого — psycopg2. Для асинхронного подключения можно использовать asyncpg.

Установим необходимые библиотеки.

Для простого подключения:

pip install psycopg2-binary

Для асинхронного подключения:

pip install asyncpg sqlalchemy[asyncio]

Создание подключения

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

# URL подключения к PostgreSQL
DATABASE_URL = "postgresql://username:password@localhost:5432/mydatabase"

# Создаём движок SQLAlchemy
engine = create_engine(DATABASE_URL)

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

# Создаём сессию для взаимодействия с базой
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

На этом этапе мы создали подключение к PostgreSQL через SQLAlchemy. Движок (engine) отвечает за подключение к базе и отправку запросов, а сессия — за выполнение этих запросов.

Если вы работаете с FastAPI и хотите использовать возможности асинхронного программирования:

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "postgresql+asyncpg://username:password@localhost:5432/mydatabase"

# Асинхронный движок SQLAlchemy
async_engine = create_async_engine(DATABASE_URL)

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

# Асинхронная сессия
async_session_local = sessionmaker(
    autocommit=False,
    autoflush=False,
    bind=async_engine,
    class_=AsyncSession
)

Здесь в строке подключения используется postgresql+asyncpg, что говорит SQLAlchemy о необходимости использовать асинхронный драйвер.

---

Пример работы с PostgreSQL

Давайте создадим таблицу пользователей с ключевыми атрибутами: id, name, email и created_at.

from sqlalchemy import Column, Integer, String, DateTime, func

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, nullable=False)
    email = Column(String, unique=True, index=True, nullable=False)
    created_at = Column(DateTime(timezone=True), server_default=func.now())

Теперь создадим таблицы в нашей базе данных:

Base.metadata.create_all(bind=engine)

Выполнение базовых операций

Создание записи

def create_user(db, name, email):
    new_user = User(name=name, email=email)
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return new_user

Получение списка пользователей

def get_users(db, skip: int = 0, limit: int = 10):
    return db.query(User).offset(skip).limit(limit).all()

Обновление записи

def update_user(db, user_id, new_email):
    user = db.query(User).filter(User.id == user_id).first()
    if user:
        user.email = new_email
        db.commit()
        db.refresh(user)
    return user

Удаление записи

def delete_user(db, user_id):
    user = db.query(User).filter(User.id == user_id).first()
    if user:
        db.delete(user)
        db.commit()
    return user

---

Оптимизация запросов для PostgreSQL

Индексы в PostgreSQL помогают ускорить выполнение запросов. SQLAlchemy позволяет создавать их простым добавлением ключевого аргумента index=True в поле модели. Мы уже сделали это для id и email.

email = Column(String, unique=True, index=True)

Если нужно создать составной индекс, мы можем использовать Index:

from sqlalchemy import Index

Index('idx_name_created_at', User.name, User.created_at)

Иногда несколько операций нужно выполнять как единое целое, чтобы избежать неконсистентности. Например:

from sqlalchemy.orm import Session

def transactional_operation(db: Session):
    try:
        # Начало транзакции
        user = create_user(db, "Alice", "alice@example.com")
        update_user(db, user.id, "newalice@example.com")
        # Окончание транзакции
        db.commit()
    except Exception:
        db.rollback()
        raise

PostgreSQL поддерживает полнотекстовый поиск, и SQLAlchemy позволяет его использовать через функцию to_tsvector.

from sqlalchemy.sql import text

def search_users(db, query):
    search_query = f"%{query}%"
    return db.execute(
        text("SELECT * FROM users WHERE name ILIKE :search OR email ILIKE :search"),
        {"search": search_query}
    ).fetchall()

---

Поддержка JSON и JSONB

PostgreSQL умеет обрабатывать JSON-данные. SQLAlchemy поддерживает это через JSON и JSONB типы.

from sqlalchemy import JSON

class Product(Base):
    __tablename__ = "products"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, nullable=False)
    specifications = Column(JSON)  # Хранит свойства продукта в формате JSON

Теперь можно записывать и читать данные в формате JSON:

def create_product(db, name, specifications):
    product = Product(name=name, specifications=specifications)
    db.add(product)
    db.commit()
    db.refresh(product)
    return product

---

Асинхронная работа с PostgreSQL

Асинхронность помогает повысить производительность приложений. Это логично: одновременно идёт обработка большого числа запросов.

Пример асинхронного запроса записи

from sqlalchemy.future import select

async def async_get_users(db):
    result = await db.execute(select(User))
    return result.scalars().all()

Создание записи в асинхронном режиме

async def async_create_user(db, name, email):
    new_user = User(name=name, email=email)
    db.add(new_user)
    await db.commit()
    await db.refresh(new_user)
    return new_user

---

PostgreSQL и SQLAlchemy — это мощное сочетание, которое позволяет не только облегчить взаимодействие с базой данных, но и использовать её возможности на полную катушку. Вместе они делают наш код красивее, а приложения производительнее.