Подключение к базам данных: библиотеки и примеры для Python

Расширенные примеры работы с базами данных

1. Пул соединений в psycopg2

Для уменьшения накладных расходов на создание соединений используется пул. Библиотека psycopg2.pool предоставляет ThreadedConnectionPool и SimpleConnectionPool.

from psycopg2.pool import SimpleConnectionPool

pool = SimpleConnectionPool(1, 10, user='user', password='pass', host='localhost', dbname='mydb')

# Получить соединение из пула
with pool.getconn() as conn:
    with conn.cursor() as cur:
        cur.execute("SELECT current_database();")
        print(f"Подключено к {cur.fetchone()[0]}")
# Соединение автоматически возвращается в пул

pool.closeall()

Подключено к mydb

Параметры: minconn (минимальное количество соединений), maxconn (максимальное). Пул потокобезопасен.

2. Транзакции с явным контролем

Управление транзакциями вручную полезно для сложных операций.

import sqlite3

conn = sqlite3.connect(':memory:')
cur = conn.cursor()

try:
    cur.execute("CREATE TABLE accounts (id INTEGER PRIMARY KEY, balance REAL)")
    cur.execute("INSERT INTO accounts VALUES (1, 1000.0)")
    cur.execute("INSERT INTO accounts VALUES (2, 500.0)")
    
    # Начало транзакции (автоматически при первом DML)
    cur.execute("UPDATE accounts SET balance = balance - 200 WHERE id = 1")
    cur.execute("UPDATE accounts SET balance = balance + 200 WHERE id = 2")
    conn.commit()  # Фиксация изменений
    print("Перевод выполнен успешно")
except Exception as e:
    conn.rollback()  # Откат при ошибке
    print(f"Ошибка: {e}")
finally:
    conn.close()

Перевод выполнен успешно

Если один из UPDATE вызовет ошибку, все изменения откатятся.

3. Использование ORM в SQLAlchemy

Пример объявления модели и запроса через ORM.

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, sessionmaker

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))

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

Session = sessionmaker(bind=engine)
session = Session()

user = User(name='Bob')
session.add(user)
session.commit()

# Запрос
result = session.query(User).filter_by(name='Bob').first()
print(f"ID: {result.id}, Name: {result.name}")
session.close()

ID: 1, Name: Bob

ORM автоматически генерирует SQL, скрывая диалектные особенности.

4. Асинхронное подключение к PostgreSQL через asyncpg

Пример с выполнением запроса и получением результатов.

import asyncio
import asyncpg

async def main():
    conn = await asyncpg.connect(user='user', password='pass',
                                 database='test', host='localhost')
    values = await conn.fetch('SELECT * FROM generate_series(1, 3)')
    for row in values:
        print(f"Число: {row[0]}")
    await conn.close()

asyncio.run(main())

Число: 1
Число: 2
Число: 3

asyncpg работает напрямую с asyncio, не блокируя event loop.

5. Подключение к MySQL с обработкой всех возможных исключений

import mysql.connector
from mysql.connector import errorcode

try:
    conn = mysql.connector.connect(
        host='localhost',
        database='testdb',
        user='root',
        password='',
        connection_timeout=5
    )
    cur = conn.cursor()
    cur.execute("SELECT 1+1")
    result = cur.fetchone()
    print(f"Результат: {result[0]}")
except mysql.connector.InterfaceError as err:
    print(f"Ошибка интерфейса: {err}")
except mysql.connector.DatabaseError as err:
    print(f"Ошибка базы данных: {err}")
except Exception as e:
    print(f"Общая ошибка: {e}")
finally:
    if 'conn' in locals() and conn.is_connected():
        cur.close()
        conn.close()

Результат: 2

Разделение исключений помогает точнее диагностировать проблемы.