Интеграция Python с реляционными СУБД: от SQLite до PostgreSQL
Раздел: Базы данных -> Популярные СУБД
Работа с SQL базами данных в Python
Как использовать встроенный модуль sqlite3 для работы с SQLite?
SQLite является встроенной реляционной базой данных, не требующей отдельного сервера. Модуль sqlite3 входит в стандартную библиотеку Python, поэтому для его использования не нужно устанавливать дополнительные пакеты. Это наиболее эффективное решение для небольших проектов, прототипов и тестирования.
import sqlite3
# Создание подключения к файлу базы данных
conn = sqlite3.connect('example.db')
cursor = conn.cursor()
# Создание таблицы
cursor.execute('''CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
email TEXT UNIQUE)''')
# Вставка данных
cursor.execute("INSERT INTO users (name, email) VALUES (?, ?)", ("Alice", "alice@example.com"))
conn.commit()
# Выборка данных
cursor.execute("SELECT * FROM users")
rows = cursor.fetchall()
for row in rows:
print(row)
conn.close()база данных sql python (sql база данных в python)
Возможные проблемы и ошибки:
- Ошибка sqlite3.OperationalError: no such table возникает, если таблица не создана перед запросом. Проверяется порядок выполнения команд.
- Ошибка sqlite3.IntegrityError: UNIQUE constraint failed при попытке вставить дубликат уникального поля. Следует обрабатывать исключения с помощью try/except.
- Блокировка файла базы данных при параллельных записях. Для многопользовательских сценариев лучше использовать клиент-серверные СУБД.
Как подключиться к PostgreSQL через psycopg2?
Для работы с PostgreSQL требуется установка драйвера psycopg2 (или его версии psycopg2-binary). Вариант подходит для промышленных приложений, где необходимы транзакции, сложные запросы и масштабирование.
import psycopg2
conn = psycopg2.connect(
host="localhost",
database="testdb",
user="postgres",
password="secret"
)
cursor = conn.cursor()
# Создание таблицы
cursor.execute("""
CREATE TABLE IF NOT EXISTS employees (
id SERIAL PRIMARY KEY,
name VARCHAR(100),
salary NUMERIC
)""")
# Вставка данных
cursor.execute("INSERT INTO employees (name, salary) VALUES (%s, %s)", ("Bob", 50000))
conn.commit()
# Чтение
cursor.execute("SELECT * FROM employees")
records = cursor.fetchall()
print(records)
conn.close()Распространенные ошибки:
- psycopg2.OperationalError: connection to server at ... failed - сервер PostgreSQL не запущен или параметры подключения неверны. Проверяется статус сервера.
- psycopg2.errors.UniqueViolation - нарушение уникальности. Используется обработка исключений с откатом транзакции (rollback).
- Утечка соединений: не забывать закрывать connection после работы или использовать контекстный менеджер with.
Как выполнять запросы к MySQL через pymysql?
Библиотека pymysql является чистым Python-драйвером для MySQL. Работа с ней аналогична psycopg2, но синтаксис плейсхолдеров отличается: используется %s.
import pymysql
conn = pymysql.connect(
host='localhost',
user='root',
password='password',
database='shop',
charset='utf8mb4'
)
cursor = conn.cursor()
# Создание таблицы
cursor.execute("""
CREATE TABLE IF NOT EXISTS products (
id INT AUTO_INCREMENT PRIMARY KEY,
title VARCHAR(200),
price DECIMAL(10,2)
)""")
# Вставка
sql = "INSERT INTO products (title, price) VALUES (%s, %s)"
cursor.execute(sql, ('Laptop', 750.00))
conn.commit()
# Выборка
cursor.execute("SELECT * FROM products")
for row in cursor.fetchall():
print(row)
conn.close()Типичные трудности:
- pymysql.err.OperationalError: (1045, ...) - неверные учетные данные. Проверяется пользователь и пароль.
- pymysql.err.ProgrammingError: 1064 - синтаксическая ошибка SQL. Рекомендуется экранировать значения через плейсхолдеры, а не подставлять строки.
- Отсутствие обработки транзакций: по умолчанию autocommit выключен, требуется явно вызывать commit().
Как абстрагироваться от конкретной СУБД с помощью SQLAlchemy?
SQLAlchemy - это мощная ORM (Object-Relational Mapping) и SQL Toolkit, позволяющая работать с различными базами данных через единый интерфейс. Вариант подходит для больших проектов, где важна гибкость смены СУБД.
from sqlalchemy import create_engine, Column, Integer, String, Float
from sqlalchemy.orm import declarative_base, sessionmaker
# Определение модели
Base = declarative_base()
class Product(Base):
__tablename__ = 'products'
id = Column(Integer, primary_key=True)
name = Column(String(100))
price = Column(Float)
# Подключение (SQLite для примера)
engine = create_engine('sqlite:///store.db')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
# Добавление записи
new_product = Product(name='Keyboard', price=45.99)
session.add(new_product)
session.commit()
# Чтение
products = session.query(Product).all()
for p in products:
print(p.name, p.price)
session.close()Сложности при использовании SQLAlchemy:
- Необходимо правильно настраивать строку подключения (connection URI). Ошибка в URI приводит к sqlalchemy.exc.NoSuchModuleError.
- Производительность ORM может быть ниже прямых SQL-запросов из-за накладных расходов на отображение объектов. Для высоконагруженных операций рекомендуется использовать Core API.
- Конфликт между сессиями при конкурентном доступе - требуется управление изоляцией транзакций.
Расширенные примеры работы с SQL базами данных в Python
Как выполнять транзакции с контролем ошибок?
Пример
import sqlite3
def transfer_funds(from_id, to_id, amount):
conn = sqlite3.connect('finance.db')
try:
cursor = conn.cursor()
cursor.execute("BEGIN")
cursor.execute("UPDATE accounts SET balance = balance - ? WHERE id = ?", (amount, from_id))
cursor.execute("UPDATE accounts SET balance = balance + ? WHERE id = ?", (amount, to_id))
conn.commit()
print("Транзакция выполнена")
except sqlite3.Error as e:
conn.rollback()
print(f"Ошибка: транзакция откачена: {e}")
finally:
conn.close()Вызов transfer_funds(1, 2, 100) переведет 100 единиц, либо откатит изменения при ошибке.
Как выполнять массовую вставку (bulk insert) в PostgreSQL?
Пример
import psycopg2
from psycopg2.extras import execute_values
conn = psycopg2.connect("dbname=test user=postgres password=secret")
cursor = conn.cursor()
data = [('Alice', 55000), ('Bob', 48000), ('Charlie', 62000)]
# execute_values - эффективная вставка многих строк
sql = "INSERT INTO employees (name, salary) VALUES %s"
execute_values(cursor, sql, data)
conn.commit()
cursor.execute("SELECT COUNT(*) FROM employees")
print(cursor.fetchone()[0]) # выведет 3
conn.close()Результат: 3 записи добавлены одной операцией.
Как использовать пул соединений для повышения производительности?
Пример
from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool
# Создание движка с пулом (по умолчанию QueuePool)
engine = create_engine(
'postgresql://user:pass@localhost/mydb',
poolclass=QueuePool,
pool_size=5,
max_overflow=10
)
# Использование в нескольких потоках
with engine.connect() as conn:
result = conn.execute("SELECT version()")
print(result.fetchone())Вывод: ('PostgreSQL 15.x on x86_64-pc-linux-gnu',)Как загрузить результаты SQL-запроса в Pandas DataFrame?
Пример
import pandas as pd
import sqlite3
conn = sqlite3.connect('example.db')
df = pd.read_sql_query("SELECT * FROM users", conn)
print(df.head())
conn.close()id name email 0 1 Alice alice@example.com
Как использовать raw SQL с SQLAlchemy Core без ORM?
Пример
from sqlalchemy import create_engine, text
engine = create_engine('sqlite:///test.db')
with engine.connect() as conn:
# Параметризованный запрос
result = conn.execute(text("SELECT * FROM products WHERE price > :min_price"), {"min_price": 100})
for row in result:
print(row)Вывод: (2, 'Laptop', 750.0) и другие строки с ценой более 100.
Как асинхронно работать с базой данных через aiosqlite?
Пример
import asyncio
import aiosqlite
async def main():
async with aiosqlite.connect('async.db') as db:
await db.execute('CREATE TABLE IF NOT EXISTS tasks (id INTEGER PRIMARY KEY, description TEXT)')
await db.execute('INSERT INTO tasks (description) VALUES (?)', ('Finish report',))
await db.commit()
async with db.execute('SELECT * FROM tasks') as cursor:
async for row in cursor:
print(row)
asyncio.run(main())Вывод: (1, 'Finish report')
SQL база данных в Python - comments
En
база данных sql python (python)