Flask-SQLAlchemy 的基本使用

安装

python -m pip install flask
python -m pip install flask-sqlalchemy
python -m pip install pymysql

建议直接安装当前稳定版（本笔记基准：Flask 3.1.3 + Flask-SQLAlchemy 3.x）。

连接数据库

连接mysql数据库

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import text

app = Flask(__name__)

DB_HOST = "192.168.1.112"
DB_PORT = 3306
DB_USER = "root"
DB_PASS = "123456"
DB_NAME = "flask"

app.config['SQLALCHEMY_DATABASE_URI'] = f"mysql+pymysql://{DB_USER}:{DB_PASS}@{DB_HOST}:{DB_PORT}/{DB_NAME}"

db = SQLAlchemy(app)

# 低版本flask不需要写 app_context
with app.app_context():
    with db.engine.connect() as conn:
        # 低版本sqlalchemy不需要 text
        rs = conn.execute(text("select 1"))
        print(rs.fetchone())

运行输出结果

(1,)

ORM模型

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import text

app = Flask(__name__)

DB_HOST = "192.168.1.112"
DB_PORT = 3306
DB_USER = "root"
DB_PASS = "123456"
DB_NAME = "flask"

app.config['SQLALCHEMY_DATABASE_URI'] = f"mysql+pymysql://{DB_USER}:{DB_PASS}@{DB_HOST}:{DB_PORT}/{DB_NAME}"

db = SQLAlchemy(app)

class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    username = db.Column(db.String(80), nullable=False)
    password = db.Column(db.String(80), nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)

    def __init__(self, username, password, email):
        self.username = username
        self.password = password
        self.email = email

    def __repr__(self) -> str:
        return f"<User {self.id}>"

with app.app_context():
    # 将User Model映射为数据表
    db.create_all()

Flask-SQLAlchemy还支持以下字段类型

类型	说明
`db.Integer`	整型
`db.SmallInteger`	短整型
`db.BigInteger`	长整型
`db.Decimal`	定点类型，可以指定小数位
`db.Boolean`	布尔类型
`db.Date`	日期类型。存储`datetime.date`对象
`db.DateTime`	日期时间类型。存储`datetime.datetime`对象
`db.Time`	时间类型。存储Python的`datetime.time`对象
`db.Interval`	时间间隔。存储Python的`datetime.timedelay`对象
`db.String`	字符串类型
`db.Text`	文本类型
`db.Enum`	枚举类型
`db.PickleType`	存储经过pickle的对象
`db.LargeBinary`	存储二进制数据

CRUD操作

一个简单的用户增删改查服务

from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy
import json
# from sqlalchemy import text

app = Flask(__name__)

DB_HOST = "192.168.1.112"
DB_PORT = 3306
DB_USER = "root"
DB_PASS = "123456"
DB_NAME = "flask"

app.config['SQLALCHEMY_DATABASE_URI'] = f"mysql+pymysql://{DB_USER}:{DB_PASS}@{DB_HOST}:{DB_PORT}/{DB_NAME}"

db = SQLAlchemy(app)

class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    username = db.Column(db.String(80), nullable=False)
    password = db.Column(db.String(80), nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)

    def __init__(self, username:str, password:str, email:str):
        self.username = username
        self.password = password
        self.email = email

@app.get("/")
def root():
    return "Hello World"

@app.post("/moduser")
def mod_user():
    # 根据查询参数获取 user id
    userId = request.args.get('userId')
    if userId is None:
        return jsonify({
            "code": 400,
            "msg": "incomplete arguments",
        })
    
    req_body = request.get_data()
    req_body = json.loads(req_body)
    if req_body is None:
        return jsonify({
            "code": 400,
            "msg": "incomplete arguments",
        })

    user = User.query.filter_by(id=userId).first()
    if user is None:
        return jsonify({
            "code": 400,
            "msg": "user not found",
        })
    
    user.username = req_body.get('username')
    db.session.commit()
    resp = {
        "code": 200,
        "msg": "success",
    }
    return jsonify(resp)

@app.post('/deluser')
def del_user():
    userId = request.args.get('userId')
    if userId is None:
        return jsonify({
            "code": 400,
            "msg": "incomplete arguments",
        })
    user = User.query.filter_by(id=userId).first()
    if user is None:
        return jsonify({
            "code": 400,
            "msg": "user not found",
        })
    else:
        db.session.delete(user)
        db.session.commit()
        resp = {
            "code": 200,
            "msg": "success",
        }
        return jsonify(resp)

@app.post('/adduser')
def add_user():
    req_data = request.get_data()
    req_data = json.loads(req_data)
    req_username = req_data.get('username')
    req_password = req_data.get('password')
    req_email = req_data.get('email')
    if not req_username or not req_password or not req_email:
        return jsonify({
            "code": 400,
            "msg": "incomplete arguments",
        })
    
    select_email = User.query.filter_by(email = req_email).first()
    if select_email is not None:
        return jsonify({
            "code": 400,
            "msg": "The email address has been registered",
        })

    user = User(username=req_username, password=req_password, email=req_email)
    db.session.add(user)
    db.session.commit()

    rst = User.query.filter_by(email=req_email).first()

    resp = {
        "code": 200,
        "msg": "success",
        "data": {
            "userId": rst.id,
        }
    }
    # print(resp)
    return jsonify(resp)

# 查询用户
@app.post("/user")
def get_user():
    userId = request.args.get('userId')
    if userId is None:
        return jsonify({
            "code": 400,
            "msg": "incomplete arguments",
        })
    user = User.query.filter_by(id=userId).first()
    if user is None:
        return jsonify({
            "code": 400,
            "msg": "user not found",
        })
    else:
        resp = {
            "code": 200,
            "msg": "success",
            "data": {
                "username": user.username,
            }
        }
        return jsonify(resp)

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

使用pytest测试

import requests
import json
import pytest

headers = {
    "Content-Type": "application/json"
}


def test_moduser():
    url = "http://127.0.0.1:5000/moduser?userId=13"
    payload = {
        "username": "lisi"
    }
    resp = requests.post(url,data=json.dumps(payload),headers=headers)
    print(resp.json())
    assert resp.json()["code"] == 200

def test_deluser():
    url = "http://127.0.0.1:5000/deluser?userId=13"
    resp = requests.post(url,headers=headers)
    print(resp.json())
    assert resp.json()["code"] == 200

def test_adduser():
    url = "http://127.0.0.1:5000/adduser"

    payload = {
        "username": "test",
        "password": "123456",
        "email": "test15@demo.com"
    }
    resp = requests.post(url,data=json.dumps(payload),headers=headers)
    print(resp.json())
    assert resp.json()["code"] == 200

def test_getuser():
    url = "http://127.0.0.1:5000/user?userId=14"
    resp = requests.post(url,headers=headers)
    print(resp.json())
    assert resp.json()["code"] == 200

if __name__ == "__main__":
    pytest.main(["-v","-s","test_demo.py"])

增加数据

增加数据需要创建对象，创建对象时必须通过关键字给字段赋值，尤其是那些Not Null的字段

# 创建对象
user1 = User(username="zhangda",password="111")
user2 = User(username="zhanger",password="222")
user3 = User(username="zhangsan",password="333")

# 可以先把对象都添加到session中再统一commit
db.session.add(user1)
db.session.add(user2)
db.session.add(user3)

db.session.commit()

删除数据

删除单条数据（SQLAlchemy 2.0 推荐 db.session.get）

user = db.session.get(User, 1)
db.session.delete(user)
db.session.commit()

删除多条数据

User.query.filter(User.username.contains("zhangsan")).delete(synchronize_session=False)
db.session.commit()

更新数据

修改一条数据

user = db.session.get(User, 1)
user.username = "zhangsi"
db.session.commit()

修改多条数据

User.query.filter(User.username.like("%zhangsan%")).update({"password": "qazxsw"}, synchronize_session=False)
db.session.commit()

查询数据

# 获取查询结果集中的所有对象, 是列表类型
query.all()

# 查询结果集的第一个对象
query.first()

# 获取查询结果集中的一个对象
query.one()

# 类似query.one(), 不过对象数量不等于1时不抛出异常
query.one_or_none()

# 根据主键获取数据（SQLAlchemy 2.0 推荐写法，旧版 query.get 已移除）
db.session.get(User, pk)

# 判断数据是否存在
db.session.query(User).filter(User.username == "zhangsan").exists()

# 获取结果集的个数
query.count()

# 根据查询条件过滤
users = User.query.filter(User.username=="zhangsan").all()

# 根据关键字参数过滤
users = User.query.filter_by(username="zhangsan").all()

# 对结果进行切片操作
query.slice(start,stop)

# limit限制
query.limit(limit)

# offset
query.offset(offset)

# 排序
query.order_by()

# 分组
query.group_by()

# like模糊查询
users = User.query.filter(User.username.like("%zhangsan%"))

# in
users =User.query.filter(User.username.in_(["zhangsan","lisi","wangwu"]))

# not in
users = User.query.filter(~User.username.in_(["zhangsan","lisi"]))

# is null
users = User.query.filter(User.username==None)

# is not null
users = User.query.filter(User.username != None)

# and
from sqlalchemy import and_
users = User.query.filter(and_(User.username == "zhangsan", User.email.like("%@demo.com")))

# or
from sqlalchemy import or_
users = User.query.filter(or_(User.username == "zhangsan", User.email == "lisi@demo.com"))

安装​

连接数据库​

ORM模型​

CRUD操作​

增加数据​

删除数据​

更新数据​

查询数据​

安装

连接数据库

ORM模型

CRUD操作

增加数据

删除数据

更新数据

查询数据