flask-sqlalchemy库

彩笔激流勇退。

1. 简介

ORM，对象关系映射。简单来说，ORM将数据库中的表与面向对象中的类建立了一种对应关系。这样，我们要操作数据库，表，记录就可以直接通过操作类或者类实例来完成。

SQLAlchemy 是目前python中最垃圾的 ORM框架, 功能全面, 使用复杂。

Flask-SQLAlchemy 是一个为 Flask 应用增加 SQLAlchemy 支持的扩展，把原本pymysql几句话就能搞定的事情整成抽象的，继承的，封装的，多态的更适合高级程序员体质的负离子保温杯。

抛开兼容性不谈，Flask-SQLAlchemy无疑是磨练程序员改bug的磨刀石，是检验程序员记忆能力的试金石，是凝结了人类精华的草酸钙结石。

常用字段类型

常用的字段选项

如果没有给对应字段的类属性设置default参数, 且添加数据时也没有给该字段赋值, 则sqlalchemy会给该字段设置默认值 None。

常见命令

db.create_all() #创建所有表
db.drop_all() #删除所有表

2. 创建表

pip install pymysql

pip install flask-sqlalchemy

数据库URL(连接地址)格式: 协议名://用户名:密码@数据库IP:端口号/数据库名

main.py

​ 在下面代码中，我们使用了 with app.app_context(): 语句来确保当前应用实例的操作db.create_all() 是在flask应用上下文中被调用的。

from app import *
from models import User

@app.route('/',methods=['GET','POST'])
def login():
    print(db)
    return 'hello world'

if __name__ == '__main__':
    with app.app_context():
        db.create_all()#创建表
    app.run(host='0.0.0.0',port=9901,debug=1)

modules.py

表名默认为类名小写, 可以通过 __tablename__类属性 进行修改

from app import db

class User(db.Model):# User表
    __tablename__ = 't_user'
    id= db.Column(db.Integer,primary_key=True) # 必须要有主键存在
    name=db.Column(db.String(20),nullable=True)# 可空
    age=db.Column(db.SmallInteger)
    gender=db.Column(db.Boolean)
    birthday=db.Column(db.Date)

对应MySQL语句

CREATE TABLE `t_user` (
  `id` int NOT NULL AUTO_INCREMENT,
  `name` varchar(20) COLLATE utf8mb4_general_ci DEFAULT NULL,
  `age` smallint DEFAULT NULL,
  `gender` tinyint(1) DEFAULT NULL,
  `birthday` date DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;

app.py

from flask import Flask,url_for,request,render_template,make_response,redirect,jsonify
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__) # 用本脚本名实例化Flask对象
# 设置数据库连接地址
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://root:123456@127.0.0.1:3306/test1'

# 是否追踪数据库修改(开启后会触发一些钩子函数)  一般不开启, 会影响性能
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

# 是否显示底层执行的SQL语句
app.config['SQLALCHEMY_ECHO'] = True

# 初始化组件对象, 直接关联Flask应用
db = SQLAlchemy(app)

3. 数据表简单查询

user=User.query

if __name__ == '__main__':
    with app.app_context():
        db.create_all()
        user=User.query
        print(user)#显示sql语句, 返回Query对象
        print(user.all())#[<User 1>, <User 2>, <User 3>],每个元素都是models.User类型
        print(user.count()) #返回query中的Model对象数量
        print(user.filter_by(age=30,id=1)) #显示SQL语句, 返回Query对象，内部条件为交集

4. 映射查询 db.session.query

映射查询在SQLAlchemy中，可以通过session对象的query方法完成。

注意关键字书写顺序

db.session.query().filter().group_by().having().order_by().paginate().all()

if __name__ == '__main__':
    with app.app_context():
        db.create_all()
        query=db.session.query(User) #<class 'flask_sqlalchemy.query.Query'>
        print(query.filter(User.id>1).all())# [<User 2>, <User 3>]
        query2=db.session.query(User.id,User.name)
        print(query2.filter(User.age==99).all())# [(2, 'tom')]
    app.run(host='0.0.0.0',port=9901,debug=1)

5. 排序 order_by

from sqlalchemy import desc

if __name__ == '__main__':
    with app.app_context():
        db.create_all()
        query = db.session.query(User.id, User.name,User.age)
        print(query.order_by(User.age).all())  # 默认升序排序,asc()
        print(query.order_by(desc(User.age)).all())  # 降序排序
        print(query.order_by(User.age,User.id).all())  # 先排age，后排id
    app.run(host='0.0.0.0', port=9901, debug=1)# 看起来是一个阻塞函数

6. 聚合函数

数据库先添加一个age为30的记录。

from sqlalchemy import func

if __name__ == '__main__':
    with app.app_context():
        db.create_all()
        query = db.session.query(func.max(User.age),func.avg(User.age))
            #相当于 SELECT max(age) , avg(age) FROM t_user
        print(query.all())  # [(99, Decimal('51.0000'))]
    app.run(host='0.0.0.0', port=9901, debug=1)

7. 分组查询 group_by

if __name__ == '__main__':
    with app.app_context():
        db.create_all()
        query = db.session.query(func.count(User.id))
        print(query.group_by(User.age).all())  
        # [(2,), (1,), (1,)]
    app.run(host='0.0.0.0', port=9901, debug=1)

8. 增删改

flask开了debug模式，删除数据会导致main函数重新执行，给爷整笑了。

if __name__ == '__main__':
    with app.app_context():
        db.create_all()

        #更新
        u = db.session.query(User.id==1)# 查询主键为1的记录
        u.name="Jack"
        #db.session.rollback() 事务回滚，默认遇到错误自动回滚
        db.session.commit()# 事务提交

        #删除
        u2=db.session.query(User).filter(User.id==6).all() # User 模型的实例
        if len(u2)!=0:
            db.session.delete(u2[0])
            db.session.commit()  # 事务提交

        #增加
        u3=User(id=7,name="lihua",age=35,gender=1,birthday='2077-1-1')
        db.session.add(u3)
        #db.session.add_all([u1,u2,u3]) 一次添加多个
        db.session.commit()

    app.run(host='0.0.0.0', port=9901, debug=0)
    # debug=1时，上面的delete操作会执行多次
    # 大概是是检测到了文件变化重启了一次main函数???

9. 分页查询 paginate

分页查询不老老实实用limit，非要整个paginate装什么高大上。

if __name__ == '__main__':
    with app.app_context():
        db.create_all()
        pg=db.session.query(User).paginate(page=2,per_page=2)# QueryPagination object
        print(pg.items)#[<User 3>, <User 4>],当前页数据
        print(pg.pages) #3 ,一共三页
        for i in pg.iter_pages(): #迭代Pagination.iter_pages对象
            print(i)#1 2 3
    app.run(host='0.0.0.0', port=9901, debug=0)

10. 原生sql支持

if __name__ == '__main__':
    with app.app_context():
        db.create_all()
        statement=text('select * from t_user where id> :id').params(id=1)
        query=db.session.query(User).from_statement(statement)
        print(query.all())# [<User 2>, <User 3>, <User 4>, <User 7>]

        #最傻逼的地方来了，新版本下面语句不支持
        # statement2 = text('select max(id) as mmid,max(age) as mage from t_user where id> :id').params(id=2)
        # query2 = db.session.query('mmid','mage').from_statement(statement2)# sqlalchemy.exc.ArgumentError

        # 感觉不如直接pymysql
        sql=text('select max(id) as mmid,max(age) as mage from t_user where id> :id')
        result=db.session.execute(sql,{'id':2})# CursorResult object
        #print(result.fetchall())# [(7, 45)] 如果这里获取了，下面就获取不了，有点类似游标后移导致没数据读
        for i in result:
            print(i)# (7, 45)
    app.run(host='0.0.0.0', port=9901, debug=0)

参考

flask框架与mysql开发入门到实践 白菜爱科技