Flask-SQLAlchemy 与数据建模

## 第 5 章 · Flask-SQLAlchemy 与数据建模

本章目标：使用 Flask-SQLAlchemy 定义 Category 与 Product 模型及一对多关系；配置 Flask-Migrate 生成与应用迁移；在 admin 视图与 API 蓝图中完成 CRUD；理解 db.session 事务与查询 API；将 ch04 的 ProductForm 与模型绑定保存。

学时建议：5～6 小时（含 2 小时跟练）

前置：完成 flask-web ch04（ProductForm、admin 创建/编辑流程）。

5.1 场景说明：持久化商品数据

ch04 用内存字典 _PRODUCTS 存商品，重启即丢失。api-demo 需要与 shop-demo 类似的商品模型，但采用 SQLAlchemy ORM。

开发数据库使用 SQLite 文件 api_demo.db；生产可换 PostgreSQL（ch11 配置分环境）。

5.2 安装与扩展注册

pip install "flask-sqlalchemy>=3.1,<4" "flask-migrate>=4.0,<5"

api_demo/extensions.py：

from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_wtf.csrf import CSRFProtect

db = SQLAlchemy()
migrate = Migrate()
csrf = CSRFProtect()

api_demo/config.py 增加：

class Config:
    SECRET_KEY = os.environ.get("SECRET_KEY", "dev-only-change-in-prod")
    SQLALCHEMY_DATABASE_URI = os.environ.get(
        "DATABASE_URL",
        "sqlite:///api_demo.db",
    )
    SQLALCHEMY_TRACK_MODIFICATIONS = False

api_demo/__init__.py：

from api_demo.extensions import csrf, db, migrate

def create_app(config_name=None):
    # ...
    db.init_app(app)
    migrate.init_app(app, db)
    csrf.init_app(app)

    with app.app_context():
        import api_demo.models  # noqa: F401

    # register blueprints ...
    return app

5.3 模型定义

api_demo/models/category.py：

from api_demo.extensions import db


class Category(db.Model):
    __tablename__ = "categories"

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), nullable=False, unique=True)
    slug = db.Column(db.String(64), nullable=False, unique=True, index=True)

    products = db.relationship(
        "Product",
        back_populates="category",
        lazy="dynamic",
    )

    def __repr__(self):
        return f"<Category {self.slug}>"

api_demo/models/product.py：

from api_demo.extensions import db


class Product(db.Model):
    __tablename__ = "products"

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(120), nullable=False)
    slug = db.Column(db.String(64), nullable=False, unique=True, index=True)
    price = db.Column(db.Numeric(10, 2), nullable=False)
    stock = db.Column(db.Integer, nullable=False, default=0)
    is_published = db.Column(db.Boolean, default=False, nullable=False)

    category_id = db.Column(
        db.Integer,
        db.ForeignKey("categories.id"),
        nullable=False,
    )
    category = db.relationship("Category", back_populates="products")

    def __repr__(self):
        return f"<Product {self.slug}>"

api_demo/models/__init__.py：

from api_demo.models.category import Category
from api_demo.models.product import Product

__all__ = ["Category", "Product"]

5.4 Flask-Migrate 迁移

在项目根（含 wsgi.py）执行：

$env:FLASK_APP = "wsgi:app"
flask db init
flask db migrate -m "add category and product"
flask db upgrade

5.5 种子数据

api_demo/seed.py（可选脚本）：

from api_demo import create_app
from api_demo.extensions import db
from api_demo.models import Category, Product

app = create_app("development")

with app.app_context():
    if Category.query.first():
        print("already seeded")
    else:
        c1 = Category(name="数码", slug="digital")
        c2 = Category(name="家居", slug="home")