SQLite в Python: руководство для начинающих и профи

Основы работы с SQLite в Python

Работа с базами данных представляет собой важный аспект практически любого современного приложения. Однако далеко не всегда необходимы сложные и ресурсоёмкие системы управления базами данных, такие как MySQL или PostgreSQL. Для небольших проектов, создания прототипов или разработки локальных приложений отлично подходит встроенная в Python система управления базами данных — SQLite.

В этом руководстве мы подробно разберём принципы работы с SQLite в Python. Вы научитесь выполнять основные операции с базой данных. Также рассмотрим практические примеры использования команды UPDATE в SQLite с помощью Python.

Что представляет собой SQLite и её преимущества

SQLite — это легковесная встроенная реляционная база данных. Она не требует отдельного сервера для функционирования. База данных SQLite хранится непосредственно в одном файле на диске компьютера.

Основные преимущества SQLite:

Встроена в стандартную библиотеку Python через модуль sqlite3
Не требует установки и настройки отдельного сервера
Отлично подходит для локальных приложений и мобильных решений
Простота использования и высокая скорость работы на малых объёмах данных
Поддерживает SQL-запросы стандарта
Кроссплатформенность и портативность
Надёжность и стабильность работы

Подключение к базе данных SQLite

Для работы с SQLite используется стандартный модуль sqlite3. Устанавливать его дополнительно не требуется, поскольку он уже входит в стандартную поставку Python.

import sqlite3

# Создание или подключение к базе данных
conn = sqlite3.connect('example.db')

# Создание курсора для выполнения SQL-запросов
cursor = conn.cursor()

При выполнении данного кода происходит создание файла базы данных example.db в текущей директории. Если файл уже существует, то происходит подключение к существующей базе данных.

Создание структуры базы данных

Создание таблиц

Создание таблицы в SQLite выполняется с помощью команды CREATE TABLE. Рассмотрим пример создания таблицы для хранения информации о пользователях:

cursor.execute('''
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        age INTEGER,
        email TEXT UNIQUE
    )
''')

conn.commit()

В данном примере создаётся таблица users с четырьмя столбцами:

id — уникальный идентификатор с автоинкрементом
name — имя пользователя (обязательное поле)
age — возраст пользователя
email — электронная почта (уникальное значение)

Создание индексов для оптимизации

Для повышения производительности запросов можно создавать индексы:

cursor.execute('''
    CREATE INDEX IF NOT EXISTS idx_users_email ON users(email)
''')

cursor.execute('''
    CREATE INDEX IF NOT EXISTS idx_users_age ON users(age)
''')

conn.commit()

Операции с данными

Добавление данных в таблицу

Вставка одной записи в таблицу выполняется с помощью команды INSERT:

cursor.execute('''
    INSERT INTO users (name, age, email) 
    VALUES (?, ?, ?)
''', ('Alice', 30, 'alice@example.com'))

conn.commit()

Массовое добавление данных

Для добавления нескольких записей одновременно используется метод executemany:

users_data = [
    ('Bob', 25, 'bob@example.com'),
    ('Charlie', 35, 'charlie@example.com'),
    ('Diana', 28, 'diana@example.com'),
    ('Eve', 32, 'eve@example.com')
]

cursor.executemany('''
    INSERT INTO users (name, age, email) 
    VALUES (?, ?, ?)
''', users_data)

conn.commit()

Чтение данных из таблицы

Для извлечения данных используется команда SELECT. Существует несколько способов получения результатов:

# Получение всех записей
cursor.execute('SELECT * FROM users')
all_rows = cursor.fetchall()

for row in all_rows:
    print(f"ID: {row[0]}, Имя: {row[1]}, Возраст: {row[2]}, Email: {row[3]}")

# Получение одной записи
cursor.execute('SELECT * FROM users WHERE name = ?', ('Alice',))
single_row = cursor.fetchone()
print(single_row)

# Получение ограниченного количества записей
cursor.execute('SELECT * FROM users ORDER BY age DESC')
limited_rows = cursor.fetchmany(3)
print(limited_rows)

Фильтрация и сортировка данных

# Поиск пользователей по возрасту
cursor.execute('SELECT * FROM users WHERE age > ? ORDER BY age', (25,))
filtered_users = cursor.fetchall()

# Поиск по частичному совпадению имени
cursor.execute('SELECT * FROM users WHERE name LIKE ?', ('%li%',))
matching_users = cursor.fetchall()

# Группировка и агрегация
cursor.execute('SELECT COUNT(*), AVG(age) FROM users')
stats = cursor.fetchone()
print(f"Количество пользователей: {stats[0]}, Средний возраст: {stats[1]:.2f}")

Обновление данных с помощью команды UPDATE

Команда UPDATE позволяет изменять существующие записи в таблице. Это одна из наиболее важных операций при работе с базой данных.

Обновление одной записи

cursor.execute('''
    UPDATE users 
    SET age = ? 
    WHERE name = ?
''', (32, 'Alice'))

conn.commit()

Обновление нескольких записей

# Увеличиваем возраст всех пользователей старше 30 лет на один год
cursor.execute('''
    UPDATE users 
    SET age = age + 1 
    WHERE age > 30
''')

conn.commit()

# Обновление email для всех пользователей с определённым доменом
cursor.execute('''
    UPDATE users 
    SET email = REPLACE(email, '@example.com', '@newdomain.com')
    WHERE email LIKE '%@example.com'
''')

conn.commit()

Условное обновление с проверкой

# Обновление возраста только если новое значение больше текущего
cursor.execute('''
    UPDATE users 
    SET age = ? 
    WHERE name = ? AND age < ?
''', (35, 'Bob', 35))

conn.commit()

# Проверка количества затронутых записей
affected_rows = cursor.rowcount
print(f"Обновлено записей: {affected_rows}")

Удаление данных из таблицы

Удаление отдельных записей

cursor.execute('DELETE FROM users WHERE name = ?', ('Bob',))
conn.commit()

# Удаление по нескольким условиям
cursor.execute('DELETE FROM users WHERE age < ? AND email LIKE ?', (25, '%@example.com'))
conn.commit()

Очистка таблицы

# Удаление всех записей из таблицы
cursor.execute('DELETE FROM users')
conn.commit()

# Сброс автоинкремента
cursor.execute('DELETE FROM sqlite_sequence WHERE name = ?', ('users',))
conn.commit()

Управление соединениями и ресурсами

Правильное закрытие соединения

cursor.close()
conn.close()

Использование контекстного менеджера

Контекстный менеджер обеспечивает автоматическое закрытие соединения даже при возникновении ошибок:

with sqlite3.connect('example.db') as conn:
    cursor = conn.cursor()
    cursor.execute('SELECT * FROM users')
    results = cursor.fetchall()
    
    for row in results:
        print(row)

Создание функции для подключения

def get_db_connection():
    """Создаёт и возвращает подключение к базе данных"""
    conn = sqlite3.connect('example.db')
    conn.row_factory = sqlite3.Row  # Позволяет обращаться к колонкам по имени
    return conn

def get_user_by_id(user_id):
    """Получает пользователя по ID"""
    with get_db_connection() as conn:
        cursor = conn.cursor()
        cursor.execute('SELECT * FROM users WHERE id = ?', (user_id,))
        return cursor.fetchone()

Работа с транзакциями

Основы транзакций

Транзакции обеспечивают целостность данных при выполнении нескольких связанных операций:

try:
    conn.execute('BEGIN TRANSACTION')
    
    cursor.execute('INSERT INTO users (name, age, email) VALUES (?, ?, ?)', 
                  ('John', 25, 'john@example.com'))
    cursor.execute('UPDATE users SET age = age + 1 WHERE name = ?', ('Alice',))
    
    conn.commit()
    print("Транзакция выполнена успешно")
    
except sqlite3.Error as e:
    conn.rollback()
    print(f"Ошибка транзакции: {e}")

Автоматическое управление транзакциями

def transfer_data(from_id, to_id, amount):
    """Пример передачи данных между записями"""
    try:
        with sqlite3.connect('example.db') as conn:
            cursor = conn.cursor()
            
            # Проверяем существование записей
            cursor.execute('SELECT age FROM users WHERE id = ?', (from_id,))
            source = cursor.fetchone()
            
            if not source or source[0] < amount:
                raise ValueError("Недостаточно данных для передачи")
            
            # Выполняем операции
            cursor.execute('UPDATE users SET age = age - ? WHERE id = ?', (amount, from_id))
            cursor.execute('UPDATE users SET age = age + ? WHERE id = ?', (amount, to_id))
            
            conn.commit()
            
    except sqlite3.Error as e:
        print(f"Ошибка базы данных: {e}")
    except ValueError as e:
        print(f"Ошибка валидации: {e}")

Безопасность и предотвращение SQL-инъекций

Использование параметризованных запросов

Всегда используйте параметризованные запросы для предотвращения SQL-инъекций:

# Правильный способ
user_input = 'Alice'
cursor.execute('SELECT * FROM users WHERE name = ?', (user_input,))

# Неправильный способ - уязвимость к SQL-инъекциям
# cursor.execute(f'SELECT * FROM users WHERE name = "{user_input}"')

Валидация входных данных

def safe_user_search(name):
    """Безопасный поиск пользователя по имени"""
    if not isinstance(name, str) or len(name.strip()) == 0:
        return None
    
    # Ограничиваем длину для предотвращения DoS-атак
    if len(name) > 50:
        name = name[:50]
    
    with sqlite3.connect('example.db') as conn:
        cursor = conn.cursor()
        cursor.execute('SELECT * FROM users WHERE name = ?', (name.strip(),))
        return cursor.fetchall()

Работа с датами и временем

Сохранение временных меток

import datetime

cursor.execute('''
    CREATE TABLE IF NOT EXISTS logs (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        event TEXT NOT NULL,
        created_at TEXT NOT NULL,
        updated_at TEXT
    )
''')

# Добавление записи с текущей датой
now = datetime.datetime.now()
cursor.execute('''
    INSERT INTO logs (event, created_at) 
    VALUES (?, ?)
''', ('User Login', now.isoformat()))

conn.commit()

Работа с датами в запросах

# Поиск записей за последние 7 дней
week_ago = (datetime.datetime.now() - datetime.timedelta(days=7)).isoformat()
cursor.execute('''
    SELECT * FROM logs 
    WHERE created_at > ? 
    ORDER BY created_at DESC
''', (week_ago,))

recent_logs = cursor.fetchall()

Форматирование дат

def format_date_for_display(iso_date):
    """Преобразует ISO дату в читаемый формат"""
    try:
        dt = datetime.datetime.fromisoformat(iso_date)
        return dt.strftime("%d.%m.%Y %H:%M")
    except ValueError:
        return "Неверная дата"

# Получение и форматирование дат
cursor.execute('SELECT event, created_at FROM logs')
logs = cursor.fetchall()

for event, created_at in logs:
    formatted_date = format_date_for_display(created_at)
    print(f"{event} - {formatted_date}")

Полноценное приложение-пример

Класс для работы с задачами

import sqlite3
import datetime
from typing import List, Optional

class TaskManager:
    def __init__(self, db_path: str = 'tasks.db'):
        self.db_path = db_path
        self.init_database()
    
    def init_database(self):
        """Инициализация базы данных"""
        with sqlite3.connect(self.db_path) as conn:
            conn.execute('''
                CREATE TABLE IF NOT EXISTS tasks (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    title TEXT NOT NULL,
                    description TEXT,
                    completed BOOLEAN DEFAULT 0,
                    priority INTEGER DEFAULT 1,
                    created_at TEXT NOT NULL,
                    updated_at TEXT
                )
            ''')
            
            conn.execute('''
                CREATE INDEX IF NOT EXISTS idx_tasks_completed 
                ON tasks(completed)
            ''')
            
            conn.execute('''
                CREATE INDEX IF NOT EXISTS idx_tasks_priority 
                ON tasks(priority)
            ''')
    
    def add_task(self, title: str, description: str = "", priority: int = 1) -> int:
        """Добавление новой задачи"""
        now = datetime.datetime.now().isoformat()
        
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute('''
                INSERT INTO tasks (title, description, priority, created_at) 
                VALUES (?, ?, ?, ?)
            ''', (title, description, priority, now))
            
            return cursor.lastrowid
    
    def get_all_tasks(self) -> List[sqlite3.Row]:
        """Получение всех задач"""
        with sqlite3.connect(self.db_path) as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.cursor()
            cursor.execute('''
                SELECT * FROM tasks 
                ORDER BY priority DESC, created_at ASC
            ''')
            return cursor.fetchall()
    
    def get_pending_tasks(self) -> List[sqlite3.Row]:
        """Получение невыполненных задач"""
        with sqlite3.connect(self.db_path) as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.cursor()
            cursor.execute('''
                SELECT * FROM tasks 
                WHERE completed = 0 
                ORDER BY priority DESC, created_at ASC
            ''')
            return cursor.fetchall()
    
    def complete_task(self, task_id: int) -> bool:
        """Отметка задачи как выполненной"""
        now = datetime.datetime.now().isoformat()
        
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute('''
                UPDATE tasks 
                SET completed = 1, updated_at = ? 
                WHERE id = ?
            ''', (now, task_id))
            
            return cursor.rowcount > 0
    
    def update_task(self, task_id: int, title: str = None, 
                   description: str = None, priority: int = None) -> bool:
        """Обновление информации о задаче"""
        updates = []
        values = []
        
        if title is not None:
            updates.append("title = ?")
            values.append(title)
        
        if description is not None:
            updates.append("description = ?")
            values.append(description)
        
        if priority is not None:
            updates.append("priority = ?")
            values.append(priority)
        
        if not updates:
            return False
        
        updates.append("updated_at = ?")
        values.append(datetime.datetime.now().isoformat())
        values.append(task_id)
        
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            query = f"UPDATE tasks SET {', '.join(updates)} WHERE id = ?"
            cursor.execute(query, values)
            
            return cursor.rowcount > 0
    
    def delete_task(self, task_id: int) -> bool:
        """Удаление задачи"""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute('DELETE FROM tasks WHERE id = ?', (task_id,))
            return cursor.rowcount > 0
    
    def get_task_statistics(self) -> dict:
        """Получение статистики по задачам"""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute('''
                SELECT 
                    COUNT(*) as total,
                    SUM(completed) as completed,
                    COUNT(*) - SUM(completed) as pending,
                    AVG(priority) as avg_priority
                FROM tasks
            ''')
            
            row = cursor.fetchone()
            return {
                'total': row[0],
                'completed': row[1] or 0,
                'pending': row[2] or 0,
                'avg_priority': round(row[3] or 0, 2)
            }

Использование менеджера задач

# Создание экземпляра менеджера
task_manager = TaskManager()

# Добавление задач
task1_id = task_manager.add_task("Изучить SQLite", "Прочитать документацию", 3)
task2_id = task_manager.add_task("Написать код", "Создать пример приложения", 2)
task3_id = task_manager.add_task("Тестирование", "Проверить функциональность", 1)

# Получение и отображение задач
print("Все задачи:")
for task in task_manager.get_all_tasks():
    status = "✓" if task['completed'] else "○"
    print(f"{status} {task['title']} (Приоритет: {task['priority']})")

# Выполнение задачи
task_manager.complete_task(task1_id)

# Обновление задачи
task_manager.update_task(task2_id, description="Создать полноценное приложение с GUI")

# Статистика
stats = task_manager.get_task_statistics()
print(f"\nСтатистика:")
print(f"Всего задач: {stats['total']}")
print(f"Выполнено: {stats['completed']}")
print(f"Осталось: {stats['pending']}")
print(f"Средний приоритет: {stats['avg_priority']}")

Оптимизация производительности

Создание индексов для ускорения запросов

# Создание составного индекса
cursor.execute('''
    CREATE INDEX IF NOT EXISTS idx_users_age_email 
    ON users(age, email)
''')

# Индекс для поиска по подстроке
cursor.execute('''
    CREATE INDEX IF NOT EXISTS idx_users_name_text 
    ON users(name COLLATE NOCASE)
''')

conn.commit()

Использование EXPLAIN для анализа запросов

cursor.execute('EXPLAIN QUERY PLAN SELECT * FROM users WHERE age > 25')
plan = cursor.fetchall()
for step in plan:
    print(step)

Настройка SQLite для повышения производительности

def optimize_connection(conn):
    """Применяет настройки оптимизации для соединения"""
    cursor = conn.cursor()
    
    # Увеличиваем размер страницы для лучшей производительности
    cursor.execute('PRAGMA page_size = 4096')
    
    # Включаем WAL режим для лучшей параллельности
    cursor.execute('PRAGMA journal_mode = WAL')
    
    # Увеличиваем размер кэша
    cursor.execute('PRAGMA cache_size = 10000')
    
    # Синхронизируем данные реже (осторожно в критических приложениях)
    cursor.execute('PRAGMA synchronous = NORMAL')
    
    # Включаем поддержку внешних ключей
    cursor.execute('PRAGMA foreign_keys = ON')
    
    conn.commit()

# Применение оптимизации
with sqlite3.connect('example.db') as conn:
    optimize_connection(conn)

Частые вопросы и решения

Проверка структуры таблицы

def get_table_info(table_name):
    """Получает информацию о структуре таблицы"""
    with sqlite3.connect('example.db') as conn:
        cursor = conn.cursor()
        cursor.execute(f'PRAGMA table_info({table_name})')
        columns = cursor.fetchall()
        
        print(f"Структура таблицы {table_name}:")
        for column in columns:
            print(f"  {column[1]} {column[2]} {'NOT NULL' if column[3] else ''}")

get_table_info('users')

Получение списка всех таблиц

def list_tables():
    """Получает список всех таблиц в базе данных"""
    with sqlite3.connect('example.db') as conn:
        cursor = conn.cursor()
        cursor.execute('''
            SELECT name FROM sqlite_master 
            WHERE type='table' AND name NOT LIKE 'sqlite_%'
        ''')
        
        tables = cursor.fetchall()
        return [table[0] for table in tables]

tables = list_tables()
print("Таблицы в базе данных:", tables)

Резервное копирование базы данных

import shutil
import os

def backup_database(source_db, backup_path):
    """Создаёт резервную копию базы данных"""
    try:
        # Создаём директорию для бэкапов если её нет
        os.makedirs(os.path.dirname(backup_path), exist_ok=True)
        
        # Копируем файл базы данных
        shutil.copy2(source_db, backup_path)
        print(f"Резервная копия создана: {backup_path}")
        
    except Exception as e:
        print(f"Ошибка создания резервной копии: {e}")

# Создание резервной копии
backup_database('example.db', 'backups/example_backup.db')

Миграция схемы базы данных

def migrate_database():
    """Выполняет миграцию схемы базы данных"""
    with sqlite3.connect('example.db') as conn:
        cursor = conn.cursor()
        
        # Проверяем текущую версию схемы
        try:
            cursor.execute('SELECT version FROM schema_version')
            current_version = cursor.fetchone()[0]
        except sqlite3.OperationalError:
            # Таблица не существует, создаём её
            cursor.execute('''
                CREATE TABLE schema_version (
                    version INTEGER PRIMARY KEY
                )
            ''')
            cursor.execute('INSERT INTO schema_version (version) VALUES (0)')
            current_version = 0
            conn.commit()
        
        # Выполняем миграции
        migrations = {
            1: [
                'ALTER TABLE users ADD COLUMN phone TEXT',
                'CREATE INDEX idx_users_phone ON users(phone)'
            ],
            2: [
                'ALTER TABLE users ADD COLUMN created_at TEXT DEFAULT ""',
                'UPDATE users SET created_at = datetime("now") WHERE created_at = ""'
            ]
        }
        
        for version, queries in migrations.items():
            if current_version < version:
                for query in queries:
                    cursor.execute(query)
                
                cursor.execute('UPDATE schema_version SET version = ?', (version,))
                print(f"Применена миграция версии {version}")
        
        conn.commit()

migrate_database()

Заключение

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

SQLite особенно хорошо подходит для разработки прототипов, небольших и средних приложений, мобильных решений и ситуаций, когда не требуется сложная серверная инфраструктура. Простота использования SQLite делает её идеальным выбором для изучения основ работы с базами данных.

Приобретённые навыки работы с SQLite станут отличной основой для дальнейшего изучения более сложных систем управления базами данных, таких как PostgreSQL, MySQL или MongoDB. Понимание принципов работы с реляционными базами данных и SQL-запросами поможет в профессиональном развитии любого разработчика Python.