Python и системы управления базами: от встроенного sqlite до ORM

Раздел: Базы данных -> Работа с базами данных

Основные подходы к взаимодействию с базами данных в Python

При разработке на языке Python часто возникает необходимость сохранять, извлекать и обрабатывать данные. Для этого применяются различные библиотеки и драйверы. Выбор подходящего инструмента зависит от типа базы данных, требований к производительности, асинхронности и сложности запросов. Рассмотрим наиболее популярные решения.

Как обеспечить удобную работу с реляционными базами данных, используя объектно-реляционное отображение?

SQLAlchemy - мощная ORM (Object-Relational Mapping), которая позволяет работать с базами данных через объекты Python, скрывая синтаксис SQL. Это основное эффективное решение для большинства проектов, где требуется гибкость, миграции и поддержка различных СУБД (PostgreSQL, MySQL, SQLite и др.).

# Установка: pip install sqlalchemy

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

# Определяем модель
Base = declarative_base()

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

# Подключаемся к SQLite
engine = create_engine('sqlite:///example.db')
Base.metadata.create_all(engine)

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

# Создаём запись
new_user = User(name='Анна', age=28)
session.add(new_user)
session.commit()

# Запрос
for user in session.query(User).all():
    print(user.name, user.age)

создание базы данных на python (создание базы данных с помощью python)

Возможные проблемы:

  • Несовместимость версий SQLAlchemy и драйвера базы данных
  • Ошибки маппинга при неправильном определении связей
  • Проблемы с сессиями в многопоточном окружении

Решение: Использовать пулы соединений (create_engine с параметром pool_size) и явно закрывать сессии. Для миграций применять Alembic.

Цель и случаи использования: SQLAlchemy подходит для приложений, где важна читаемость кода, переносимость между СУБД и возможность работы с объектами. Это стандарт де-факто для веб-фреймворков (Flask, FastAPI).

Когда достаточно лёгкого встроенного решения без установки сервера?

Модуль sqlite3 - встроенная поддержка SQLite, не требующая дополнительных зависимостей. Идеален для прототипирования, тестирования и приложений с малым числом одновременных запросов.

import sqlite3

conn = sqlite3.connect('test.db')
cursor = conn.cursor()
cursor.execute('''CREATE TABLE IF NOT EXISTS items (id INTEGER PRIMARY KEY, name TEXT)''')
cursor.execute('INSERT INTO items (name) VALUES (?)', ('Книга',))
conn.commit()

for row in cursor.execute('SELECT * FROM items'):
    print(row)

conn.close()

база данных на python (база данных на python)

Ограничения: Нет поддержки конкурентных записей; слабая типизация; отсутствие хранимых процедур.

Как получить максимальную производительность при работе с PostgreSQL?

Psycopg2 - один из самых популярных драйверов для PostgreSQL. Он предоставляет полный доступ к возможностям СУБД: транзакции, пулы соединений, копирование данных.

import psycopg2

conn = psycopg2.connect(
    host='localhost',
    dbname='mydb',
    user='user',
    password='secret'
)
cur = conn.cursor()
cur.execute('SELECT version()')
print(cur.fetchone())
cur.close()
conn.close()

Python интерфейс базы данных (интерфейс базы данных в python)

Типичные ошибки: Утечка соединений (необходимо использовать контекстный менеджер или пул через psycopg2.pool). Неправильная обработка исключений при разрыве связи.

Как выполнять асинхронные запросы к базам данных?

Для асинхронного кода существуют специальные драйверы: asyncpg (для PostgreSQL) и aiosqlite (для SQLite). Они работают с asyncio.

import asyncio
import asyncpg

async def run():
    conn = await asyncpg.connect(user='user', password='secret', database='mydb', host='localhost')
    values = await conn.fetch('''SELECT * FROM users''')
    await conn.close()
    return values

asyncio.run(run())

Fastapi python postgresql (работа с postgresql в fastapi)

Сложности: Смешивание синхронного и асинхронного кода (требуется отдельный event loop), управление сессиями в веб-приложениях.

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

MongoDB с драйвером pymongo используется для документо-ориентированных данных, где схема может меняться. Подходит для логов, каталогов, контента.

from pymongo import MongoClient

client = MongoClient('mongodb://localhost:27017/')
db = client['mydb']
collection = db['items']
collection.insert_one({'name': 'Товар', 'price': 100})
for doc in collection.find():
    print(doc)

Недостатки: Сложность выполнения JOIN-подобных операций, ограниченная поддержка транзакций (до версии 4.0).

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

Ниже приведены примеры, которые демонстрируют более сложные сценарии: миграции, асинхронные операции, bulk-вставки, транзакции, работа с JSON полями в PostgreSQL.

Миграции с помощью Alembic (для SQLAlchemy)

Пример 
# После установки alembic (pip install alembic)
# Инициализация: alembic init alembic
# Создание миграции: alembic revision --autogenerate -m "add column"
# Применение: alembic upgrade head

Bulk-вставка с SQLAlchemy

Пример 
from sqlalchemy.orm import Session

users_data = [{'name': 'Иван', 'age': 30}, {'name': 'Мария', 'age': 25}]
session = Session(engine)
session.bulk_insert_mappings(User, users_data)
session.commit()

Асинхронный SQLAlchemy (через asyncpg)

Пример 
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

async_engine = create_async_engine('postgresql+asyncpg://user:pass@localhost/mydb')
AsyncSession = sessionmaker(async_engine, class_=AsyncSession)

async def get_users():
    async with AsyncSession() as session:
        result = await session.execute('SELECT * FROM users')
        return result.fetchall()

Транзакции с явным ручным управлением (psycopg2)

Пример 
import psycopg2

conn = psycopg2.connect('...')
conn.autocommit = False
try:
    cur = conn.cursor()
    cur.execute('UPDATE accounts SET balance = balance - 100 WHERE id = 1')
    cur.execute('UPDATE accounts SET balance = balance + 100 WHERE id = 2')
    conn.commit()
except:
    conn.rollback()
    raise
finally:
    conn.close()

Работа с JSONB в PostgreSQL через SQLAlchemy

Пример 
from sqlalchemy.dialects.postgresql import JSONB

class Product(Base):
    __tablename__ = 'products'
    id = Column(Integer, primary_key=True)
    attributes = Column(JSONB)

session.add(Product(attributes={'color': 'red', 'size': 42}))
session.commit()

Результат выполнения запроса к SQLite с группировкой

Пример 
import sqlite3
conn = sqlite3.connect(':memory:')
c = conn.cursor()
c.execute('CREATE TABLE sales (product TEXT, amount INT)')
c.execute('INSERT INTO sales VALUES ("ПК", 1200), ("ПК", 800), ("Ноутбук", 1500)')
c.execute('SELECT product, SUM(amount) FROM sales GROUP BY product')
print(c.fetchall())
conn.close()
[('Ноутбук', 1500), ('ПК', 2000)]

Обработка ошибок подключения к PostgreSQL

Пример 
import psycopg2
from psycopg2 import OperationalError

try:
    conn = psycopg2.connect(host='nonexistent', dbname='test')
except OperationalError as e:
    print(f"Ошибка подключения: {e}")

Примечание: Для каждого примера требуется предварительная установка соответствующих библиотек (psycopg2-binary, asyncpg, pymongo, alembic).

База данных на Python - comments

En
база данных на python (python)