Flask - SQLAlchemy

Flask – 集成 SQLAlchemy

虽然 Flask 不强制要求使用特定的数据库或对象关系映射器（Object-Relational Mapper，简称 ORM），但 SQLAlchemy 是在 Python 生态系统中处理关系型数据库（如 PostgreSQL、MySQL、SQLite）的事实标准。Flask-SQLAlchemy 是一个 Flask 扩展，它无缝集成了 SQLAlchemy，为在 Flask 应用程序中管理数据库连接、会话和模型（Models）提供了有用的默认设置和实用工具。

什么是 ORM？

对象关系映射（ORM）是一种技术，它允许你使用你偏好的编程语言（在此例中为 Python）中的面向对象范式与数据库进行交互，例如查询数据或保存记录。你不再需要编写原始的 SQL 查询（例如，SELECT * FROM users WHERE city = 'New York'），而是使用 Python 对象和方法（例如，User.query.filter_by(city='New York').all()）。ORM 负责在你的 Python 代码和底层的 SQL 数据库操作之间进行转换。

使用 Flask-SQLAlchemy 的好处：

• 抽象性： 隐藏了不同 SQL 数据库之间的差异。
• 生产力： 减少了常见数据库操作（CRUD - 创建、读取、更新、删除）的样板代码。
• 可维护性： 将数据库表映射到 Python 类（模型），使数据结构清晰且更易于管理。
• 集成： 在 Flask 请求生命周期内自动管理数据库会话和连接。

设置 Flask-SQLAlchemy

步骤 1：安装扩展。

pip install Flask-SQLAlchemy

你还需要为你打算使用的特定数据库安装数据库驱动程序（例如，PostgreSQL 的 psycopg2-binary，MySQL 的 mysqlclient 或 PyMySQL）。SQLite 已内置于 Python 中，因此不需要额外的驱动程序。

步骤 2：配置数据库 URI。

在你的 Flask 应用程序设置中，配置 SQLALCHEMY_DATABASE_URI。这个字符串告诉 SQLAlchemy 你的数据库在哪里。

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)

# 配置示例：
# SQLite (相对路径)
# app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///mydatabase.db'
# SQLite (绝对路径)
# app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:////path/to/mydatabase.db'
# PostgreSQL
# app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://user:password@host:port/dbname'
# MySQL
# app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://user:password@host:port/dbname'

# 在此示例中使用 SQLite
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///students.sqlite3'
# 可选：如果不需要，禁用 SQLAlchemy 事件系统，可提高性能
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
# 后面示例中用于闪现消息所需
app.config['SECRET_KEY'] = 'your-secret-key'

步骤 3：初始化扩展。

创建 SQLAlchemy 类的一个实例，并传入你的 Flask 应用程序对象。

db = SQLAlchemy(app)

这个 db 对象将成为你访问 SQLAlchemy 功能的主要入口点，包括定义模型和管理会话。

步骤 4：定义模型（Models）。

模型是继承自 db.Model 的 Python 类。每个类代表数据库中的一个表，类属性（使用 db.Column 定义）代表表中的列。

class Student(db.Model):  # 模型类名使用单数、首字母大写
    # 明确定义表名（可选，默认为类名的小写形式）
    # __tablename__ = 'students_table'

    # 定义列
    id = db.Column(db.Integer, primary_key=True) # 主键列
    name = db.Column(db.String(100), nullable=False) # 字符串类型，不可为空
    city = db.Column(db.String(50))
    address = db.Column(db.String(200))
    pin_code = db.Column(db.String(10)) # 为清晰起见，将 'pin' 重命名为 'pin_code'

    # 可选：定义一个 __init__ 方法以便于创建对象
    def __init__(self, name, city, address, pin_code):
        self.name = name
        self.city = city
        self.address = address
        self.pin_code = pin_code

    # 可选：定义一个 __repr__ 方法以便于调试输出
    def __repr__(self):
        return f'<Student {self.name}>'

步骤 5：创建数据库表。

在与数据库交互之前，需要创建由你的模型定义的表。对于简单情况或初始设置，你可以使用 db.create_all()。

# 通常这应该运行一次，可能在设置脚本中，或者
# 在应用程序启动时，并在应用程序上下文中进行有条件的运行。
with app.app_context():
    db.create_all()

关于迁移（Migrations）的重要说明： db.create_all() 只创建不存在的表。它不处理对现有表结构的更新（例如，添加新列）。要在生产应用程序中随时间管理数据库模式更改，请使用数据库迁移工具，如 Alembic，通常通过 Flask-Migrate 扩展集成。（pip install Flask-Migrate）

执行 CRUD 操作

Flask-SQLAlchemy 为你管理数据库会话（sessions）。你通过 db.session 对象与数据库进行交互。

• 创建（添加）： 创建你的模型类的实例，并使用 db.session.add() 将其添加到会话。然后，使用 db.session.commit() 提交会话以将更改保存到数据库。
• 读取（查询）： 使用模型类上的 query 属性（例如，Student.query）。常用的查询方法包括 all()（获取所有记录），get(id)（按主键获取），filter_by(...)（按关键字参数过滤），filter(...)（使用类似 SQL 的表达式过滤），first()（获取第一个结果），count()（获取结果数量）。
• 更新： 查询你想要更新的对象，修改其属性，然后调用 db.session.commit()。
• 删除： 查询你想要删除的对象，将其传递给 db.session.delete()，然后调用 db.session.commit()。

示例应用：学生记录

让我们构建一个小型应用程序来列出和添加学生记录。

显示所有学生的视图函数 (/) 并渲染 show_all.html：

from flask import render_template

@app.route('/')
def show_all():
    students = Student.query.all() # 获取所有学生记录
    return render_template('show_all.html', students=students)

模板文件 templates/show_all.html：

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Student List</title>
</head>
<body>
    <h3>
        <a href="{{ url_for('show_all') }}">Student Records - Flask-SQLAlchemy</a>
    </h3>
    <hr/>

    {# 显示闪现消息 #}
    {% with messages = get_flashed_messages() %}
      {% if messages %}
        <ul class=flashes>
        {% for message in messages %}
          <li>{{ message }}</li>
        {% endfor %}
        </ul>
      {% endif %}
    {% endwith %}

    <h3>
        Students
        (<a href="{{ url_for('add_student') }}">Add New Student</a>)
    </h3>
    <table border="1">
        <thead>
            <tr>
                <th>Name</th>
                <th>City</th>
                <th>Address</th>
                <th>PIN Code</th>
            </tr>
        </thead>
        <tbody>
            {% for student in students %}
            <tr>
                <td>{{ student.name }}</td>
                <td>{{ student.city }}</td>
                <td>{{ student.address }}</td>
                <td>{{ student.pin_code }}</td>
            </tr>
            {% else %}
            <tr>
                <td colspan="4">No students found.</td>
            </tr>
            {% endfor %}
        </tbody>
    </table>
</body>
</html>

处理添加新学生的视图函数 (/new)，渲染 new_student.html（GET 请求）并处理表单数据（POST 请求）：

from flask import request, flash, redirect, url_for

@app.route('/new', methods=['GET', 'POST'])
def add_student():
    if request.method == 'POST':
        # 基础验证（考虑使用 Flask-WTF 进行更健壮的验证）
        if not request.form.get('name') or not request.form.get('city') or                not request.form.get('address') or not request.form.get('pin_code'):
            flash('Please enter all the fields', 'error')
        else:
            # 从表单数据创建新的 Student 对象
            new_student = Student(
                name=request.form['name'],
                city=request.form['city'],
                address=request.form['address'],
                pin_code=request.form['pin_code']
            )

            try:
                # 将新学生添加到会话并提交
                db.session.add(new_student)
                db.session.commit()
                flash('Record was successfully added', 'success')
                return redirect(url_for('show_all')) # 重定向回列表页面
            except Exception as e:
                db.session.rollback() # 发生错误时回滚
                flash(f'Error adding record: {e}', 'error')

    # 如果是 GET 请求或验证失败，渲染表单模板
    return render_template('new_student.html')

模板文件 templates/new_student.html：

<!DOCTYPE html>
<html>
<head>
    <title>Add Student</title>
</head>
<body>
    <h3>Add New Student</h3>
    <hr/>
    {% with messages = get_flashed_messages(with_categories=true) %}
      {% if messages %}
        <ul>
        {% for category, message in messages %}
          <li class="{{ category }}">{{ message }}</li>
        {% endfor %}
        </ul>
      {% endif %}
    {% endwith %}

    <form action="{{ url_for('add_student') }}" method="post">
        <div>
            <label for="name">Name</label><br/>
            <input type="text" name="name" placeholder="Full Name" required/>
        </div>
        <div>
            <label for="city">City</label><br/>
            <input type="text" name="city" placeholder="City" required/>
        </div>
        <div>
            <label for="address">Address</label><br/>
            <textarea name="address" placeholder="Street Address" required></textarea>
        </div>
        <div>
            <label for="pin_code">PIN Code</label><br/>
            <input type="text" name="pin_code" placeholder="PIN Code" required/>
        </div>
        <div>
            <input type="submit" value="Submit"/>
        </div>
    </form>
    <br>
    <a href="{{ url_for('show_all') }}">Back to List</a>
</body>
</html>

完整的应用程序代码 (app.py)：

# --- app.py 开始 ---
from flask import Flask, request, flash, url_for, redirect, render_template
from flask_sqlalchemy import SQLAlchemy

# --- 应用初始化和配置 ---
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///students.sqlite3'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SECRET_KEY'] = "very-random-secret-string"

db = SQLAlchemy(app)

# --- 模型定义 ---
class Student(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100), nullable=False)
    city = db.Column(db.String(50))
    address = db.Column(db.String(200))
    pin_code = db.Column(db.String(10))

    def __init__(self, name, city, address, pin_code):
        self.name = name
        self.city = city
        self.address = address
        self.pin_code = pin_code

    def __repr__(self):
        return f'<Student {self.name}>'

# --- 路由和视图函数 ---
@app.route('/')
def show_all():
    students = Student.query.order_by(Student.name).all() # 按姓名排序
    return render_template('show_all.html', students=students)

@app.route('/new', methods=['GET', 'POST'])
def add_student():
    if request.method == 'POST':
        if not request.form.get('name') or not request.form.get('city') or \
           not request.form.get('address') or not request.form.get('pin_code'):
            flash('Please enter all the fields', 'error')
        else:
            new_student = Student(
                name=request.form['name'],
                city=request.form['city'],
                address=request.form['address'],
                pin_code=request.form['pin_code']
            )
            try:
                db.session.add(new_student)
                db.session.commit()
                flash('Record was successfully added', 'success')
                return redirect(url_for('show_all'))
            except Exception as e:
                db.session.rollback()
                flash(f'Error adding record: {e}', 'error')

    return render_template('new_student.html')

# --- 用于创建表的工具函数（运行一次）---
def create_tables():
     with app.app_context():
        print("Creating database tables...")
        db.create_all()
        print("Tables created.")

# --- 运行应用程序 ---
if __name__ == '__main__':
    # 可选：如果直接运行脚本，则创建表
    # 在生产环境中要小心使用；请改用迁移。
    # create_tables()
    app.run(debug=True)

运行此应用：

• 将 Python 代码保存为 app.py。
• 创建一个名为 templates 的文件夹，并将 show_all.html 和 new_student.html 保存到其中。
• 运行应用程序：python app.py。
• 在浏览器中打开 http://localhost:5000/。
• 第一次运行（或如果 students.sqlite3 文件不存在）时，你可能需要创建表。你可以临时取消注释 if __name__ == '__main__': 块内的 create_tables() 调用，运行一次脚本，然后再注释掉。更好的方法是使用 Flask shell：运行 flask shell，然后在 shell 中输入：from app import db，db.create_all()，exit()。
• 你应该会看到（最初是空的）学生列表。点击“Add New Student”，填写表单并提交。新学生应该会出现在列表中。

此示例涵盖了基础知识。SQLAlchemy 提供了更多强大的功能，包括模型之间的关系（一对多、多对多）、更复杂的查询和事务管理。请参考 Flask-SQLAlchemy 和 SQLAlchemy 文档以获取高级用法。