- Home »

Как использовать 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.
Полезные ресурсы:
В этой статье собрана информация и материалы из различных интернет-источников. Мы признаем и ценим работу всех оригинальных авторов, издателей и веб-сайтов. Несмотря на то, что были приложены все усилия для надлежащего указания исходного материала, любая непреднамеренная оплошность или упущение не являются нарушением авторских прав. Все упомянутые товарные знаки, логотипы и изображения являются собственностью соответствующих владельцев. Если вы считаете, что какой-либо контент, использованный в этой статье, нарушает ваши авторские права, немедленно свяжитесь с нами для рассмотрения и принятия оперативных мер.
Данная статья предназначена исключительно для ознакомительных и образовательных целей и не ущемляет права правообладателей. Если какой-либо материал, защищенный авторским правом, был использован без должного упоминания или с нарушением законов об авторском праве, это непреднамеренно, и мы исправим это незамедлительно после уведомления. Обратите внимание, что переиздание, распространение или воспроизведение части или всего содержимого в любой форме запрещено без письменного разрешения автора и владельца веб-сайта. Для получения разрешений или дополнительных запросов, пожалуйста, свяжитесь с нами.