SQLAlchemy для баз данных PostgreSQL

Раздел: Базы данных -> postgresql

Основы работы с SQLAlchemy и PostgreSQL

SQLAlchemy предоставляет универсальный интерфейс для взаимодействия с базами данных. При использовании PostgreSQL основным драйвером выступает psycopg2 или asyncpg. Ниже рассматривается наиболее распространённая конфигурация.

Базовое подключение и создание сессии

Для начала устанавливаются необходимые пакеты:

pip install sqlalchemy psycopg2-binary

Sqlalchemy python postgresql (sqlalchemy с postgresql)

Далее создаётся движок и сессия:

from sqlalchemy import create_engine\nfrom sqlalchemy.orm import sessionmaker\n\nengine = create_engine('postgresql://user:pass@localhost/mydb')\nSession = sessionmaker(bind=engine)\nsession = Session()

После завершения работы сессию рекомендуется закрывать:

session.close()

При использовании в веб-приложениях целесообразно применять ScopedSession для управления жизненным циклом.

Как выполнить асинхронное подключение?

Для асинхронной работы требуется драйвер asyncpg и расширение asyncio из SQLAlchemy:

pip install asyncpg sqlalchemy[asyncio]

Создание асинхронного движка и сессии:

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession\nfrom sqlalchemy.orm import sessionmaker\n\nasync_engine = create_async_engine('postgresql+asyncpg://user:pass@localhost/mydb')\nAsyncSessionLocal = sessionmaker(async_engine, class_=AsyncSession)\n\nasync def get_session():\n    async with AsyncSessionLocal() as session:\n        yield session

Асинхронный подход позволяет эффективно обрабатывать множество запросов без блокировки потока.

Как управлять миграциями схемы данных?

Для отслеживания изменений в структуре таблиц применяется Alembic. После его установки инициализируется миграционная среда:

pip install alembic\nalembic init alembic

В файле alembic/env.py указывается целевая метадата моделей:

from mymodels import Base\ntarget_metadata = Base.metadata

Создание миграции выполняется командой:

alembic revision --autogenerate -m \"описание\"

Применение изменений:

alembic upgrade head

Данный механизм гарантирует синхронизацию кода моделей и реальной схемы базы данных.

Как организовать связи между таблицами?

SQLAlchemy поддерживает декларативное определение отношений. Пример модели с внешним ключом:

from sqlalchemy import Column, Integer, String, ForeignKey\nfrom sqlalchemy.orm import relationship\nfrom sqlalchemy.ext.declarative import declarative_base\n\nBase = declarative_base()\n\nclass User(Base):\n    __tablename__ = 'users'\n    id = Column(Integer, primary_key=True)\n    name = Column(String)\n    addresses = relationship('Address', back_populates='user')\n\nclass Address(Base):\n    __tablename__ = 'addresses'\n    id = Column(Integer, primary_key=True)\n    email = Column(String)\n    user_id = Column(Integer, ForeignKey('users.id'))\n    user = relationship('User', back_populates='addresses')

Запросы с подгрузкой связанных данных выполняются через joinedload или selectinload.

Типичные ошибки и способы их устранения

  • Ошибка подключения: неправильный URI или отсутствие драйвера. Проверяется наличие установленного psycopg2 или asyncpg.
  • Ошибка кодировки: возникает при работе с UTF-8 без указания кодировки в сессии. Рекомендуется явно указывать client_encoding.
  • Timeout соединения: при долгих запросах нужно настроить pool_timeout и pool_recycle.
  • Deadlock: возникает при конкурирующих транзакциях. Решение - правильная последовательность блокировок и использование SELECT FOR UPDATE с осторожностью.

Расширенные примеры использования SQLAlchemy с PostgreSQL

1. Модель с отношениями и запросы с join

Ниже создаются модели для блога: User, Post, Tag. Many-to-many через ассоциативную таблицу.

Пример 
from sqlalchemy import Column, Integer, String, Text, ForeignKey, Table, DateTime\nfrom sqlalchemy.orm import relationship\nfrom sqlalchemy.sql import func\nfrom sqlalchemy.ext.declarative import declarative_base\n\nBase = declarative_base()\n\npost_tag = Table('post_tag', Base.metadata,\n    Column('post_id', Integer, ForeignKey('posts.id'), primary_key=True),\n    Column('tag_id', Integer, ForeignKey('tags.id'), primary_key=True)\n)\n\nclass User(Base):\n    __tablename__ = 'users'\n    id = Column(Integer, primary_key=True)\n    username = Column(String(50), unique=True)\n    posts = relationship('Post', back_populates='author')\n\nclass Post(Base):\n    __tablename__ = 'posts'\n    id = Column(Integer, primary_key=True)\n    title = Column(String(200))\n    content = Column(Text)\n    created_at = Column(DateTime, server_default=func.now())\n    user_id = Column(Integer, ForeignKey('users.id'))\n    author = relationship('User', back_populates='posts')\n    tags = relationship('Tag', secondary=post_tag, back_populates='posts')\n\nclass Tag(Base):\n    __tablename__ = 'tags'\n    id = Column(Integer, primary_key=True)\n    name = Column(String(30), unique=True)\n    posts = relationship('Post', secondary=post_tag, back_populates='tags')

Запрос всех постов с авторами и тегами:

Пример 
from sqlalchemy.orm import joinedload\n\nsession = Session()\nposts = session.query(Post).options(\n    joinedload(Post.author),\n    joinedload(Post.tags)\n).all()\nfor post in posts:\n    print(post.title, post.author.username, [t.name for t in post.tags])

Результат (пример):

Post Title 1 author1 ['python', 'sqlalchemy']\nPost Title 2 author2 ['postgresql']

2. Асинхронная работа с bulk insert и возвратом сгенерированных id

Используется асинхронная сессия и метод insert().returning():

Пример 
from sqlalchemy import insert\nfrom sqlalchemy.ext.asyncio import AsyncSession\n\nasync def bulk_insert_users(session: AsyncSession, names: list):\n    stmt = insert(User).values([{'username': name} for name in names]).returning(User.id, User.username)\n    result = await session.execute(stmt)\n    await session.commit()\n    rows = result.fetchall()\n    for row in rows:\n        print(f\"Inserted user {row.username} with id {row.id}\")\n    return rows

Вызов:

Пример 
import asyncio\nasync def main():\n    async with AsyncSessionLocal() as session:\n        await bulk_insert_users(session, ['alice', 'bob', 'charlie'])\nasyncio.run(main())

Вывод:

Inserted user alice with id 1\nInserted user bob with id 2\nInserted user charlie with id 3

3. Использование JSON полей PostgreSQL

PostgreSQL поддерживает тип JSONB. Пример модели с метаданными в JSON:

Пример 
from sqlalchemy import Column, Integer, String\nfrom sqlalchemy.dialects.postgresql import JSONB\n\nclass Article(Base):\n    __tablename__ = 'articles'\n    id = Column(Integer, primary_key=True)\n    title = Column(String)\n    metadata = Column(JSONB)\n\nsession = Session()\nsession.add(Article(title='JSON Example', metadata={'views': 100, 'tags': ['news']}))\nsession.commit()\n\n# Фильтрация по JSON ключу\nresult = session.query(Article).filter(Article.metadata['views'].as_integer() > 50).all()

Результат для такого запроса возвращает статьи с количеством просмотров более 50.

4. Настройка пула соединений

При высокой нагрузке настраивается пул с указанием минимального и максимального размера:

Пример 
engine = create_engine(\n    'postgresql://user:pass@localhost/mydb',\n    pool_size=10,\n    max_overflow=20,\n    pool_pre_ping=True,\n    pool_recycle=3600\n)

Параметр pool_pre_ping выполняет проверку соединения перед использованием, pool_recycle пересоздаёт соединения через заданный интервал, предотвращая разрывы.

5. Транзакции с вложенными точками сохранения

Использование savepoint позволяет откатывать часть операций:

Пример 
with session.begin():\n    session.add(User(username='test'))\n    sp = session.begin_nested()\n    try:\n        session.add(User(username='test'))  # дубликат username\n        sp.commit()\n    except:\n        sp.rollback()\n        print('Duplicate ignored, outer transaction continues')\n    session.add(User(username='another'))

Внешняя транзакция фиксируется даже при ошибке во вложенной.

6. Использование text() для сложных запросов

Когда ORM неудобен, применяются сырые SQL запросы с параметрами:

Пример 
from sqlalchemy import text\n\nstmt = text(\"SELECT id, username FROM users WHERE created_at > :date ORDER BY id\")\nresult = session.execute(stmt, {'date': '2024-01-01'})\nfor row in result:\n    print(row.id, row.username)

Результат:

5 user5\n6 user6

7. Миграции с Alembic и автоматическое обнаружение изменений

После изменения модели (например, добавления колонки) создаётся миграция. Команды:

Пример 
alembic revision --autogenerate -m \"add email column\"

Файл миграции автоматически содержит операции add_column. Затем выполняется:

Пример 
alembic upgrade head

Для отката: alembic downgrade -1.

8. Логирование SQL запросов

Для отладки включается логгер:

Пример 
import logging\nlogging.basicConfig()\nlogging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)

Теперь все SQL запросы выводятся в консоль.

SQLAlchemy с PostgreSQL - comments

En
Sqlalchemy python postgresql (python)