Сьогодні підкоримо вершини 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"
# Створюємо engine SQLAlchemy
engine = create_engine(DATABASE_URL)
# Базовий клас для всіх моделей
Base = declarative_base()
# Створюємо сесію для взаємодії з базою
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
На цьому етапі ми створили підключення до PostgreSQL через SQLAlchemy. Engine (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"
# Асинхронний engine 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 — це потужне поєднання, яке дозволяє не лише полегшити взаємодію з базою даних, а й використовувати її можливості на повну. Разом вони роблять наш код красивішим, а додатки продуктивнішими.
ПЕРЕЙДІТЬ В ПОВНУ ВЕРСІЮ