ТОП-5 Python библиотек для эффективной работы с базами данных

pip install sqlalchemy

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

# Создаём подключение к БД
engine = create_engine('sqlite:///example.db', echo=True)
Base = declarative_base()

# Определяем модели
class User(Base):
__tablename__ = 'users'

id = Column(Integer, primary_key=True)
name = Column(String)
email = Column(String, unique=True)
posts = relationship("Post", back_populates="author")

def __repr__(self):
return f"<User(name='{self.name}', email='{self.email}')>"

class Post(Base):
__tablename__ = 'posts'

id = Column(Integer, primary_key=True)
title = Column(String)
content = Column(String)
user_id = Column(Integer, ForeignKey('users.id'))
author = relationship("User", back_populates="posts")

def __repr__(self):
return f"<Post(title='{self.title}')>"

# Создаём таблицы в БД
Base.metadata.create_all(engine)

# Создаём сессию для работы с БД
Session = sessionmaker(bind=engine)
session = Session()

# CRUD операции
# Create
new_user = User(name="John Doe", email="john@example.com")
session.add(new_user)
session.commit()

# Добавляем пост для пользователя
new_post = Post(title="First Post", content="Hello, SQLAlchemy!", author=new_user)
session.add(new_post)
session.commit()

# Read
users = session.query(User).all()
for user in users:
print(user.name, user.email)
for post in user.posts:
print(f"- {post.title}: {post.content}")

# Update
user_to_update = session.query(User).filter_by(name="John Doe").first()
user_to_update.name = "John Smith"
session.commit()

# Delete
post_to_delete = session.query(Post).first()
session.delete(post_to_delete)
session.commit()

from sqlalchemy import Table, MetaData, select, join

metadata = MetaData()
users = Table('users', metadata, autoload=True, autoload_with=engine)
posts = Table('posts', metadata, autoload=True, autoload_with=engine)

# Сложный запрос с JOIN
query = select([users.c.name, posts.c.title]).select_from(
join(users, posts, users.c.id == posts.c.user_id)
).where(users.c.name.like('%John%'))

result = engine.execute(query)
for row in result:
print(row)

# models.py в Django
from django.db import models

class Author(models.Model):
name = models.CharField(max_length=100)
email = models.EmailField(unique=True)

def __str__(self):
return self.name

class Book(models.Model):
title = models.CharField(max_length=200)
pages = models.IntegerField()
author = models.ForeignKey(Author, on_delete=models.CASCADE, related_name='books')
published_date = models.DateField()

def __str__(self):
return self.title

# CRUD операции в Django
# Create
author = Author.objects.create(name="Jane Doe", email="jane@example.com")
Book.objects.create(
title="Python Best Practices",
pages=350,
author=author,
published_date="2023-01-15"
)

# Read
books = Book.objects.filter(pages__gt=300).select_related('author')
for book in books:
print(f"{book.title} by {book.author.name}")

# Update
author = Author.objects.get(email="jane@example.com")
author.name = "Jane Smith"
author.save()

# Delete
Book.objects.filter(title__contains="Legacy").delete()

# Пример использования Peewee
from peewee import *
import datetime

db = SqliteDatabase('library.db')

class BaseModel(Model):
class Meta:
database = db

class Author(BaseModel):
name = CharField()
email = CharField(unique=True)

class Book(BaseModel):
title = CharField()
pages = IntegerField()
author = ForeignKeyField(Author, backref='books')
published_date = DateField()

# Создаём таблицы
db.create_tables([Author, Book])

# CRUD операции в Peewee
# Create
author = Author.create(name="Alex Johnson", email="alex@example.com")
Book.create(
title="Python Data Analysis",
pages=275,
author=author,
published_date=datetime.date(2022, 5, 10)
)

# Read
books = Book.select().where(Book.pages > 200).join(Author)
for book in books:
print(f"{book.title} by {book.author.name}")

# Update
query = Author.update(name="Alexander Johnson").where(Author.email == "alex@example.com")
query.execute()

# Delete
Book.delete().where(Book.title.contains("Outdated")).execute()

import pandas as pd
import sqlite3

# Создаём соединение с SQLite
conn = sqlite3.connect('example.db')

# Чтение данных из SQL-запроса
df = pd.read_sql_query("SELECT * FROM users JOIN posts ON users.id = posts.user_id", conn)

# Анализ данных
print(df.head())
print(df.describe())

# Группировка и агрегация
grouped = df.groupby('name').agg({
'title': 'count',
'content': lambda x: ', '.join(x)
})
print(grouped)

# Фильтрация данных
filtered = df[(df['name'].str.contains('John')) & (df['title'].str.len() > 10)]
print(filtered)

# Запись результатов обратно в БД
filtered.to_sql('filtered_posts', conn, if_exists='replace', index=False)

# Можно также работать с CSV, Excel, JSON, Parquet и другими форматами
df.to_csv('export.csv')
excel_data = pd.read_excel('data.xlsx')

# Преобразование данных
df['content_length'] = df['content'].str.len()
df['date_created'] = pd.to_datetime(df['date_created'])

from pymongo import MongoClient
import datetime

# Подключение к MongoDB
client = MongoClient('mongodb://localhost:27017/')
db = client['blog_database']
users_collection = db['users']
posts_collection = db['posts']

# Create – добавление документов
user_data = {
"name": "Robert Johnson",
"email": "robert@example.com",
"registration_date": datetime.datetime.now(),
"interests": ["Python", "Data Science", "Machine Learning"]
}
user_id = users_collection.insert_one(user_data).inserted_id

post_data = {
"title": "Working with MongoDB in Python",
"content": "MongoDB is a document database that offers...",
"author_id": user_id,
"tags": ["database", "nosql", "python"],
"comments": [
{"user": "alice", "text": "Great post!"},
{"user": "bob", "text": "Thanks for sharing"}
]
}
posts_collection.insert_one(post_data)

# Read – чтение документов
# Простой поиск
user = users_collection.find_one({"email": "robert@example.com"})
print(user)

# Сложный поиск с условиями
posts = posts_collection.find({"tags": "python"}).sort("title")
for post in posts:
print(f"{post['title']} by {post['author_id']}")

# Агрегации
pipeline = [
{"$match": {"tags": "python"}},
{"$group": {"_id": "$author_id", "post_count": {"$sum": 1}}},
{"$sort": {"post_count": -1}}
]
authors_by_posts = list(posts_collection.aggregate(pipeline))

# Update – обновление документов
result = users_collection.update_one(
{"email": "robert@example.com"},
{"$set": {"name": "Bob Johnson"}, "$push": {"interests": "Web Development"}}
)
print(f"Modified {result.modified_count} document(s)")

# Delete – удаление документов
result = posts_collection.delete_many({"tags": "outdated"})
print(f"Deleted {result.deleted_count} document(s)")

import pandas as pd
from pymongo import MongoClient

# Получаем данные из MongoDB
client = MongoClient('mongodb://localhost:27017/')
db = client['analytics']
data = list(db.events.find({"event_type": "purchase"}))

# Преобразуем в DataFrame для анализа
df = pd.DataFrame(data)

# Очистка и обработка
df['timestamp'] = pd.to_datetime(df['timestamp'])
df['day'] = df['timestamp'].dt.date
df['amount'] = pd.to_numeric(df['amount'])

# Агрегация и анализ
daily_sales = df.groupby('day').agg({
'amount': ['sum', 'mean', 'count'],
'user_id': 'nunique'
})

# Экспорт результатов
daily_sales.to_csv('sales_report.csv')

# Или сохранение результатов обратно в MongoDB
result_records = daily_sales.to_dict('records')
db.sales_reports.insert_many(result_records)

# Паттерн пула соединений с использованием контекстного менеджера
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from contextlib import contextmanager

engine = create_engine('postgresql://user:password@localhost/dbname', 
pool_size=10, max_overflow=20)
Session = scoped_session(sessionmaker(bind=engine))

@contextmanager
def session_scope():
"""Обеспечивает транзакционную область для операций с БД"""
session = Session()
try:
yield session
session.commit()
except Exception as e:
session.rollback()
raise
finally:
session.close()

# Использование
def create_user(name, email):
with session_scope() as session:
user = User(name=name, email=email)
session.add(user)
return user.id # ID будет доступен благодаря flush в commit()

# Пакетная вставка с использованием SQLAlchemy
def bulk_insert_users(user_data, batch_size=1000):
"""Вставка большого количества пользователей порциями"""
with session_scope() as session:
for i in range(0, len(user_data), batch_size):
batch = user_data[i:i+batch_size]
user_objects = [User(**data) for data in batch]
session.bulk_save_objects(user_objects)
# Очищаем сессию после каждого пакета, чтобы не накапливать объекты в памяти
session.flush()
session.expunge_all()

# Фабрика подключений к БД на основе конфигурации
import os
from sqlalchemy import create_engine
from dotenv import load_dotenv

load_dotenv() # Загружаем переменные окружения из .env файла

def get_database_url():
"""Формирует URL подключения к БД на основе переменных окружения"""
env = os.getenv('ENVIRONMENT', 'development')

if env == 'production':
return os.getenv('DATABASE_URL')
elif env == 'testing':
return 'sqlite:///:memory:'
else: # development
return 'sqlite:///dev.db'

def create_db_engine(echo=False):
"""Создаёт движок БД с нужными параметрами"""
url = get_database_url()
return create_engine(url, echo=echo)

# Паттерн репозитория для инкапсуляции логики доступа к данным
class UserRepository:
def __init__(self, session):
self.session = session

def get_by_id(self, user_id):
return self.session.query(User).filter_by(id=user_id).first()

def get_by_email(self, email):
return self.session.query(User).filter_by(email=email).first()

def create(self, name, email):
user = User(name=name, email=email)
self.session.add(user)
self.session.flush() # Чтобы получить ID без коммита
return user

def update(self, user_id, **kwargs):
rows_affected = self.session.query(User).filter_by(id=user_id).update(kwargs)
return rows_affected > 0

def delete(self, user_id):
user = self.get_by_id(user_id)
if user:
self.session.delete(user)
return True
return False

# Использование репозитория
with session_scope() as session:
user_repo = UserRepository(session)
user = user_repo.get_by_email("user@example.com")
if user:
user_repo.update(user.id, name="New Name")

# Иерархия исключений для работы с БД
class DatabaseError(Exception):
"""Базовый класс для всех исключений, связанных с базой данных"""
pass

class EntityNotFoundError(DatabaseError):
"""Возникает, когда запрашиваемая сущность не найдена в БД"""
pass

class DuplicateEntityError(DatabaseError):
"""Возникает при попытке создать дубликат уникальной сущности"""
pass

class RepositoryWithErrorHandling:
def get_user_by_id(self, user_id):
user = session.query(User).filter_by(id=user_id).first()
if not user:
raise EntityNotFoundError(f"User with ID {user_id} not found")
return user

def create_user(self, email, name):
try:
user = User(email=email, name=name)
session.add(user)
session.flush()
return user
except IntegrityError as e:
if "unique constraint" in str(e).lower() and "email" in str(e).lower():
raise DuplicateEntityError(f"User with email {email} already exists")
raise DatabaseError(f"Failed to create user: {e}")