【Flask从入门到精通：第十一课：逻辑外键、数据迁移】

  • 开发
  • 18

逻辑外键

也叫虚拟外键。主要就是在开发中为了减少数据库的性能消耗，提升系统运行效率，一般项目中如果单表数据太大[千万级别]就不会使用数据库本身维护的物理外键，而是采用由ORM或者我们逻辑代码进行查询关联的逻辑外键。

SQLAlchemy设置外键模型的虚拟外键，有2种方案：

方案1，查询数据时临时指定逻辑外键的映射关系：

模型类.query.join(模型类,主模型.主键==外键模型.外键).join(模型类,主模型.主键==外键模型.外键).with_entities(字段1,字段2.label("字段别名"),....).all()

方案2，在模型声明时指定逻辑外键的映射关系(最常用，这种设置方案，在操作模型时与原来默认设置的物理外键的关联操作是一模一样的写法)：

class Student(db.Model):
    id = db.Column(db.Integer, primary_key=True, comment="主键")
    # 虚拟外键，原有参数不变，新增2个表达关联关系的属性：
    # primaryjoin, 指定2个模型之间的主外键关系，相当于原生SQL语句中的join
    # foreign_keys，指定外键
    address_list = db.relationship("StudentAddress", uselist=True, backref="student", lazy="subquery", primaryjoin="Student.id==StudentAddress.student_id", foreign_keys="StudentAddress.student_id")

class StudentAddress(db.Model):
    # 原来的外键设置为普通索引即可。
    student_id = db.Column(db.Integer, comment="学生id")

例1，虚拟外键使用的方案1，代码：

import json
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.orm import backref
from datetime import datetime

db = SQLAlchemy()
app = Flask(__name__, template_folder="templates", static_folder="static")

# 配置
app.config.update({
    "DEBUG": True,
    "SQLALCHEMY_DATABASE_URI": "mysql://root:123@127.0.0.1:3306/flaskdemo?charset=utf8mb4",
    # 如果使用pymysql，则需要在连接时指定pymysql
    # "SQLALCHEMY_DATABASE_URI": "mysql+pymysql://root:123@127.0.0.1:3306/flaskdemo?charset=utf8mb4"
    # 动态追踪修改设置，如未设置只会提示警告，设置False即可
    "SQLALCHEMY_TRACK_MODIFICATIONS": False,
    # ORM执行SQL查询时是哦否显示原始SQL语句，debug模式下可以开启
    "SQLALCHEMY_ECHO": True,
})

db.init_app(app)


class Student(db.Model):
    """学生信息模型"""
    # 声明与当前模型绑定的数据表名称
    __tablename__ = "td_student"
    id = db.Column(db.Integer, primary_key=True,comment="主键")
    name = db.Column(db.String(15), comment="姓名")
    age = db.Column(db.SmallInteger, comment="年龄")
    sex = db.Column(db.Boolean, default=True, comment="性别")
    email = db.Column(db.String(128), comment="邮箱地址")
    money = db.Column(db.Numeric(10,2), default=0.0, comment="钱包")

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


class Course(db.Model):
    """课程数据模型"""
    __tablename__ = "td_course"
    id = db.Column(db.Integer, primary_key=True, comment="主键")
    name = db.Column(db.String(64), unique=True, comment="课程")
    price = db.Column(db.Numeric(7, 2))

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


class StudentCourse(db.Model):
    """学生和课程之间的关系模型"""
    __tablename__ = "td_student_course"
    id = db.Column(db.Integer, primary_key=True, comment="主键")
    student_id = db.Column(db.Integer, index=True)
    course_id = db.Column(db.Integer, index=True)
    score = db.Column(db.Numeric(4,1), default=0, comment="成绩")
    time = db.Column(db.DateTime, default=datetime.now, comment="考试时间")


@app.route("/create")
def create_table():
    db.create_all()  # 为项目中被识别的所有模型创建数据表
    return "ok"


@app.route("/drop")
def drop_table():
    db.drop_all()  # 为项目中被识别的所有模型删除数据表
    return "ok"


@app.route("/")
def index():
    return "ok"

@app.route("/a1")
def a1():
    # 添加测试数据
    stu0 = Student(name="xiaoming-0",age=15,sex=True,email="xiaoming0@qq.com", money=1000)
    stu1 = Student(name="xiaoming-1",age=15,sex=True,email="xiaoming1@qq.com", money=1000)
    stu2 = Student(name="xiaoming-2",age=15,sex=True,email="xiaoming2@qq.com", money=1000)
    stu3 = Student(name="xiaoming-3",age=15,sex=True,email="xiaoming3@qq.com", money=1000)
    stu4 = Student(name="xiaoming-4",age=15,sex=True,email="xiaoming4@qq.com", money=1000)

    db.session.add_all([stu0,stu1,stu2,stu3,stu4])
    course1 = Course(name="python基础第1季", price=1000)
    course2 = Course(name="python基础第2季", price=1000)
    course3 = Course(name="python基础第3季", price=1000)
    course4 = Course(name="python基础第4季", price=1000)
    course5 = Course(name="python基础第5季", price=1000)
    db.session.add_all([course1, course2, course3, course4, course5])

    data = [
        StudentCourse(student_id=1,course_id=1,score=60,time=datetime.now()),
        StudentCourse(student_id=1,course_id=2,score=60,time=datetime.now()),
        StudentCourse(student_id=1,course_id=3,score=60,time=datetime.now()),
        StudentCourse(student_id=2,course_id=1,score=60,time=datetime.now()),
        StudentCourse(student_id=2,course_id=2,score=60,time=datetime.now()),
        StudentCourse(student_id=3,course_id=3,score=60,time=datetime.now()),
        StudentCourse(student_id=3,course_id=4,score=60,time=datetime.now()),
        StudentCourse(student_id=4,course_id=5,score=60,time=datetime.now()),
        StudentCourse(student_id=4,course_id=1,score=60,time=datetime.now()),
        StudentCourse(student_id=4,course_id=2,score=60,time=datetime.now()),
        StudentCourse(student_id=5,course_id=1,score=60,time=datetime.now()),
        StudentCourse(student_id=5,course_id=2,score=60,time=datetime.now()),
        StudentCourse(student_id=5,course_id=3,score=60,time=datetime.now()),
        StudentCourse(student_id=5,course_id=4,score=60,time=datetime.now()),
    ]
    db.session.add_all(data)
    db.session.commit()

    return "ok"


@app.route("/q1")
def q1():
    # 使用逻辑外键来查询数据
    # 主模型.query.join(从模型类名, 关系语句)
    # 主模型.query.join(从模型类名, 主模型.主键==从模型类名.外键)

    # 课程[python基础第3季]有多少学生在读？
    # 分2步查询
    course = Course.query.filter(Course.name == "python基础第3季").first()
    num = StudentCourse.query.filter(StudentCourse.course_id == course.id).count()
    print(course, num)

    # 关联查询, 2表关联
    # data = Course.query.join(
    #     StudentCourse,
    #     Course.id==StudentCourse.course_id
    # ).with_entities(
    #     Course.id, Course.name, Course.price, StudentCourse.student_id, StudentCourse.score
    # ).filter(Course.name == "python基础第3季").all()
    # print(data)

    # 直接统计，不需要任何字段
    ret = Course.query.join(
        StudentCourse,
        Course.id==StudentCourse.course_id
    ).filter(Course.name == "python基础第3季").count()

    print(ret)

    return "ok"


@app.route("/q2")
def q2():
    # 查询课程[python基础第3季]有哪些学生在读？3表关联
    # 正向查询和逆向查询都是通过声明临时外键关系来完成关联查询操作。
    student_list = Course.query.join(
        StudentCourse,
        Course.id == StudentCourse.course_id
    ).join(
        Student,
        Student.id == StudentCourse.student_id
    ).with_entities(
        Course.name.label("course_name"), StudentCourse.score.label("score"), Student.name.label("student_name"),
    ).filter(Course.name == "python基础第3季").all()

    print(student_list)

    return "ok"


@app.route("/q3")
def q3():
    # xiaoming-2 购买了那些课程？
    course_list = Student.query.join(
        StudentCourse,
        Student.id == StudentCourse.student_id
    ).join(
        Course,
        Course.id == StudentCourse.course_id
    ).with_entities(
        Course.name.label("course_name"),
        StudentCourse.score.label("score")
    ).filter(Student.name == "xiaoming-2").all()

    print(course_list)

    return "ok"

if __name__ == '__main__':
    app.run()

例2，虚拟外键使用的方案2，代码：

import json
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.orm import backref
from datetime import datetime

db = SQLAlchemy()
app = Flask(__name__, template_folder="templates", static_folder="static")

# 配置
app.config.update({
    "DEBUG": True,
    "SQLALCHEMY_DATABASE_URI": "mysql://root:123@127.0.0.1:3306/flaskdemo?charset=utf8mb4",
    # 如果使用pymysql，则需要在连接时指定pymysql
    # "SQLALCHEMY_DATABASE_URI": "mysql+pymysql://root:123@127.0.0.1:3306/flaskdemo?charset=utf8mb4"
    # 动态追踪修改设置，如未设置只会提示警告，设置False即可
    "SQLALCHEMY_TRACK_MODIFICATIONS": False,
    # ORM执行SQL查询时是哦否显示原始SQL语句，debug模式下可以开启
    "SQLALCHEMY_ECHO": True,
})

db.init_app(app)


class Student(db.Model):
    """学生信息模型"""
    # 声明与当前模型绑定的数据表名称
    __tablename__ = "ts_student"
    id = db.Column(db.Integer, primary_key=True,comment="主键")
    name = db.Column(db.String(15), comment="姓名")
    age = db.Column(db.SmallInteger, comment="年龄")
    sex = db.Column(db.Boolean, default=True, comment="性别")
    email = db.Column(db.String(128), comment="邮箱地址")
    money = db.Column(db.Numeric(10,2), default=0.0, comment="钱包")

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


class Course(db.Model):
    """课程数据模型"""
    __tablename__ = "ts_course"
    id = db.Column(db.Integer, primary_key=True, comment="主键")
    name = db.Column(db.String(64), unique=True, comment="课程")
    price = db.Column(db.Numeric(7, 2))

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


class StudentCourse(db.Model):
    """学生和课程之间的关系模型"""
    __tablename__ = "ts_student_course"
    id = db.Column(db.Integer, primary_key=True, comment="主键")
    student_id = db.Column(db.Integer, index=True)
    course_id = db.Column(db.Integer, index=True)
    score = db.Column(db.Numeric(4,1), default=0, comment="成绩")
    time = db.Column(db.DateTime, default=datetime.now, comment="考试时间")

    student = db.relationship(
        "Student",
        uselist=False,
        backref=backref("to_course", uselist=True),
        lazy="subquery",
        primaryjoin="Student.id==StudentCourse.student_id",
        foreign_keys="StudentCourse.student_id"
    )

    course = db.relationship(
        "Course",
        uselist=False,
        backref=backref("to_student", uselist=True),
        lazy="subquery",
        primaryjoin="Course.id==StudentCourse.course_id",
        foreign_keys="StudentCourse.course_id"
    )

@app.route("/create")
def create_table():
    db.create_all()  # 为项目中被识别的所有模型创建数据表
    return "ok"


@app.route("/drop")
def drop_table():
    db.drop_all()  # 为项目中被识别的所有模型删除数据表
    return "ok"


@app.route("/")
def index():
    return "ok"

@app.route("/a1")
def a1():
    # 添加测试数据
    stu0 = Student(name="xiaoming-0",age=15,sex=True,email="xiaoming0@qq.com", money=1000)
    stu1 = Student(name="xiaoming-1",age=15,sex=True,email="xiaoming1@qq.com", money=1000)
    stu2 = Student(name="xiaoming-2",age=15,sex=True,email="xiaoming2@qq.com", money=1000)
    stu3 = Student(name="xiaoming-3",age=15,sex=True,email="xiaoming3@qq.com", money=1000)
    stu4 = Student(name="xiaoming-4",age=15,sex=True,email="xiaoming4@qq.com", money=1000)

    db.session.add_all([stu0,stu1,stu2,stu3,stu4])
    course1 = Course(name="python基础第1季", price=1000)
    course2 = Course(name="python基础第2季", price=1000)
    course3 = Course(name="python基础第3季", price=1000)
    course4 = Course(name="python基础第4季", price=1000)
    course5 = Course(name="python基础第5季", price=1000)
    db.session.add_all([course1, course2, course3, course4, course5])

    data = [
        StudentCourse(student_id=1,course_id=1,score=60,time=datetime.now()),
        StudentCourse(student_id=1,course_id=2,score=60,time=datetime.now()),
        StudentCourse(student_id=1,course_id=3,score=60,time=datetime.now()),
        StudentCourse(student_id=2,course_id=1,score=60,time=datetime.now()),
        StudentCourse(student_id=2,course_id=2,score=60,time=datetime.now()),
        StudentCourse(student_id=3,course_id=3,score=60,time=datetime.now()),
        StudentCourse(student_id=3,course_id=4,score=60,time=datetime.now()),
        StudentCourse(student_id=4,course_id=5,score=60,time=datetime.now()),
        StudentCourse(student_id=4,course_id=1,score=60,time=datetime.now()),
        StudentCourse(student_id=4,course_id=2,score=60,time=datetime.now()),
        StudentCourse(student_id=5,course_id=1,score=60,time=datetime.now()),
        StudentCourse(student_id=5,course_id=2,score=60,time=datetime.now()),
        StudentCourse(student_id=5,course_id=3,score=60,time=datetime.now()),
        StudentCourse(student_id=5,course_id=4,score=60,time=datetime.now()),
    ]
    db.session.add_all(data)
    db.session.commit()

    return "ok"


@app.route("/q1")
def q1():
    student = Student.query.filter(Student.name == "xiaoming-0").first()
    # print(student.to_course)
    # [<StudentCourse 1>, <StudentCourse 2>, <StudentCourse 3>]

    print([{
        "course_name": item.course.name,
        "score": item.score,
        "student_name": item.student.name,
    } for item in student.to_course])
    return "ok"


if __name__ == '__main__':
    app.run()

数据迁移

  • 在开发过程中，需要修改数据库模型，而且还要在修改之后更新数据库。最直接的方式就是删除旧表，但这样会丢失数据，所以往往更常见的方式就是使用alter来改变数据结构，原有数据中的新字段值设置默认值或null=True.
  • 更好的解决办法是使用数据迁移，它可以追踪数据库表结构的变化，然后把变动应用到数据库中。
  • 在Flask中可以使用Flask-Migrate的第三方扩展，来实现数据迁移。并且集成到Flask终端脚本中，所有操作通过flask db 命令就能完成。
  • 为了导出数据库迁移命令，Flask-Migrate提供了一个MigrateCommand类，可以附加到flask框架中。

首先要在虚拟环境中安装Flask-Migrate。

pip install Flask-Migrate

官网地址：https://flask-migrate.readthedocs.io/en/latest/

代码文件内容：

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
# 导入数据迁移核心类
from flask_migrate import Migrate

app = Flask(__name__)

class Config(object):
    DEBUG = True
    # 数据库连接配置
    # SQLALCHEMY_DATABASE_URI = "数据库类型://数据库账号:密码@数据库地址:端口/数据库名称?charset=utf8mb4"
    SQLALCHEMY_DATABASE_URI = "mysql://root:123@127.0.0.1:3306/flaskdemo?charset=utf8mb4"
    # 动态追踪修改设置，如未设置只会提示警告
    SQLALCHEMY_TRACK_MODIFICATIONS = True
    # 查询时会显示原始SQL语句
    SQLALCHEMY_ECHO = True

app.config.from_object(Config)

db = SQLAlchemy(app=app)

# 初始化数据迁移
migrate = Migrate(app, db)

# migrate = Migrate()
# migrate.init_app(app, db)

"""模型类定义"""
# 关系表的声明方式
achieve = db.Table('tb_achievement',
    db.Column('student_id', db.Integer, db.ForeignKey('tb_student.id')),
    db.Column('course_id', db.Integer, db.ForeignKey('tb_course.id'))
)

class Course(db.Model):
    # 定义表名
    __tablename__ = 'tb_course'
    # 定义字段对象
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True)
    price = db.Column(db.Numeric(6,2))
    teacher_id = db.Column(db.Integer, db.ForeignKey('tb_teacher.id'))
    students = db.relationship('Student', secondary=achieve, backref='courses', lazy='subquery')
    # repr()方法类似于django的__str__，用于打印模型对象时显示的字符串信息
    def __repr__(self):
        return 'Course:%s'% self.name

class Student(db.Model):
    __tablename__ = 'tb_student'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True)
    email = db.Column(db.String(64),unique=True)
    age = db.Column(db.SmallInteger,nullable=False)
    sex = db.Column(db.Boolean,default=1)

    def __repr__(self):
        return 'Student:%s' % self.name

class Teacher(db.Model):
    __tablename__ = 'tb_teacher'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True)
    # 课程与老师之间的多对一关联
    courses = db.relationship('Course', backref='teacher', lazy='subquery')

    def __repr__(self):
        return 'Teacher:%s' % self.name

@app.route("/")
def index():
    return "ok"

if __name__ == '__main__':
    app.run(debug=True)

创建迁移版本仓库

# 切换到项目根目录下
cd ~/Desktop/flaskdemo
# 设置flask项目的启动脚本位置，例如我们现在的脚本叫manage.py
export FLASK_APP=manage.py
# 数据库迁移初始化，这个命令会在当前项目根目录下创建migrations文件夹，将来所有数据表相关的迁移文件都放在里面。
flask db init

创建迁移版本

  • 自动创建迁移版本文件中有两个函数，用于进行数据迁移同步到数据库操作的。
    • upgrade()：把迁移中的改动代码同步到数据库中。
    • downgrade()：则将改动代码从数据库中进行还原。
  • 自动创建的迁移脚本会根据模型定义和数据库当前状态的差异，生成upgrade()和downgrade()函数的内容。
  • 生成的迁移文件不一定完全正确，有可能代码中存在细节遗漏导致报错，需要开发者进行检查，特别在多对多的时候
# 根据flask项目的模型生成迁移文件 -m的后面你不要使用中文！！
flask db migrate -m 'initial migration'
# 这里等同于django里面的 makemigrations，生成迁移版本文件
# 完成2件事情：
# 1. 在migrations/versions生成一个数据库迁移文件
# 2. 如果是首次生成迁移文件的项目，则迁移工具还会在数据库创建一个记录数据库版本的version表

升级版本库的版本

把当前ORM模型中的代码改动同步到数据库。

# 从migations目录下的versions中根据迁移文件upgrade方法把数据表的结构同步到数据库中。
flask db upgrade

降级版本库的版本

# 从migations目录下的versions中根据迁移文件downgrade把数据表的结构同步到数据库中。
flask db downgrade

版本库的历史管理

可以根据history命令找到版本号,然后传给downgrade命令:

flask db history

输出格式：<base> ->  版本号 (head), initial migration

回滚到指定版本

flask db downgrade # 默认返回上一个版本
flask db downgrade 版本号   # 回滚到指定版本号对应的版本
flask db upgrade 版本号     # 升级到指定版本号对应的版本

数据迁移的步骤：

1. 初始化数据迁移的目录
export FLASK_APP=4-manage.py
flask db init

2. 数据库的数据迁移版本初始化，生成迁移文件
flask db migrate -m 'initial migration'

3. 升级版本[新增一个迁移记录]
flask db upgrade 

4. 降级版本[回滚一个迁移记录]
flask db downgrade
阅读全部

相关推荐

  1. 【Flask从入门到精通：第十一课：逻辑外键、数据迁移】

    2024-07-19 01:34:02       19 阅读

  2. 【Flask从入门到精通：第十课：数据基本查询、关联查询】

    2024-07-19 01:34:02       21 阅读

  4. 【Flask从入门到精通：第八课：ORM、Flask-SQLAlchemy】

    2024-07-19 01:34:02       23 阅读

  5. 【Flask从入门到精通：第三课：http的请求与响应】

    2024-07-19 01:34:02       22 阅读

  7. 嵌入式——从入门到精通第十一天

    2024-07-19 01:34:02       49 阅读

最近更新

  3. docker php8.1+nginx base 镜像 dockerfile 配置

    2024-07-19 01:34:02       67 阅读

  4. Could not load dynamic library ‘cudart64_100.dll‘

    2024-07-19 01:34:02       72 阅读

  9. 在Django里面运行非项目文件

    2024-07-19 01:34:02       58 阅读

  19. Python语言-面向对象

    2024-07-19 01:34:02       69 阅读

  20. 如何分清楚常见的 Git 分支管理策略Git Flow、GitHub Flow 和 GitLab Flow

    2024-07-19 01:34:02       64 阅读

热门阅读

  1. ChatGPT：Stream 和 数据源

    2024-07-19 01:34:02       18 阅读

  3. 1.虚拟机相关的博文推荐

    2024-07-19 01:34:02       19 阅读
  4. Flink HA

    Flink HA

    2024-07-19 01:34:02      20 阅读

  6. vault正式环境安装部署

    2024-07-19 01:34:02       23 阅读

  7. 【Git】Git解除仓库关联或关联新仓库

    2024-07-19 01:34:02       18 阅读

  10. AIGC笔记--Classifer Guidance的代码理解

    2024-07-19 01:34:02       25 阅读

  12. rust 构建自己的库和模块

    2024-07-19 01:34:02       20 阅读

  13. 大语言模型系列-Transformer

    2024-07-19 01:34:02       24 阅读

  14. Git入门

    2024-07-19 01:34:02       25 阅读

  17. g++: internal compiler error: Killed (program cc1plus) 解决办法

    2024-07-19 01:34:02       20 阅读

  18. JVM高频面试题

    2024-07-19 01:34:02       23 阅读

  21. AIGC高频产品面试题（专有名词科普及其在AI领域中的应用）

    2024-07-19 01:34:02       25 阅读

  23. 割点（Articulation Point）

    2024-07-19 01:34:02       24 阅读

  24. [C/C++入门][变量和运算]4、带余除法

    2024-07-19 01:34:02       20 阅读