Соединение с PostgreSQL через Python: варианты и примеры
Подключение к PostgreSQL из Python
Взаимодействие с PostgreSQL в Python осуществляется через специализированные библиотеки. Рассмотрены основные варианты, их установка, типичные проблемы и сферы применения.
Как подключиться к PostgreSQL с помощью psycopg2?
psycopg2 – самый зрелый и производительный драйвер для работы с PostgreSQL. Подходит для высоконагруженных приложений, где важен прямой контроль соединений и транзакций.
Установка: pip install psycopg2-binary (избегает компиляции).
import psycopg2
conn = psycopg2.connect(
host='localhost',
port=5432,
dbname='mydb',
user='postgres',
password='secret'
)
cur = conn.cursor()
cur.execute('SELECT version();')
version = cur.fetchone()
print(version)
cur.close()
conn.close()
Sql server python (sql server и python)
Шаги: импорт – подключение – курсор – запрос – обработка результата – закрытие курсора и соединения. Использование менеджера контекста (with) автоматически закрывает курсор, но соединение следует закрывать явно или через conn.autocommit = True.
Цель: быстрое выполнение транзакционных запросов, поддержка COPY, подготовленных запросов, работа с бинарными данными.
Ошибка: ModuleNotFoundError: No module named 'psycopg2' – библиотека не установлена. Решение: pip install psycopg2-binary. Если возникает ошибка компиляции на этапе установки, пакет psycopg2-binary является предварительно скомпилированным.
Ошибка: OperationalError: could not connect to server: Connection refused – не запущен сервер PostgreSQL или неверные параметры подключения. Проверить статус сервера и правильность host/port.
Ошибка: SQL-инъекции – нельзя вставлять значения через f-строки. Решение: использовать параметризованные запросы с %s или %(name)s.
Как использовать SQLAlchemy для подключения к PostgreSQL?
SQLAlchemy предоставляет высокоуровневый ORM и Core (низкоуровневый SQL-компилятор). Удобен для проектов, где требуется абстракция от конкретной СУБД, работа с моделями и автоматическое создание схемы.
Установка: pip install sqlalchemy psycopg2-binary (драйвер можно выбирать: psycopg2, asyncpg и др.).
from sqlalchemy import create_engine, text
engine = create_engine('postgresql://postgres:secret@localhost:5432/mydb')
with engine.connect() as conn:
result = conn.execute(text('SELECT version();'))
row = result.fetchone()
print(row[0])
Sql через python (работа с sql через python)
Для ORM: определение моделей через декларативный базовый класс, создание сессии.
Цель: быстрая разработка с автоматическим отображением таблиц, миграции (Alembic), поддержка множества СУБД.
Ошибка: ModuleNotFoundError: No module named 'psycopg2' – отсутствует драйвер. Установить нужный драйвер (psycopg2 или asyncpg).
Ошибка: OperationalError: (psycopg2.OperationalError) FATAL: password authentication failed – неверные учётные данные или не настроен pg_hba.conf. Проверить пароль и метод аутентификации.
Проблема: слишком много открытых соединений – использовать пул соединений через engine = create_engine('...', pool_size=10, max_overflow=20).
Как выполнять асинхронные запросы с asyncpg?
asyncpg – высокопроизводительный асинхронный драйвер для PostgreSQL, работающий через asyncio. Подходит для веб-фреймворков aiohttp, FastAPI, Sanic.
Установка: pip install asyncpg.
import asyncio
import asyncpg
async def main():
conn = await asyncpg.connect(
host='localhost',
port=5432,
database='mydb',
user='postgres',
password='secret'
)
version = await conn.fetchval('SELECT version();')
print(version)
await conn.close()
asyncio.run(main())
Python postgresql (подключение к postgresql из python)
Цель: неблокирующие операции, конкурентное выполнение множества запросов, интеграция с async/await.
Ошибка: asyncpg.exceptions.ConnectionDoesNotExistError: connection is closed – попытка использовать закрытое соединение. Использовать контекстный менеджер (async with conn.transaction():) или проверять статус.
Ошибка: asyncpg.exceptions.InvalidPasswordError – неверный пароль. Проверить данные.
Проблема: пул соединений необходимо реализовать вручную? asyncpg предоставляет asyncpg.create_pool для управления пулом.
Как использовать новую библиотеку psycopg (psycopg3)?
psycopg3 – современное переосмысление psycopg2 с поддержкой async, pipelines, dataclasses и улучшенной производительностью. Активно развивается.
Установка: pip install psycopg (также доступен пакет psycopg-binary).
import psycopg
conn = psycopg.connect('postgresql://postgres:secret@localhost:5432/mydb')
cur = conn.cursor()
cur.execute('SELECT version();')
print(cur.fetchone()[0])
cur.close()
conn.close()
Цель: получение преимуществ async, новые возможности без необходимости переходить на другую библиотеку.
Ошибка: psycopg.errors.OperationalError: could not connect to server – те же причины, что у psycopg2. Решение аналогично.
Проблема: несовместимость с устаревшими расширениями (например, HSTORE в старом формате). Использовать современные альтернативы.
Расширенные примеры для psycopg2
Пример 1: массовая вставка данных (bulk insert)
import psycopg2
conn = psycopg2.connect('postgresql://postgres:secret@localhost:5432/mydb')
data = [('Alice', 30), ('Bob', 25), ('Charlie', 35)]
with conn.cursor() as cur:
cur.executemany(
'INSERT INTO users (name, age) VALUES (%s, %s)',
data
)
conn.commit()
print('Вставлено строк:', cur.rowcount)
Вставлено строк: 3
Пример 2: транзакции и точки сохранения (savepoint)
import psycopg2
conn = psycopg2.connect(...)
conn.autocommit = False
cur = conn.cursor()
try:
cur.execute('INSERT INTO logs (event) VALUES (\'start\');')
sp = conn.cursor('sp1')
sp.execute('INSERT INTO logs (event) VALUES (\'mid\');')
# если ошибка - откат к savepoint
sp.execute('INSERT INTO logs (event) VALUES (\'end\');')
conn.commit()
except Exception:
conn.rollback()
finally:
cur.close()
conn.close()
(нет вывода, только успешный commit)
Пример 3: использование DictCursor для доступа по именам столбцов
from psycopg2.extras import DictCursor
conn = psycopg2.connect(...)
cur = conn.cursor(cursor_factory=DictCursor)
cur.execute('SELECT id, name FROM users LIMIT 2;')
for row in cur:
print(row['id'], row['name'])
1 Alice 2 Bob
Расширенные примеры для SQLAlchemy
Пример 1: ORM – определение модели и сессия
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, Session
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String)
age = Column(Integer)
engine = create_engine('postgresql://...')
Base.metadata.create_all(engine)
with Session(engine) as session:
user = User(name='Alice', age=30)
session.add(user)
session.commit()
print('Добавлен пользователь с id:', user.id)
Добавлен пользователь с id: 1
Пример 2: Core – прямое выполнение запроса с bindparams
from sqlalchemy import create_engine, text, bindparam
engine = create_engine('postgresql://...')
with engine.connect() as conn:
result = conn.execute(
text('SELECT name FROM users WHERE age > :age'),
{'age': 25}
)
for row in result:
print(row.name)
Alice Bob
Расширенные примеры для asyncpg
Пример 1: конкурентные запросы с asyncio.gather
import asyncio
import asyncpg
async def fetch_users(conn):
return await conn.fetch('SELECT * FROM users')
async def fetch_count(conn):
return await conn.fetchval('SELECT count(*) FROM users')
async def main():
conn = await asyncpg.connect(...)
users, count = await asyncio.gather(
fetch_users(conn),
fetch_count(conn)
)
print('Пользователи:', users)
print('Всего:', count)
await conn.close()
asyncio.run(main())
Пользователи: [, ...] Всего: 1
Пример 2: использование пула соединений
import asyncio
import asyncpg
async def query(pool, name):
async with pool.acquire() as conn:
return await conn.fetchrow('SELECT * FROM users WHERE name=$1', name)
async def main():
pool = await asyncpg.create_pool(
user='postgres', password='secret',
host='localhost', database='mydb',
min_size=2, max_size=10
)
result = await asyncio.gather(
query(pool, 'Alice'),
query(pool, 'Bob')
)
print(result)
await pool.close()
asyncio.run(main())
[, None]
Расширенные примеры для psycopg3
Пример 1: pipeline mode – группировка запросов
import psycopg
with psycopg.connect(...) as conn:
with conn.cursor() as cur:
cur.execute('INSERT INTO users (name) VALUES (%s)', ('Alice',))
cur.execute('INSERT INTO users (name) VALUES (%s)', ('Bob',))
conn.commit() # отправляются оба запроса одним пакетом
(нет вывода)
Пример 2: работа с массивами и типами PostgreSQL
import psycopg
with psycopg.connect(...) as conn:
with conn.cursor() as cur:
cur.execute(
'SELECT array[1,2,3]::int[] AS arr'
)
row = cur.fetchone()
print('Массив:', row[0])
print('Тип:', type(row[0]))
Массив: [1, 2, 3] Тип: