Sqlite3.connect: примеры (PYTHON)

Примеры использования sqlite3.connect

Базовое подключение к файлу базы данных

import sqlite3

# Подключение к существующей или создание новой базы данных
conn = sqlite3.connect('example.db')
print(type(conn))
conn.close()

Создание базы данных в оперативной памяти

import sqlite3

# Использование базы данных в оперативной памяти
conn = sqlite3.connect(':memory:')
cursor = conn.cursor()
cursor.execute("CREATE TABLE test (id INTEGER, name TEXT)")
cursor.execute("INSERT INTO test VALUES (1, 'Test')")
conn.commit()
conn.close()

# База данных создана и удалена после закрытия соединения

Подключение с различными параметрами

import sqlite3

# Подключение с указанием нескольких параметров
conn = sqlite3.connect(
    database='test.db',
    timeout=10.0,
    detect_types=sqlite3.PARSE_DECLTYPES,
    isolation_level='EXCLUSIVE',
    check_same_thread=False,
    cached_statements=200
)
print(f"Isolation level: {conn.isolation_level}")
conn.close()

Isolation level: EXCLUSIVE

Использование URI для подключения

import sqlite3

# Подключение с использованием URI
conn = sqlite3.connect('file:test.db?mode=rw', uri=True)
print(f"Connection established: {conn}")
conn.close()

Connection established: 

Похожие функции в Python

Модуль sqlite3

В рамках модуля sqlite3 функция connect является основным методом создания соединений. Альтернативой может быть использование классов-наследников sqlite3.Connection через параметр factory.

ORM-библиотеки

Для работы с SQLite в Python также используются ORM-библиотеки:

• SQLAlchemy - создает соединения через create_engine() с поддержкой пулинга и продвинутых возможностей.
• Peewee - использует класс SqliteDatabase для инициализации соединения с базой данных.
• Django ORM - настраивает соединение через параметры DATABASES в настройках проекта.

ORM-библиотеки предпочтительны в крупных проектах с объектно-ориентированной архитектурой, тогда как sqlite3.connect подходит для простых сценариев и скриптов.

Альтернативы в других языках программирования

JavaScript (Node.js)

const sqlite3 = require('sqlite3').verbose();
const db = new sqlite3.Database('./test.db', (err) => {
    if (err) console.error(err.message);
    console.log('Connected to SQLite database');
});
db.close();

Connected to SQLite database

Java

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;

public class SQLiteExample {
    public static void main(String[] args) {
        Connection conn = null;
        try {
            conn = DriverManager.getConnection("jdbc:sqlite:test.db");
            System.out.println("Connection established");
        } catch (SQLException e) {
            System.out.println(e.getMessage());
        } finally {
            try {
                if (conn != null) conn.close();
            } catch (SQLException e) {
                System.out.println(e.getMessage());
            }
        }
    }
}

Connection established

PHP

<?
$db = new SQLite3('test.db');
echo 'Connected to SQLite database';
$db->close();
?>

Connected to SQLite database

C#

using System.Data.SQLite;

class Program
{
    static void Main()
    {
        SQLiteConnection conn = new SQLiteConnection("Data Source=test.db");
        conn.Open();
        Console.WriteLine("Connection established");
        conn.Close();
    }
}

Connection established

Golang

package main

import (
    "database/sql"
    _ "github.com/mattn/go-sqlite3"
    "log"
)

func main() {
    db, err := sql.Open("sqlite3", "./test.db")
    if err != nil {
        log.Fatal(err)
    }
    defer db.Close()
    log.Println("Connected to SQLite database")
}

2023/10/15 12:00:00 Connected to SQLite database

Основные отличия от реализации в Python включают использование драйверов, интерфейсов и методов управления соединениями, характерных для каждого языка.

Типичные ошибки при использовании sqlite3.connect

Проблемы с доступом к файлу базы данных

import sqlite3

try:
    # Попытка подключения к файлу в защищенной директории
    conn = sqlite3.connect('/root/database.db')
except sqlite3.OperationalError as e:
    print(f"Ошибка: {e}")

Ошибка: unable to open database file

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

import sqlite3
import threading

def query_db():
    cursor = conn.cursor()
    cursor.execute("SELECT 1")

conn = sqlite3.connect('test.db', check_same_thread=True)
# Попытка использования соединения из другого потока
thread = threading.Thread(target=query_db)
thread.start()
thread.join()

sqlite3.ProgrammingError: SQLite objects created in a thread can only be used in that same thread.

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

import sqlite3

conn = sqlite3.connect(':memory:')
conn.close()
# Попытка выполнить запрос после закрытия соединения
try:
    cursor = conn.cursor()
except sqlite3.ProgrammingError as e:
    print(f"Ошибка: {e}")

Ошибка: Cannot operate on a closed database.

Некорректные параметры в URI

import sqlite3

try:
    conn = sqlite3.connect('file:test.db?mode=invalid', uri=True)
except sqlite3.OperationalError as e:
    print(f"Ошибка: {e}")

Ошибка: unable to open database file

Изменения в последних версиях Python

В Python 3.10 добавлена поддержка контекстного менеджера для автоматического закрытия соединения. В Python 3.12 улучшена производительность обработки параметризованных запросов и добавлены новые флаги для настройки поведения соединения.

В версии Python 3.11 улучшена обработка ошибок в транзакциях и добавлена поддержка новых режимов изоляции. Также была оптимизирована работа с большими объемами данных при использовании detect_types.

Начиная с Python 3.9, параметр timeout может принимать значения с плавающей точкой для более точного контроля времени ожидания.

Расширенные примеры использования sqlite3.connect

Создание пользовательского класса соединения

import sqlite3

class CustomConnection(sqlite3.Connection):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.execute("PRAGMA foreign_keys = ON")
        self.execute("PRAGMA journal_mode = WAL")
    
    def custom_method(self):
        return "Custom connection method"

# Использование пользовательского класса соединения
conn = sqlite3.connect(':memory:', factory=CustomConnection)
print(conn.custom_method())
cursor = conn.cursor()
cursor.execute("PRAGMA foreign_keys")
print(f"Foreign keys: {cursor.fetchone()[0]}")
conn.close()

Custom connection method
Foreign keys: 1

Работа с различными режимами изоляции

import sqlite3

# Сравнение различных уровней изоляции
isolation_levels = [None, 'DEFERRED', 'IMMEDIATE', 'EXCLUSIVE']

for level in isolation_levels:
    conn = sqlite3.connect(':memory:', isolation_level=level)
    cursor = conn.cursor()
    cursor.execute("CREATE TABLE test (id INTEGER PRIMARY KEY, data TEXT)")
    
    # Вставка данных с различными уровнями изоляции
    conn.execute("INSERT INTO test (data) VALUES ('test')")
    
    print(f"Isolation level {level}: autocommit mode - {conn.isolation_level is None}")
    conn.close()

Isolation level None: autocommit mode - True
Isolation level DEFERRED: autocommit mode - False
Isolation level IMMEDIATE: autocommit mode - False
Isolation level EXCLUSIVE: autocommit mode - False

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

import sqlite3
import datetime
import pickle

# Регистрация адаптеров для пользовательских типов
sqlite3.register_adapter(datetime.datetime, lambda dt: dt.isoformat())
sqlite3.register_converter("datetime", lambda b: datetime.datetime.fromisoformat(b.decode()))

# Регистрация адаптера для сложных объектов
class ComplexObject:
    def __init__(self, value):
        self.value = value

sqlite3.register_adapter(ComplexObject, lambda obj: pickle.dumps(obj))
sqlite3.register_converter("COMPLEX", lambda b: pickle.loads(b))

# Подключение с использованием detect_types
conn = sqlite3.connect(':memory:', detect_types=sqlite3.PARSE_DECLTYPES)
cursor = conn.cursor()

# Создание таблицы с пользовательскими типами
cursor.execute("""
    CREATE TABLE test (
        id INTEGER PRIMARY KEY,
        created_at datetime,
        complex_data COMPLEX
    )""")

# Вставка данных с пользовательскими типами
now = datetime.datetime.now()
obj = ComplexObject([1, 2, 3])
cursor.execute(
    "INSERT INTO test (created_at, complex_data) VALUES (?, ?)",
    (now, obj)
)
conn.commit()

# Чтение данных
cursor.execute("SELECT created_at, complex_data FROM test WHERE id = 1")
row = cursor.fetchone()
print(f"Created at: {row[0]}, type: {type(row[0])}")
print(f"Complex data: {row[1].value}, type: {type(row[1])}")

conn.close()

Created at: 2023-10-15 12:00:00, type: 
Complex data: [1, 2, 3], type: 

Настройка кэширования подготовленных запросов

import sqlite3
import time

# Сравнение производительности с разными размерами кэша
for cache_size in [0, 10, 100, 1000]:
    conn = sqlite3.connect(':memory:', cached_statements=cache_size)
    cursor = conn.cursor()
    cursor.execute("CREATE TABLE test (id INTEGER, value TEXT)")
    
    start_time = time.time()
    for i in range(1000):
        cursor.execute("INSERT INTO test VALUES (?, ?)", (i, f'value_{i}'))
    
    conn.commit()
    end_time = time.time()
    
    print(f"Cache size {cache_size}: {end_time - start_time:.4f} seconds")
    conn.close()

Cache size 0: 0.0451 seconds
Cache size 10: 0.0324 seconds
Cache size 100: 0.0318 seconds
Cache size 1000: 0.0315 seconds

Многопоточное приложение с разделением соединений

import sqlite3
import threading
from queue import Queue

# Создание общей базы данных
main_conn = sqlite3.connect('shared.db', check_same_thread=False)
main_conn.execute("CREATE TABLE IF NOT EXISTS counters (thread_id INTEGER, count INTEGER)")
main_conn.commit()
main_conn.close()

def worker(thread_id, results_queue):
    # Каждый поток создает свое соединение
    conn = sqlite3.connect('shared.db', check_same_thread=False)
    cursor = conn.cursor()
    
    for i in range(10):
        cursor.execute(
            "INSERT INTO counters (thread_id, count) VALUES (?, ?)",
            (thread_id, i)
        )
    
    conn.commit()
    
    # Получение результатов для этого потока
    cursor.execute(
        "SELECT COUNT(*) FROM counters WHERE thread_id = ?",
        (thread_id,)
    )
    count = cursor.fetchone()[0]
    results_queue.put((thread_id, count))
    
    conn.close()

# Запуск нескольких потоков
results_queue = Queue()
threads = []

for i in range(5):
    thread = threading.Thread(target=worker, args=(i, results_queue))
    threads.append(thread)
    thread.start()

# Ожидание завершения всех потоков
for thread in threads:
    thread.join()

# Вывод результатов
print("Results:")
while not results_queue.empty():
    thread_id, count = results_queue.get()
    print(f"Thread {thread_id}: {count} records")

Results:
Thread 0: 10 records
Thread 1: 10 records
Thread 2: 10 records
Thread 3: 10 records
Thread 4: 10 records