Home » Как использовать SQLite в приложении Flask
Как использовать SQLite в приложении Flask

Как использовать SQLite в приложении Flask

Если задачи вашего проекта решаются локально или не требуют сложных распределённых систем, то SQLite — идеальный выбор для small-to-medium проектов на Flask. Эта статья покажет, как быстро и эффективно интегрировать SQLite в ваше Flask-приложение, настроить базу данных и избежать типичных подводных камней. Разберём всё от создания схемы до production-ready решений.

## Как это работает?

SQLite — это встроенная СУБД, которая не требует отдельного сервера. Весь механизм работает через один файл .db, что упрощает развёртывание и обслуживание. Flask взаимодействует с SQLite через Python-модуль sqlite3 (встроенный в стандартную библиотеку) или через ORM-решения типа SQLAlchemy.

Принцип работы:

  • Flask-приложение создаёт соединение с базой данных
  • Каждый запрос может открывать новое соединение или использовать пул соединений
  • SQLite блокирует базу при записи (WAL-режим частично решает проблему)
  • Данные хранятся в одном файле, что упрощает backup и миграции

## Быстрая настройка: от нуля до результата

Начнём с минимального рабочего примера:


import sqlite3
from flask import Flask, request, jsonify, g
import os

app = Flask(__name__)
DATABASE = 'database.db'

def get_db():
if 'db' not in g:
g.db = sqlite3.connect(DATABASE)
g.db.row_factory = sqlite3.Row
return g.db

def close_db(e=None):
db = g.pop('db', None)
if db is not None:
db.close()

@app.teardown_appcontext
def close_db(error):
close_db()

def init_db():
with app.app_context():
db = get_db()
with app.open_resource('schema.sql', mode='r') as f:
db.cursor().executescript(f.read())
db.commit()

@app.route('/users', methods=['GET'])
def get_users():
db = get_db()
users = db.execute('SELECT * FROM users').fetchall()
return jsonify([dict(row) for row in users])

@app.route('/users', methods=['POST'])
def create_user():
db = get_db()
data = request.get_json()
db.execute('INSERT INTO users (name, email) VALUES (?, ?)',
(data['name'], data['email']))
db.commit()
return jsonify({'status': 'created'})

if __name__ == '__main__':
if not os.path.exists(DATABASE):
init_db()
app.run(debug=True)

Создайте файл schema.sql:


CREATE TABLE users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
email TEXT UNIQUE NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX idx_users_email ON users(email);
CREATE INDEX idx_users_created_at ON users(created_at);

## Продвинутая настройка с SQLAlchemy

Для более серьёзных проектов используйте Flask-SQLAlchemy:


from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)

class User(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(100), nullable=False)
email = db.Column(db.String(100), unique=True, nullable=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow)

def to_dict(self):
return {
'id': self.id,
'name': self.name,
'email': self.email,
'created_at': self.created_at.isoformat()
}

@app.route('/users', methods=['GET'])
def get_users():
users = User.query.all()
return jsonify([user.to_dict() for user in users])

@app.route('/users', methods=['POST'])
def create_user():
data = request.get_json()
user = User(name=data['name'], email=data['email'])
db.session.add(user)
db.session.commit()
return jsonify(user.to_dict()), 201

if __name__ == '__main__':
with app.app_context():
db.create_all()
app.run(debug=True)

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

SQLite имеет свои особенности, которые нужно учитывать:

Проблема Решение Пример кода
Блокировка при записи Включить WAL-режим PRAGMA journal_mode=WAL;
Медленные запросы Создать индексы CREATE INDEX idx_name ON table(column);
Большие транзакции Батчинг операций executemany() вместо execute()

Настройки для production:


import sqlite3

def optimize_db(db_path):
conn = sqlite3.connect(db_path)
conn.execute('PRAGMA journal_mode=WAL;')
conn.execute('PRAGMA synchronous=NORMAL;')
conn.execute('PRAGMA cache_size=1000;')
conn.execute('PRAGMA temp_store=MEMORY;')
conn.execute('PRAGMA mmap_size=268435456;') # 256MB
conn.commit()
conn.close()

# Использовать при инициализации
optimize_db('database.db')

## Миграции и версионирование

Создайте систему миграций:


import sqlite3
import os

class Migration:
def __init__(self, db_path):
self.db_path = db_path
self.conn = sqlite3.connect(db_path)
self.init_migrations_table()

def init_migrations_table(self):
self.conn.execute('''
CREATE TABLE IF NOT EXISTS migrations (
id INTEGER PRIMARY KEY,
name TEXT UNIQUE,
applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')

def apply_migration(self, name, sql):
cursor = self.conn.cursor()
cursor.execute('SELECT name FROM migrations WHERE name = ?', (name,))
if cursor.fetchone():
print(f"Migration {name} already applied")
return

cursor.executescript(sql)
cursor.execute('INSERT INTO migrations (name) VALUES (?)', (name,))
self.conn.commit()
print(f"Applied migration: {name}")

# Использование
migration = Migration('database.db')
migration.apply_migration('001_create_users', '''
CREATE TABLE users (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
email TEXT UNIQUE
);
''')

## Мониторинг и отладка

Добавьте логирование SQL-запросов:


import logging
import sqlite3

# Настройка логирования
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

class LoggingCursor(sqlite3.Cursor):
def execute(self, sql, *args):
logger.debug(f"SQL: {sql} | Args: {args}")
return super().execute(sql, *args)

class LoggingConnection(sqlite3.Connection):
def cursor(self):
return LoggingCursor(self)

# Использование
sqlite3.register_adapter(dict, lambda d: json.dumps(d))
sqlite3.register_converter("json", json.loads)

conn = LoggingConnection('database.db')

## Сравнение с альтернативами

СУБД Простота настройки Производительность Масштабируемость Подходит для
SQLite ⭐⭐⭐⭐⭐ ⭐⭐⭐ ⭐⭐ Прототипы, малые проекты
PostgreSQL ⭐⭐ ⭐⭐⭐⭐⭐ ⭐⭐⭐⭐⭐ Enterprise, сложные проекты
MySQL ⭐⭐⭐ ⭐⭐⭐⭐ ⭐⭐⭐⭐ Веб-приложения, CMS

## Интеграция с другими инструментами

SQLite отлично работает с:

  • Redis — для кэширования часто запрашиваемых данных
  • Celery — для фоновых задач по обработке данных
  • Alembic — для автоматических миграций
  • pytest — для тестирования с временными базами

Пример интеграции с Redis:


import redis
import json
from functools import wraps

redis_client = redis.Redis(host='localhost', port=6379, db=0)

def cache_query(timeout=300):
def decorator(f):
@wraps(f)
def wrapper(*args, **kwargs):
cache_key = f"query:{f.__name__}:{hash(str(args) + str(kwargs))}"
cached = redis_client.get(cache_key)
if cached:
return json.loads(cached)

result = f(*args, **kwargs)
redis_client.setex(cache_key, timeout, json.dumps(result))
return result
return wrapper
return decorator

@cache_query(timeout=600)
def get_user_stats():
db = get_db()
return db.execute('SELECT COUNT(*) as total FROM users').fetchone()

## Автоматизация и CI/CD

Создайте скрипт для автоматического бэкапа:


#!/bin/bash
# backup_db.sh
DATE=$(date +%Y%m%d_%H%M%S)
DB_PATH="database.db"
BACKUP_DIR="backups"
BACKUP_FILE="backup_${DATE}.db"

mkdir -p $BACKUP_DIR
sqlite3 $DB_PATH ".backup $BACKUP_DIR/$BACKUP_FILE"

# Удаляем бэкапы старше 7 дней
find $BACKUP_DIR -name "backup_*.db" -mtime +7 -delete

echo "Backup created: $BACKUP_DIR/$BACKUP_FILE"

Для автоматизации тестов:


import pytest
import tempfile
import os
from your_app import create_app, db

@pytest.fixture
def client():
db_fd, db_path = tempfile.mkstemp()
app = create_app({'TESTING': True, 'DATABASE': db_path})

with app.test_client() as client:
with app.app_context():
db.create_all()
yield client

os.close(db_fd)
os.unlink(db_path)

def test_create_user(client):
response = client.post('/users', json={
'name': 'Test User',
'email': 'test@example.com'
})
assert response.status_code == 201

## Развёртывание в production

Для production-окружения рекомендую использовать VPS или выделенный сервер. Настройте systemd-сервис:


# /etc/systemd/system/flask-app.service
[Unit]
Description=Flask App
After=network.target

[Service]
Type=simple
User=www-data
WorkingDirectory=/var/www/flask-app
Environment=FLASK_ENV=production
ExecStart=/var/www/flask-app/venv/bin/gunicorn --bind 0.0.0.0:5000 app:app
Restart=always

[Install]
WantedBy=multi-user.target

Интересные факты про SQLite:

  • Используется в Android, iOS, Firefox, Chrome
  • Один из самых тестируемых кодов в мире (100% покрытие тестами)
  • Поддерживает полнотекстовый поиск через FTS5
  • JSON-функции доступны начиная с версии 3.38

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

SQLite как конфигурационное хранилище:


class Config:
def __init__(self, db_path):
self.conn = sqlite3.connect(db_path)
self.conn.execute('''
CREATE TABLE IF NOT EXISTS config (
key TEXT PRIMARY KEY,
value TEXT,
type TEXT DEFAULT 'str'
)
''')

def get(self, key, default=None):
cursor = self.conn.cursor()
cursor.execute('SELECT value, type FROM config WHERE key = ?', (key,))
row = cursor.fetchone()
if not row:
return default

value, type_name = row
if type_name == 'int':
return int(value)
elif type_name == 'bool':
return value.lower() == 'true'
return value

def set(self, key, value):
type_name = type(value).__name__
self.conn.execute(
'INSERT OR REPLACE INTO config (key, value, type) VALUES (?, ?, ?)',
(key, str(value), type_name)
)
self.conn.commit()

# Использование
config = Config('config.db')
config.set('debug', True)
config.set('max_connections', 100)

## Статистика и мониторинг

Мониторинг размера базы и производительности:


def get_db_stats():
conn = sqlite3.connect('database.db')
stats = {}

# Размер базы
stats['file_size'] = os.path.getsize('database.db')

# Количество таблиц
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM sqlite_master WHERE type='table'")
stats['tables_count'] = cursor.fetchone()[0]

# Статистика по таблицам
cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
tables = cursor.fetchall()

for table in tables:
table_name = table[0]
cursor.execute(f"SELECT COUNT(*) FROM {table_name}")
stats[f'{table_name}_rows'] = cursor.fetchone()[0]

conn.close()
return stats

# Использование в endpoint
@app.route('/db-stats')
def db_stats():
return jsonify(get_db_stats())

## Заключение и рекомендации

SQLite идеально подходит для:

  • Прототипирования — быстрый старт без настройки инфраструктуры
  • Малых и средних проектов — до 100GB данных и умеренной нагрузки
  • Embedded-приложений — мобильные и IoT-устройства
  • Аналитики — локальная обработка данных

Не используйте SQLite для:

  • Высоконагруженных веб-приложений с частыми записями
  • Распределённых систем
  • Проектов, требующих репликации
  • Приложений с жёсткими требованиями к консистентности

Для развёртывания production-приложений рекомендую VPS с SSD-дисками для лучшей производительности SQLite. При росте нагрузки всегда можно мигрировать на PostgreSQL с минимальными изменениями в коде при использовании SQLAlchemy.

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


В этой статье собрана информация и материалы из различных интернет-источников. Мы признаем и ценим работу всех оригинальных авторов, издателей и веб-сайтов. Несмотря на то, что были приложены все усилия для надлежащего указания исходного материала, любая непреднамеренная оплошность или упущение не являются нарушением авторских прав. Все упомянутые товарные знаки, логотипы и изображения являются собственностью соответствующих владельцев. Если вы считаете, что какой-либо контент, использованный в этой статье, нарушает ваши авторские права, немедленно свяжитесь с нами для рассмотрения и принятия оперативных мер.

Данная статья предназначена исключительно для ознакомительных и образовательных целей и не ущемляет права правообладателей. Если какой-либо материал, защищенный авторским правом, был использован без должного упоминания или с нарушением законов об авторском праве, это непреднамеренно, и мы исправим это незамедлительно после уведомления. Обратите внимание, что переиздание, распространение или воспроизведение части или всего содержимого в любой форме запрещено без письменного разрешения автора и владельца веб-сайта. Для получения разрешений или дополнительных запросов, пожалуйста, свяжитесь с нами.

Leave a reply

Your email address will not be published. Required fields are marked