Работа с базами данных в Python: полное руководство с примерами - Часть 1

# Установка популярных библиотек для работы с БД
pip install sqlalchemy psycopg2-binary mysql-connector-python pymongo redis

# Или создайте requirements.txt файл:
echo "sqlalchemy>=2.0.0
psycopg2-binary>=2.9.0
mysql-connector-python>=8.0.0
pymongo>=4.0.0
redis>=4.0.0" > requirements.txt

pip install -r requirements.txt

import sqlite3
from datetime import datetime
from contextlib import contextmanager

# Контекстный менеджер для автоматического закрытия соединения
@contextmanager
def get_db_connection(db_path="tasks.db"):
    """Контекстный менеджер для работы с БД"""
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row  # Для доступа к столбцам по имени
    try:
        yield conn
    finally:
        conn.close()

def init_database():
    """Инициализация базы данных и создание таблиц"""
    with get_db_connection() as conn:
        cursor = conn.cursor()
        
        # Создание таблицы пользователей
        cursor.execute('''
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            username TEXT UNIQUE NOT NULL,
            email TEXT UNIQUE NOT NULL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
        ''')
        
        # Создание таблицы задач
        cursor.execute('''
        CREATE TABLE IF NOT EXISTS tasks (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            user_id INTEGER,
            title TEXT NOT NULL,
            description TEXT,
            status TEXT CHECK(status IN ('pending', 'in_progress', 'completed')) DEFAULT 'pending',
            priority INTEGER CHECK(priority BETWEEN 1 AND 5) DEFAULT 3,
            due_date TIMESTAMP,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            FOREIGN KEY (user_id) REFERENCES users (id) ON DELETE CASCADE
        )
        ''')
        
        # Создание индексов для ускорения поиска
        cursor.execute('CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks(status)')
        cursor.execute('CREATE INDEX IF NOT EXISTS idx_tasks_user_id ON tasks(user_id)')
        cursor.execute('CREATE INDEX IF NOT EXISTS idx_tasks_due_date ON tasks(due_date)')
        
        conn.commit()
        print("База данных инициализирована успешно")

class TaskManager:
    """Класс для управления задачами"""
    
    def __init__(self, db_path="tasks.db"):
        self.db_path = db_path
    
    def add_user(self, username, email):
        """Добавление нового пользователя"""
        with get_db_connection(self.db_path) as conn:
            cursor = conn.cursor()
            try:
                cursor.execute(
                    'INSERT INTO users (username, email) VALUES (?, ?)',
                    (username, email)
                )
                conn.commit()
                print(f"Пользователь '{username}' добавлен с ID: {cursor.lastrowid}")
                return cursor.lastrowid
            except sqlite3.IntegrityError as e:
                print(f"Ошибка: {e}")
                return None
    
    def add_task(self, user_id, title, description="", priority=3, due_date=None):
        """Добавление новой задачи"""
        with get_db_connection(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute('''
                INSERT INTO tasks (user_id, title, description, priority, due_date)
                VALUES (?, ?, ?, ?, ?)
            ''', (user_id, title, description, priority, due_date))
            conn.commit()
            print(f"Задача '{title}' добавлена с ID: {cursor.lastrowid}")
            return cursor.lastrowid
    
    def get_tasks(self, user_id=None, status=None, priority=None):
        """Получение задач с фильтрацией"""
        with get_db_connection(self.db_path) as conn:
            cursor = conn.cursor()
            
            query = "SELECT * FROM tasks"
            params = []
            conditions = []
            
            if user_id:
                conditions.append("user_id = ?")
                params.append(user_id)
            
            if status:
                conditions.append("status = ?")
                params.append(status)
            
            if priority:
                conditions.append("priority = ?")
                params.append(priority)
            
            if conditions:
                query += " WHERE " + " AND ".join(conditions)
            
            query += " ORDER BY priority DESC, due_date ASC"
            
            cursor.execute(query, params)
            return [dict(row) for row in cursor.fetchall()]
    
    def update_task_status(self, task_id, status):
        """Обновление статуса задачи"""
        with get_db_connection(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute('''
                UPDATE tasks
                SET status = ?, updated_at = CURRENT_TIMESTAMP
                WHERE id = ?
            ''', (status, task_id))
            conn.commit()
            return cursor.rowcount > 0
    
    def delete_task(self, task_id):
        """Удаление задачи"""
        with get_db_connection(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute('DELETE FROM tasks WHERE id = ?', (task_id,))
            conn.commit()
            return cursor.rowcount > 0
    
    def get_task_statistics(self, user_id):
        """Статистика по задачам пользователя"""
        with get_db_connection(self.db_path) as conn:
            cursor = conn.cursor()
            
            # Общая статистика
            cursor.execute('''
                SELECT
                    COUNT(*) as total,
                    SUM(CASE WHEN status = 'completed' THEN 1 ELSE 0 END) as completed,
                    SUM(CASE WHEN status = 'in_progress' THEN 1 ELSE 0 END) as in_progress,
                    SUM(CASE WHEN status = 'pending' THEN 1 ELSE 0 END) as pending
                FROM tasks
                WHERE user_id = ?
            ''', (user_id,))
            
            stats = dict(cursor.fetchone())
            
            # Задачи по приоритету
            cursor.execute('''
                SELECT priority, COUNT(*) as count
                FROM tasks
                WHERE user_id = ?
                GROUP BY priority
                ORDER BY priority DESC
            ''', (user_id,))
            
            stats['by_priority'] = [dict(row) for row in cursor.fetchall()]
            
            # Просроченные задачи
            cursor.execute('''
                SELECT COUNT(*) as overdue
                FROM tasks
                WHERE user_id = ?
                AND status != 'completed'
                AND due_date < DATE('now')
            ''', (user_id,))
            
            stats['overdue'] = cursor.fetchone()['overdue']
            
            return stats

def display_tasks(tasks):
    """Красивый вывод списка задач"""
    if not tasks:
        print("Задачи не найдены")
        return
    
    print("\n" + "="*80)
    print(f"{'ID':<5} {'Заголовок':<30} {'Статус':<15} {'Приоритет':<10} {'Срок':<15}")
    print("="*80)
    
    for task in tasks:
        due_date = task['due_date'] or "Нет срока"
        print(f"{task['id']:<5} {task['title'][:28]:<30} {task['status']:<15} {task['priority']:<10} {str(due_date):<15}")
    
    print("="*80)

def main():
    """Основная функция приложения"""
    # Инициализация базы данных
    init_database()
    
    # Создание менеджера задач
    manager = TaskManager()
    
    # Добавление тестового пользователя
    user_id = manager.add_user("test_user", "test@example.com")
    
    if not user_id:
        # Если пользователь уже существует, найдем его ID
        with get_db_connection() as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT id FROM users WHERE username = ?", ("test_user",))
            user = cursor.fetchone()
            user_id = user['id'] if user else None
    
    if not user_id:
        print("Не удалось получить ID пользователя")
        return
    
    # Добавление тестовых задач
    sample_tasks = [
        ("Изучить Python", "Освоить основы программирования на Python", 5, "2024-12-31"),
        ("Купить продукты", "Молоко, хлеб, яйца", 3, "2024-03-15"),
        ("Записаться на курс", "Курс по базам данных", 4, None),
        ("Прочитать книгу", "Чистый код", 2, "2024-04-01"),
    ]
    
    for title, desc, priority, due_date in sample_tasks:
        manager.add_task(user_id, title, desc, priority, due_date)
    
    # Демонстрация работы
    print("\nВсе задачи пользователя:")
    all_tasks = manager.get_tasks(user_id=user_id)
    display_tasks(all_tasks)
    
    print("\nТолько незавершенные задачи:")
    pending_tasks = manager.get_tasks(user_id=user_id, status="pending")
    display_tasks(pending_tasks)
    
    # Обновление статуса задачи
    if all_tasks:
        task_id = all_tasks[0]['id']
        manager.update_task_status(task_id, "completed")
        print(f"\nЗадача {task_id} отмечена как выполненная")
    
    # Статистика
    print("\nСтатистика по задачам:")
    stats = manager.get_task_statistics(user_id)
    print(f"Всего задач: {stats['total']}")
    print(f"Выполнено: {stats['completed']}")
    print(f"В процессе: {stats['in_progress']}")
    print(f"Ожидают: {stats['pending']}")
    print(f"Просрочено: {stats['overdue']}")

if __name__ == "__main__":
    main()

from sqlalchemy import create_engine, Column, Integer, String, Text, DateTime, ForeignKey, Enum, CheckConstraint
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
from datetime import datetime
import enum

# Определение базового класса
Base = declarative_base()

# Перечисление для статуса задач
class TaskStatus(enum.Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"

class User(Base):
    """Модель пользователя"""
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(100), unique=True, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    # Связь один-ко-многим с задачами
    tasks = relationship("Task", back_populates="user", cascade="all, delete-orphan")
    
    def __repr__(self):
        return f"<User(id={self.id}, username='{self.username}')>"

class Task(Base):
    """Модель задачи"""
    __tablename__ = 'tasks'
    
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('users.id', ondelete='CASCADE'))
    title = Column(String(200), nullable=False)
    description = Column(Text)
    status = Column(Enum(TaskStatus), default=TaskStatus.PENDING)
    priority = Column(Integer, CheckConstraint('priority >= 1 AND priority <= 5'), default=3)
    due_date = Column(DateTime, nullable=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    
    # Связь многие-к-одному с пользователем
    user = relationship("User", back_populates="tasks")
    
    def __repr__(self):
        return f"<Task(id={self.id}, title='{self.title}', status='{self.status.value}')>"

class SQLAlchemyTaskManager:
    """Менеджер задач с использованием SQLAlchemy"""
    
    def __init__(self, database_url="sqlite:///tasks_alchemy.db"):
        self.engine = create_engine(database_url, echo=False)
        self.Session = sessionmaker(bind=self.engine)
        Base.metadata.create_all(self.engine)
    
    def add_user(self, username, email):
        """Добавление пользователя"""
        session = self.Session()
        try:
            user = User(username=username, email=email)
            session.add(user)
            session.commit()
            print(f"Пользователь добавлен: {user}")
            return user.id
        except Exception as e:
            session.rollback()
            print(f"Ошибка: {e}")
            return None
        finally:
            session.close()
    
    def add_task(self, user_id, title, **kwargs):
        """Добавление задачи"""
        session = self.Session()
        try:
            task = Task(user_id=user_id, title=title, **kwargs)
            session.add(task)
            session.commit()
            print(f"Задача добавлена: {task}")
            return task.id
        except Exception as e:
            session.rollback()
            print(f"Ошибка: {e}")
            return None
        finally:
            session.close()
    
    def get_tasks_with_users(self):
        """Получение задач с информацией о пользователях (JOIN)"""
        session = self.Session()
        try:
            # Пример сложного запроса с JOIN
            results = session.query(Task, User).join(User).all()
            
            for task, user in results:
                print(f"Задача: {task.title}, Пользователь: {user.username}, Статус: {task.status.value}")
            
            return results
        finally:
            session.close()
    
    def bulk_operations(self):
        """Пример массовых операций"""
        session = self.Session()
        try:
            # Массовое обновление
            session.query(Task).filter(
                Task.status == TaskStatus.PENDING,
                Task.due_date < datetime.utcnow()
            ).update(
                {Task.status: TaskStatus.IN_PROGRESS},
                synchronize_session=False
            )
            
            # Массовое удаление старых завершенных задач
            one_month_ago = datetime.utcnow().replace(day=datetime.utcnow().day - 30)
            deleted_count = session.query(Task).filter(
                Task.status == TaskStatus.COMPLETED,
                Task.updated_at < one_month_ago
            ).delete(synchronize_session=False)
            
            session.commit()
            print(f"Удалено {deleted_count} старых задач")
            
        except Exception as e:
            session.rollback()
            print(f"Ошибка: {e}")
        finally:
            session.close()

def demo_sqlalchemy():
    """Демонстрация работы SQLAlchemy"""
    manager = SQLAlchemyTaskManager()
    
    # Создание пользователя
    user_id = manager.add_user("alice", "alice@example.com")
    
    if user_id:
        # Добавление задач
        manager.add_task(
            user_id=user_id,
            title="Изучить SQLAlchemy",
            description="Освоить работу с ORM в Python",
            priority=5,
            due_date=datetime(2024, 12, 31)
        )
        
        manager.add_task(
            user_id=user_id,
            title="Написать тесты",
            priority=4
        )
    
    # Получение задач с пользователями
    manager.get_tasks_with_users()
    
    # Массовые операции
    manager.bulk_operations()

# Запуск демо
# demo_sqlalchemy()