Практическое занятие 7: Базы данных SQLite

Цель занятия

Научиться работать с базой данных SQLite: создавать таблицы, вставлять и извлекать данные.

Что такое SQLite?

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

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

import sqlite3

# Подключение к базе данных (создастся файл, если не существует)
connection = sqlite3.connect("my_database.db")

# Создание объекта курсора
cursor = connection.cursor()

# После работы закрываем соединение
connection.close()

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

import sqlite3

connection = sqlite3.connect("my_database.db")
cursor = connection.cursor()

# SQL-запрос для создания таблицы
create_table_query = """
CREATE TABLE IF NOT EXISTS students (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    age INTEGER,
    course INTEGER,
    grade REAL
)
"""

cursor.execute(create_table_query)
connection.commit()
connection.close()

print("Таблица students создана")

Основные типы данных в SQLite

Тип данных	Описание
INTEGER	Целое число
REAL	Число с плавающей точкой
TEXT	Текстовая строка
BLOB	Бинарные данные
NULL	Отсутствие значения

Вставка данных (INSERT)

Вставка одной записи

import sqlite3

connection = sqlite3.connect("my_database.db")
cursor = connection.cursor()

# Вставка одной записи
insert_query = "INSERT INTO students (name, age, course, grade) VALUES (?, ?, ?, ?)"
cursor.execute(insert_query, ("Анна", 20, 2, 4.5))

connection.commit()
connection.close()

print("Запись добавлена")

Вставка нескольких записей

import sqlite3

connection = sqlite3.connect("my_database.db")
cursor = connection.cursor()

# Список данных для вставки
students_data = [
    ("Борис", 21, 3, 4.2),
    ("Вера", 19, 1, 4.8),
    ("Георгий", 22, 4, 3.9),
    ("Дарья", 20, 2, 4.6)
]

# Вставка нескольких записей
insert_query = "INSERT INTO students (name, age, course, grade) VALUES (?, ?, ?, ?)"
cursor.executemany(insert_query, students_data)

connection.commit()
connection.close()

print(f"Добавлено {len(students_data)} записей")

Извлечение данных (SELECT)

Извлечение всех записей

import sqlite3

connection = sqlite3.connect("my_database.db")
cursor = connection.cursor()

# Извлечение всех записей
select_query = "SELECT * FROM students"
cursor.execute(select_query)

# Получение всех записей
students = cursor.fetchall()

print("Все студенты:")
for student in students:
    print(f"ID: {student[0]}, Имя: {student[1]}, Возраст: {student[2]}, Курс: {student[3]}, Оценка: {student[4]}")

connection.close()

Извлечение с условием (WHERE)

import sqlite3

connection = sqlite3.connect("my_database.db")
cursor = connection.cursor()

# Извлечение студентов 2 курса
select_query = "SELECT * FROM students WHERE course = ?"
cursor.execute(select_query, (2,))

students = cursor.fetchall()

print("Студенты 2 курса:")
for student in students:
    print(f"{student[1]}, Оценка: {student[4]}")

connection.close()

Извлечение определённых столбцов

import sqlite3

connection = sqlite3.connect("my_database.db")
cursor = connection.cursor()

# Извлечение только имени и оценки
select_query = "SELECT name, grade FROM students"
cursor.execute(select_query)

students = cursor.fetchall()

print("Имена и оценки:")
for student in students:
    print(f"{student[0]}: {student[1]}")

connection.close()

Сортировка (ORDER BY)

import sqlite3

connection = sqlite3.connect("my_database.db")
cursor = connection.cursor()

# Сортировка по оценке по убыванию
select_query = "SELECT name, grade FROM students ORDER BY grade DESC"
cursor.execute(select_query)

students = cursor.fetchall()

print("Студенты, отсортированные по оценке:")
for student in students:
    print(f"{student[0]}: {student[1]}")

connection.close()

Ограничение количества записей (LIMIT)

import sqlite3

connection = sqlite3.connect("my_database.db")
cursor = connection.cursor()

# Получение 3 лучших студентов
select_query = "SELECT name, grade FROM students ORDER BY grade DESC LIMIT 3"
cursor.execute(select_query)

students = cursor.fetchall()

print("Топ-3 студента:")
for student in students:
    print(f"{student[0]}: {student[1]}")

connection.close()

Обновление данных (UPDATE)

import sqlite3

connection = sqlite3.connect("my_database.db")
cursor = connection.cursor()

# Обновление оценки для студента с ID = 1
update_query = "UPDATE students SET grade = ? WHERE id = ?"
cursor.execute(update_query, (5.0, 1))

connection.commit()
connection.close()

print("Оценка обновлена")

Удаление данных (DELETE)

import sqlite3

connection = sqlite3.connect("my_database.db")
cursor = connection.cursor()

# Удаление студента с ID = 5
delete_query = "DELETE FROM students WHERE id = ?"
cursor.execute(delete_query, (5,))

connection.commit()
connection.close()

print("Студент удалён")

Работа с несколькими таблицами

import sqlite3

connection = sqlite3.connect("university.db")
cursor = connection.cursor()

# Создание таблицы курсов
create_courses_query = """
CREATE TABLE IF NOT EXISTS courses (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    teacher TEXT,
    hours INTEGER
)
"""

cursor.execute(create_courses_query)

# Создание таблицы зачислений (связующая таблица)
create_enrollments_query = """
CREATE TABLE IF NOT EXISTS enrollments (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    student_id INTEGER,
    course_id INTEGER,
    FOREIGN KEY (student_id) REFERENCES students(id),
    FOREIGN KEY (course_id) REFERENCES courses(id)
)
"""

cursor.execute(create_enrollments_query)

connection.commit()
connection.close()

print("Таблицы созданы")

Функции для работы с базой данных

Создайте файл database_utils.py:

import sqlite3

class DatabaseManager:
    def __init__(self, db_name):
        self.db_name = db_name
        self.connection = None
        self.cursor = None

    def connect(self):
        """Подключение к базе данных"""
        self.connection = sqlite3.connect(self.db_name)
        self.cursor = self.connection.cursor()

    def close(self):
        """Закрытие соединения"""
        if self.connection:
            self.connection.close()

    def execute_query(self, query, params=None):
        """Выполнение SQL-запроса без возвращения данных"""
        self.cursor.execute(query, params or ())
        self.connection.commit()

    def fetch_all(self, query, params=None):
        """Выполнение запроса и получение всех записей"""
        self.cursor.execute(query, params or ())
        return self.cursor.fetchall()

    def fetch_one(self, query, params=None):
        """Выполнение запроса и получение одной записи"""
        self.cursor.execute(query, params or ())
        return self.cursor.fetchone()

# Использование
if __name__ == "__main__":
    db = DatabaseManager("test.db")
    db.connect()

    # Создание таблицы
    db.execute_query("""
        CREATE TABLE IF NOT EXISTS products (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            price REAL
        )
    """)

    # Вставка данных
    db.execute_query(
        "INSERT INTO products (name, price) VALUES (?, ?)",
        ("Яблоко", 99.90)
    )

    # Извлечение данных
    products = db.fetch_all("SELECT * FROM products")
    print(products)

    db.close()

Практические примеры

Полный пример: система учёта студентов

import sqlite3

class StudentDatabase:
    def __init__(self, db_name="students.db"):
        self.connection = sqlite3.connect(db_name)
        self.cursor = self.connection.cursor()
        self.create_tables()

    def create_tables(self):
        """Создание таблиц"""
        self.cursor.execute("""
            CREATE TABLE IF NOT EXISTS students (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT NOT NULL,
                email TEXT UNIQUE,
                course INTEGER
            )
        """)
        self.connection.commit()

    def add_student(self, name, email, course):
        """Добавление студента"""
        try:
            self.cursor.execute(
                "INSERT INTO students (name, email, course) VALUES (?, ?, ?)",
                (name, email, course)
            )
            self.connection.commit()
            print(f"Студент {name} добавлен")
            return self.cursor.lastrowid
        except sqlite3.IntegrityError:
            print(f"Студент с email {email} уже существует")
            return None

    def get_student(self, student_id):
        """Получение студента по ID"""
        self.cursor.execute(
            "SELECT * FROM students WHERE id = ?",
            (student_id,)
        )
        return self.cursor.fetchone()

    def get_all_students(self):
        """Получение всех студентов"""
        self.cursor.execute("SELECT * FROM students")
        return self.cursor.fetchall()

    def update_student(self, student_id, name=None, email=None, course=None):
        """Обновление данных студента"""
        updates = []
        params = []

        if name:
            updates.append("name = ?")
            params.append(name)
        if email:
            updates.append("email = ?")
            params.append(email)
        if course:
            updates.append("course = ?")
            params.append(course)

        if updates:
            query = f"UPDATE students SET {', '.join(updates)} WHERE id = ?"
            params.append(student_id)
            self.cursor.execute(query, params)
            self.connection.commit()
            print(f"Данные студента {student_id} обновлены")

    def delete_student(self, student_id):
        """Удаление студента"""
        self.cursor.execute(
            "DELETE FROM students WHERE id = ?",
            (student_id,)
        )
        self.connection.commit()
        print(f"Студент {student_id} удалён")

    def close(self):
        """Закрытие соединения"""
        self.connection.close()

# Использование
if __name__ == "__main__":
    db = StudentDatabase()

    # Добавление студентов
    db.add_student("Анна Иванова", "anna@example.com", 2)
    db.add_student("Борис Петров", "boris@example.com", 3)

    # Получение всех студентов
    students = db.get_all_students()
    print("\nВсе студенты:")
    for student in students:
        print(f"ID: {student[0]}, Имя: {student[1]}, Email: {student[2]}, Курс: {student[3]}")

    # Обновление студента
    db.update_student(1, course=3)

    # Удаление студента
    db.delete_student(2)

    db.close()

Задания для самостоятельной работы

1. Создайте базу данных для хранения информации о книгах (название, автор, год, жанр). Реализуйте функции для добавления, поиска и удаления книг.

2. Напишите функцию для экспорта данных из базы данных в JSON файл.

3. Создайте базу данных с двумя связанными таблицами (например, авторы и книги) и реализуйте запросы с JOIN.

4. Реализуйте поиск по частичному совпадению (например, поиск книг, в названии которых содержится определённое слово).

Пример решения задания 2

import sqlite3
import json

def export_to_json(db_name, table_name, output_file):
    """Экспорт таблицы в JSON"""
    connection = sqlite3.connect(db_name)
    cursor = connection.cursor()

    # Получение данных
    cursor.execute(f"SELECT * FROM {table_name}")
    rows = cursor.fetchall()

    # Получение названий столбцов
    cursor.execute(f"PRAGMA table_info({table_name})")
    columns = [column[1] for column in cursor.fetchall()]

    # Формирование списка словарей
    data = []
    for row in rows:
        data.append(dict(zip(columns, row)))

    # Сохранение в JSON
    with open(output_file, "w", encoding="utf-8") as file:
        json.dump(data, file, ensure_ascii=False, indent=4)

    connection.close()
    print(f"Экспортировано {len(data)} записей в {output_file}")

# Использование
export_to_json("students.db", "students", "students.json")

Полезные ресурсы

• Документация SQLite
• SQL Tutorial
• SQLite Tutorial

Следующий шаг

На следующем занятии мы изучим объектно-ориентированное программирование и создадим модуль с базовым классом скрапера.