flask-sqlalchemy操作mysql_flask sqlalchemy mysql

1、简介

SQLAlchemy是Python编程语言下的一款开源软件。提供了SQL工具包及对象关系映射（ORM）工具。
而Flask-SQLAlchemy是Flask对SQLAlchemy进行了一个封装，目的于简化在Flask中SQLAlchemy的使用，
提供了有用的默认值和额外的助手来更简单地完成日常任务。

orm作用
	1.省去自己拼写SQL，保证SQL语法的正确性一次编写可以适配多个数据库.
	2.防止注入攻击.
	3.在数据库表名或字段名发生变化时，只需修改模型类的映射，无需修改数据库操作的代码.
1
2
3
4
5
6
7
8

#安装Flask-SQLAlchemy和MySQL的Python客户端库以及数据库迁移工具
pip install flask-sqlalchemy
pip install mysqlclient
pip install flask_script
pip install flask_migrate ---2.7.0
1
2
3
4
5

2、mysql连接设置

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_script import Manager
from flask_migrate import Migrate,MigrateCommand#MigrateCommand在flask_migrate3.0以上已经去掉了
import pymysql
pymysql.install_as_MySQLdb()

#配置对象
class DatabaseConfig():
	#连接的数据库 URI 
	#mysql://用户名:密码@host:port/database
    SQLALCHEMY_DATABASE_URI = 
    'mysql://root:123456@127.0.0.1:3306/sqlalchemyTest'
    #是否追踪数据的修改
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    #是否显示生成sql语句
    SQLALCHEMY_ECHO = False

#实例化一个flask对象
app = Flask(__name__)
#实例化一个Manager对象，flask脚本管理对象
manager = Manager(app)

#从配置对象中加载
app.config.from_object(DatabaseConfig)

#创建SQLAlchemy对象
db  = SQLAlchemy(app)
Migrate(app,db)
#将db添加进manager中
manager.add_command("db",MigrateCommand)

#或者下面这种方法创建sqlalchemy对象,但是单独运行调试时，对数据库操作需要在Flask的应用上下文中进行
#db = SQLAlchemy()
#db.init_app(app)
#with app.app_context():
#      database.query.all()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

3、模型类字段类型

4、ORM使用方式

4.1、先创建模型类，再迁移到数据库

优点：简单快捷，不用写sql
缺点：表结构发生变化的时候，容易发生迁移错误
1
2

sql_test.py

from main import db
from main import manager

class Student(db.Model):
	#设置数据库中真实表名
    __tablename__ = "student"
    
    sid = db.Column("sid",db.Integer,primary_key=True,doc="学生id")
    sname = db.Column("sname",db.String(4),nullable=False,doc="姓名")
    gender = db.Column(db.Enum("男","女"),doc="性别")
    phone = db.Column(db.String(11),nullable=False,doc="手机号")
    #第一种关联表：relationship关联表，backref添加反向引用，通过Grade对象.stu获得student表信息
 1. grades = db.relationship("Grade",backref="stu")
    #第二种关联方式：primaryjoin
 2. grades =db.relationship("Grade",primaryjoin="Student.sid==foreign(Grade.student)",backref="stu")
    #多对多关系中中间表的数据库名字，可通过student对象.courses获取课程表信息，backref同上述一样。
    courses = db.relationship("Course",secondary="student_course",backref="stu")
	#将对象转化为字符串
    def __repr__(self):
        sname = self.sname
        phone = self.phone
        return f"{sname}:{phone}"

class StudentCourse(db.Model):
    id = db.Column(db.Integer,primary_key=True,doc="学生课程id")
    sid = db.Column(db.Integer,db.ForeignKey("student.sid"),doc="学生id")
    cid = db.Column(db.Integer,db.ForeignKey("course.cid"),doc="课程id")

class Course(db.Model):
    __tablename__ = "course"

    cid = db.Column("cid",db.Integer,primary_key=True,doc="课程id")
    cname = db.Column("cname",db.String(4),nullable=False,doc="课程名")
    teacher = db.Column("teacher",db.Integer,db.ForeignKey("teacher.tid"),doc="老师id")
    grade = db.relationship("Grade",backref="cour")

class Teacher(db.Model):
    __tablename__ = "teacher"

    tid = db.Column(db.Integer,primary_key=True,doc = "老师id")
    tname = db.Column(db.String(4),nullable=False,doc="老师姓名")
    gender = db.Column(db.Enum("男", "女"), doc="性别")
    phone = db.Column(db.String(11), nullable=False, doc="手机号")
    # course = db.relationship("Course",backref="tea")
    # uselist如果为False，不使用列表，而使用标量值,使用在一对一表中
    course = db.relationship("Course",primaryjoin="Teacher.tid==foreign(Course.teacher)", backref="tea",uselist=False)

class Grade(db.Model):
    __tablename__ = "grade"

    id = db.Column(db.Integer,primary_key=True,doc="分数id")
    score = db.Column(db.Float,default=0,doc="分数")
    student = db.Column("student",db.Integer,db.ForeignKey("student.sid"),doc="学生id")
    course = db.Column("course",db.Integer,db.ForeignKey("course.cid"),doc="课程id")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

if __name__ == '__main__':
#第一种更新和创建数据库，这种会清空数据数据
    db.create_all() #创建表
    #db.drop_all()   #删除表
    #第二种方法执行命令
    #manager.run()
# 第二种利用flask_migrate数据库迁移，终端上，当前文件目录下执行:
	# python sql_test.py db init  创建migrations文件夹，所有迁移文件放在这文件夹内
	#python sql_test.py db migrate 生成迁移脚本，类似django的makemigrations
	#python sql_test.py db upgrade 更新数据库  , 类似django的migrate
1
2
3
4
5
6
7
8
9
10

4.2、用原生SQL创建数据库表，再编写模型类作映射

优点：可以很好的控制数据库表结构，避免发生迁移错误
缺点：需要编写sql语句
1
2

先创建数据库表，然后编写模型类

create table student(
    id int primary key auto_increment comment "学生id",
    sname varchar(4) not null comment "姓名",
    gender enum("男","女") default "男" comment "性别",
    phone char(11) not null comment "手机号"
) engine="innodb",charset="utf8",comment="学生表";

create table teacher(
    id int primary key auto_increment comment "老师id",
    tname varchar(4) not null comment "老师姓名",
    gender enum("男","女") default "男" comment "性别",
    phone char(11) not null comment "手机号"
) engine="innodb" default charset=utf8 comment="教师表";

create table course(
    id int primary  key auto_increment comment "课程id",
    cname varchar(10) not null comment "课程名",
    teacher int default null comment "老师id",
    constraint c_t_key foreign key(teacher) references teacher(id)
) engine="innodb" default charset=utf8 comment="课程表";

create table grade(
    id int primary key auto_increment comment "分数id",
    score float default 0 comment "分数",
    student int default null comment "学生id",
    course int default null comment "课程id",
    constraint g_s_key foreign key(student) references student(id),
    constraint g_c_key foreign key(course) references course(id)
) engine="innodb" default charset=utf8 comment "分数表";

create table student_course(
    id int primary  key auto_increment comment "学生课程id",
    sid int default null comment "学生id",
    cid int default null comment "课程id",
    constraint s_s_c foreign key(sid) references student(id),
    constraint c_s_c foreign key(cid) references course(id)
) engine="innodb" default charset=utf8 comment="学生课程表";
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

5、CRUD

5.1、增

# 单个新增
student = Student(sname = "张三",gender="男",phone="19971251765")
db.session.add(student)
db.session.commit()
# 通过add_all([])批量新增
s1 = Student(sname = "李四",gender="男",phone="19971251766")
s2 = Student(sname = "王五",gender="男",phone="19971251767")
s3 = Student(sname = "赵六",gender="女",phone="19971251768")
s4 = Student(sname = "刘七",gender="男",phone="19971251769")
db.session.add_all([s1,s2,s3,s4])
db.session.commit()
1
2
3
4
5
6
7
8
9
10
11

5.2、查

常规查询

#返回所有student对象
s = Student.query.all()
#根据主键ID获取对象，若主键不存在返回None
s = Student.query.get(1)
# 返回第一个学生对象
s = Student.query.first()
#条件查询  filter 和 filter_by
s = Student.query.filter_by(gender="男").first()
ss = Student.query.filter_by(gender="男").all()
s = Student.query.filter(Student.gender=="女").first()
1
2
3
4
5
6
7
8
9
10

逻辑查询

# 逻辑或查询
from sqlalchemy import or_
User.query.filter(or_(Student.sname=='张三', Student.age==20)).all()
1
2
3

# 逻辑与查询
from sqlalchemy import and_
User.query.filter(and_(Student.sname=='张三', Student.age==20)).all()
1
2
3

# 逻辑非查询
from sqlalchemy import not_
User.query.filter(not_(Student.sname=='张三', Student.age==20)).all()
1
2
3

offset 偏移查询

Student.query.offset(2).all()
1

limit

s = Student.query.limit(3).all()
1

order_by

s = Student.query.order_by(Student.sid).all()  # 正序
s = Student.query.order_by(Student.sid.desc()).all()  # 倒序
1
2

load_only 指定字段查询

from sqlalchemy.orm import load_only
User.query.options(load_only(Student.sname, Student.age)).filter_by(sid=1).first() # 查询特定字段
1
2

分组聚合

from sqlalchemy import func
#  对分数表的学生id进行分组, 求学生个数
db.session.query(Grade.student, func.count(Grade.student)).filter(Grade.score>=60).group_by(Grade.student).all()
1
2
3

复合查询

#查询学生年龄大于20岁，并按照id降序排列，从第二个开始，查出5个
Student.query.filter(Student.age > 20).order_by(Student.sid.desc()).offset(2).limit(5).all()
1
2

多表关联查询

# 一（课程表）对一（教师）
class Course(db.Model):
    teacher = db.Column("teacher",db.Integer,db.ForeignKey("teacher.tid"),doc="老师id")

class Teacher(db.Model):
    course = db.relationship("Course",backref="tea"，uselist=False)

#通过课程表查询老师信息
c = Course.query.first()
t = c.tea  # tea 由backref指定
#通过老师表查询课程信息
t = Teacher.query.first()
c = t.course
1
2
3
4
5
6
7
8
9
10
11
12
13

# 一（学生）对多（分数）
class Student(db.Model):
    grades = db.relationship("Grade",backref="stu")
    
class Grade(db.Model):
    student = db.Column("student",db.Integer,db.ForeignKey("student.sid"),doc="学生id")
#通过分数查学生
g = Grade.query.first()
s = g.stu
#通过学生查分数
s = Student.query.first()
g = s.grades(列表)
1
2
3
4
5
6
7
8
9
10
11
12

# 多（学生）对多（课程）
class Student(db.Model):
    courses = db.relationship("Course",secondary="student_course",backref="stu")
#中间表
class StudentCourse(db.Model):
    sid = db.Column(db.Integer,db.ForeignKey("student.sid"),doc="学生id")
    cid = db.Column(db.Integer,db.ForeignKey("course.cid"),doc="课程id")

class Course(db.Model):
    ... ...

 s = Student.query.first()
 cs = s.courses  (列表)
 for c in cs:
     for stu in c.stu:
         print(stu.sname)
 t = Teacher.query.first()
 c = t.course
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

关联字段联合查询

# mysql 连表查询
from sqlalchemy.orm import load_only,contains_eager

s = Teacher.query.join(Teacher.course).options(load_only(Teacher.tname),contains_eager(Teacher.course).load_only(Course.cname)).first()
print(s.tname,s.course.cname)
1
2
3
4
5

5.3、改

# 第一种
c = Student.query.filter(Student.sname=="张三").update({"sname":"章三"})
db.session.commit()
#第二种
stu = Student.query.filter_by(sname="章三").first()
stu.sname = "张三"
db.session.add(stu)
db.session.commit()
1
2
3
4
5
6
7
8

5.4、删

 Student.query.filter(Student.sid == 1).delete()
  db.session.commit()
1
2

6、事务

try:
    teacher = Teacher(tname='老王',gender="男",phone="12345678910")
    db.session.add(teacher)
    db.session.flush() # 将db.session记录的sql传到数据库中执行
    course = Course(cname="数学",teacher=teacher.tid)
    db.session.add(course)
    db.session.commit()
except:
    db.session.rollback()
1
2
3
4
5
6
7
8
9

7、快速搭建

main.py

from flask import Flask, request, jsonify
import config
from dao import db
from models import User


app = Flask(__name__)
app.config.from_object(config)

#  db.create_all()  # 这个是干啥的？
db.init_app(app)

@app.route('/')
def index():
    return '这是首页'

@app.route("/api/login",methods=["POST"])
def login():
    data = request.get_json()
    username = data.get("username")
    password = data.get('password')
    try:
        user = User.query.filter_by(username=username,password=password).first()
        if user:
            return jsonify(code=200,msg="登录成功")
        else:
            return jsonify(code=400,msg="用户名或密码错误")
    except Exception as e:
        return jsonify(code=400,msg=str(e))

if __name__ == '__main__':
    app.run(debug=True,host="0.0.0.0",port=5000)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

manager.py

from flask_script import Manager
from flask_migrate import Migrate, MigrateCommand
from main import app
from dao import db

from models import User

manage = Manager(app)

Migrate(app, db)

manage.add_command('db', MigrateCommand)


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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

dao.py

from flask_sqlalchemy import SQLAlchemy
#from flask_app import app

db = SQLAlchemy()

1
2
3
4
5
6

models.py

from exts import db
class User(db.Model):
    #设置数据库中真实表名
    __tablename__ = "user"
    
    id = db.Column("sid",db.Integer,primary_key=True,doc="用户id")
    username = db.Column("username",db.String(8),nullable=False,doc="用户名")
    password = db.Column("password",db.String(8),nullable=False,doc="密码")
    phone = db.Column(db.String(11),nullable=True,doc="手机号")
1
2
3
4
5
6
7
8
9
10
11

config.py

HOSTNAME = '192.168.153.20'
DATABASE = 'c00580'
PORT = 3306
USERNAME = 'root'
PASSWORD = '123456'
DB_URL = 'mysql+pymysql://{}:{}@{}:{}/{}'.format(USERNAME, PASSWORD, HOSTNAME, PORT, DATABASE)

SQLALCHEMY_DATABASE_URI = DB_URL
SQLALCHEMY_TRACK_MODIFICATIONS = False

1
2
3
4
5
6
7
8
9
10

数据迁移

# 创建数据库
CREATE DATABASE `c00580` DEFAULT CHARACTER SET utf8 COLLATE utf8_general_ci;

# db init
python manage.py db init
# db migrate
python manage.py db migrate
# db upgrade
python manage.py db upgrade

# FLASK_APP
export FLASK_APP=main

flask run -h 192.168.153.20
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

依赖包

alembic==1.13.0
aniso8601==9.0.1
anyio==3.5.0
asgiref==3.7.2
beautifulsoup4==4.12.2
blinker==1.7.0
certifi==2024.2.2
charset-normalizer==3.3.2
click==8.1.7
Django==3.2.12
Flask==3.0.2
Flask-Cors==4.0.0
Flask-Login==0.6.3
Flask-Migrate==2.7.0
Flask-RESTful==0.3.10
Flask-Script==2.0.6
Flask-SQLAlchemy==3.1.1
flask-swagger-ui==4.11.1
greenlet==3.0.3
idna==3.6
itsdangerous==2.1.2
Jinja2==3.1.3
lxml==4.9.3
Mako==1.3.0
MarkupSafe==2.1.5
mysqlclient==2.0.3
pandoc==2.3
pip==23.3.1
plumbum==1.8.2
ply==3.11
PyMySQL==1.1.0
python-docx==1.1.0
pytz==2024.1
requests==2.31.0
setuptools==68.2.2
six==1.16.0
sniffio==1.3.1
soupsieve==2.5
SQLAlchemy==2.0.27
sqlparse==0.4.4
typing_extensions==4.10.0
urllib3==2.1.0
Werkzeug==3.0.1
wheel==0.41.2

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46