Интеграция базы данных во Flask

Большинству веб-приложений требуется постоянное хранилище данных.

В этой главе используется встроенная в Python база данных SQLite, чтобы показать, как интегрировать базу данных во Flask без установки дополнительных зависимостей.

Быстрый старт с SQLite

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

Это лучший выбор для обучения и прототипирования — не нужно устанавливать сервер базы данных или что-либо настраивать.

Пример

# Простой пример SQLite (без использования Flask)
import sqlite3

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

# Создать таблицу
conn.execute("CREATE TABLE IF NOT EXISTS user (id INTEGER PRIMARY KEY, username TEXT UNIQUE, password TEXT)")
# Вставить данные
conn.execute("INSERT INTO user (username, password) VALUES (?, ?)", ("example", "pass123"))
conn.commit()

# Запросить данные
rows = conn.execute("SELECT * FROM user").fetchall()
for row in rows:
    print(dict(row))  # Вывод: {'id': 1, 'username': 'example', 'password': 'pass123'}

conn.close()

Использование объекта g в Flask для управления подключениями к базе данных

gЭто специальный объект, жизненный цикл которого — один запрос.

Сохранение подключения к базе данныхgЭто гарантирует, что несколько операций с базой данных в рамках одного запроса используют одно и то же подключение, которое автоматически освобождается после завершения запроса.

Пример

# Путь к файлу: db.py (модуль управления базой данных)
import sqlite3
from flask import g, current_app

def get_db():
    """Получить подключение к базе данных, повторно используемое в рамках одного запроса"""
    if "db" not in g:
        # Прочитать путь к базе данных из конфигурации
        db_path = current_app.config.get("DATABASE", "example.db")
        g.db = sqlite3.connect(db_path)
        g.db.row_factory = sqlite3.Row  # Разрешить доступ к результатам по имени поля
    return g.db

def close_db(error=None):
    """Закрыть подключение к базе данных (автоматически вызывается в конце запроса)"""
    db = g.pop("db", None)
    if db is not None:
        db.close()

def init_db():
    """Инициализировать структуру таблиц базы данных"""
    db = get_db()
    db.execute("""
        CREATE TABLE IF NOT EXISTS user (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            username TEXT UNIQUE NOT NULL,
            password TEXT NOT NULL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)
    db.execute("""
        CREATE TABLE IF NOT EXISTS post (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            title TEXT NOT NULL,
            body TEXT NOT NULL,
            author_id INTEGER NOT NULL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            FOREIGN KEY (author_id) REFERENCES user (id)
        )
    """)
    db.commit()

def init_app(app):
    """Регистрирует функции, связанные с базой данных, в приложении Flask"""
    # Автоматически закрывать соединение с базой данных в конце запроса
    app.teardown_appcontext(close_db)
    # Зарегистрировать команду CLI: flask init-db
    app.cli.command("init-db")(init_db)

Интеграция паттерна Factory

Используйте паттерн фабрики для интеграции базы данных с другими модулями:

Пример

# Путь к файлу: app.py
from flask import Flask

def create_app():
    app = Flask(__name__)
    app.secret_key = "dev-secret-key"

    # Настройте путь к базе данных (хранится в папке instance)
    import os
    app.config["DATABASE"] = os.path.join(app.instance_path, "example.db")

    # Убедитесь, что папка instance существует
    os.makedirs(app.instance_path, exist_ok=True)

    # Инициализируйте базу данных
    from db import init_app as init_db
    init_db(app)

    # Зарегистрируйте blueprint
    from auth import bp as auth_bp
    from blog import bp as blog_bp
    app.register_blueprint(auth_bp, url_prefix="/auth")
    app.register_blueprint(blog_bp)

    return app

Напишите полноценный API блога

Объедините базу данных, чтобы реализовать операции CRUD (Create, Read, Update, Delete) для статей:

Пример

# Путь к файлу: blog.py
from flask import Blueprint, request, jsonify, abort, g
from db import get_db

bp = Blueprint("blog", __name__)

@bp.get("/api/posts")
def list_posts():
    """Получить список всех статей"""
    db = get_db()
    posts = db.execute(
        "SELECT p.id, p.title, p.body, p.created_at, u.username"
        " FROM post p JOIN user u ON p.author_id = u.id"
        " ORDER BY p.created_at DESC"
    ).fetchall()
    # Преобразовать список объектов Row в список словарей
    return jsonify([dict(post) for post in posts])

@bp.get("/api/posts/<int:post_id>")
def get_post(post_id):
    """Получить сведения об одной статье"""
    db = get_db()
    post = db.execute(
        "SELECT id, title, body, created_at FROM post WHERE id = ?",
        (post_id,)
    ).fetchone()

    if post is None:
        abort(404, description=f"Статья {post_id} не существует")

    return jsonify(dict(post))

@bp.post("/api/posts")
def create_post():
    """Создать новую статью"""
    data = request.json
    title = data.get("title", "").strip()
    body = data.get("body", "").strip()
    author_id = data.get("author_id", 1)  # ID автора по умолчанию

    if not title:
        return jsonify({"error": "Заголовок не может быть пустым"}), 400

    db = get_db()
    cursor = db.execute(
        "INSERT INTO post (title, body, author_id) VALUES (?, ?, ?)",
        (title, body, author_id)
    )
    db.commit()

    return jsonify({"id": cursor.lastrowid, "title": title, "message": "Статья успешно создана"}), 201

@bp.put("/api/posts/<int:post_id>")
def update_post(post_id):
    """Обновить статью"""
    data = request.json
    title = data.get("title", "").strip()
    body = data.get("body", "").strip()

    if not title:
        return jsonify({"error": "Заголовок не может быть пустым"}), 400

    db = get_db()
    db.execute(
        "UPDATE post SET title = ?, body = ? WHERE id = ?",
        (title, body, post_id)
    )
    db.commit()

    return jsonify({"id": post_id, "message": "Статья успешно обновлена"})

@bp.delete("/api/posts/<int:post_id>")
def delete_post(post_id):
    """Удалить статью"""
    db = get_db()
    db.execute("DELETE FROM post WHERE id = ?", (post_id,))
    db.commit()
    return jsonify({"message": "Статья удалена"})

Тестирование API:

# 获取文章列表
$ curl http://127.0.0.1:5000/api/posts

# 创建文章
$ curl -X POST http://127.0.0.1:5000/api/posts \
  -H "Content-Type: application/json" \
  -d '{"title":"EXAMPLE 教程","body":"Flask 入门教程内容","author_id":1}'

# 更新文章
$ curl -X PUT http://127.0.0.1:5000/api/posts/1 \
  -H "Content-Type: application/json" \
  -d '{"title":"EXAMPLE 教程（修订版）","body":"更新后的内容"}'

# 删除文章
$ curl -X DELETE http://127.0.0.1:5000/api/posts/1

Лучшие практики: Когда API возвращает данные JSON, используйте правильный код состояния HTTP — для создания используйте201, используйте200Используйте для ошибок клиента400Используйте для несуществующих ресурсов404Это упрощает фронтенд-коду определение результата запроса.

Команда инициализации базы данных

Через Flask CLI вы можете создавать пользовательские команды для инициализации базы данных:

# 初始化数据库（创建表结构）
(.venv) $ flask init-db

# 验证表已创建
(.venv) $ sqlite3 instance/example.db ".tables"
post  user

Защита от SQL-инъекций

Использование параметризованных запросов (?заполнители) — ключ к предотвращению SQL-инъекций:

Пример

# Правильный подход: используйте ? заполнители (параметризованные запросы)
# Драйвер SQLite автоматически экранирует параметры, чтобы предотвратить выполнение вредоносного ввода как SQL
db.execute("SELECT * FROM user WHERE username = ?", (username,))

# Неправильный подход: используйте f-строку для конкатенации SQL (чрезвычайно опасно!)
# db.execute(f"SELECT * FROM user WHERE username = '{username}'")
# Вредоносный ввод '; DROP TABLE user; -- удалил бы всю таблицу!

Использование Flask-SQLAlchemy (продвинутое необязательное чтение)

В крупных проектах ручное написание SQL постепенно становится утомительным.

Flask-SQLAlchemyЭто широко распространённое расширение, которое предоставляет возможность ORM (Object-Relational Mapping):

Пример

# Установка: pip install flask-sqlalchemy
from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///example.db"
db = SQLAlchemy(app)

# Определение модели данных (структура таблицы, представленная классом Python)
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)

    def __repr__(self):
        return f"<User {self.username}>"

# Создать таблицу
with app.app_context():
    db.create_all()

# Создать пользователя
user = User(username="example", email="[email protected]")
db.session.add(user)
db.session.commit()

# Запрос пользователя
users = User.query.all()
example_user = User.query.filter_by(username="example").first()

В этой главе используется встроенный SQLite, чтобы избежать дополнительных зависимостей, что делает её подходящей для обучения и небольших проектов. Для реальных производственных проектов мы рекомендуем использовать SQLAlchemy или аналогичную ORM, что может значительно сократить повторяющийся код и повысить безопасность.

Другие расширения