Flask-SQLAlchemy 在mysql数据库中使用JSON字段

野性酷女 2023-02-28 11:19 22阅读 0赞

mysql版本5.7以后，都支持直接存储 JSON 格式的数据

这次我们用 python3 和 flask-sqlalchemy 进行一个小小的实践！

**flask-sqlalchemy：一个对SQLAlchemy的包装，方便在Flask应用中使用ORM！**

值得一提的好处：从数据库中取出 JSON 数据时，会自动将其转为 python 的字典数据

话不多说，实践一下。

--------------------

首先创建我们的数据库模块 `database.py`

from flask_sqlalchemy import SQLAlchemy
    
    # mysql
    db = SQLAlchemy()
    
    # mysql config(s)
    dbServer = 'localhost'
    dbCharset = 'utf8'
    dbPort = '3306'
    dbName = '数据库名'
    dbUser = '数据库用户名'
    dbPassword = '数据库密码'
    
    
    # connect mysql
    def connect():
        return f'mysql+pymysql://{dbUser}:{dbPassword}@{dbServer}:{dbPort}/{dbName}?charset={dbCharset}'
    
    # db
    class Test(db.Model):
        __tablename__ = 'test' # 设置表名
        __table_args__ = { 'mysql_engine': 'InnoDB', 'mysql_charset': 'utf8'} # 设置引擎、字符集
        id = db.Column(db.String(100), autoincrement=True, primary_key=True) # 自增，主键id
        json_data = db.Column(db.JSON) # JSON数据
        
        # 添加了一个 getDict属性，用于把对象转换成 dict 格式
        @property
        def getDict(self):
            return = { i.name: getattr(self, i.name) for i in self.__table__.columns}

接着创建我们的 Flask app：`main.py`

from flask import Flask, jsonify, request
    from flask_cors import CORS
    import database # 接着在这里引入我们刚刚创建的 database.py
    
    
    # initial(s)
    app = Flask(__name__)
    app.config.update(RESTFUL_JSON=dict(ensure_ascii=False))
    app.config["JSON_AS_ASCII"] = False
    app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
    
    # 数据库连接
    app.config['SQLALCHEMY_DATABASE_URI'] = database.connect()
    
    # 允许跨域
    CORS(app)
    # 分步加载，在这里初始化Flask-SQLAlchemy
    database.db.init_app(app)
    with app.app_context():
        database.db.create_all() # 创建所有表。如果表不存在，那么会自动创建
    
    
    
    @app.route('/')
    def root():
        return jsonify({ 'message': 'hello', 'status': 1})
    
    
    if __name__ == '__main__':
        app.run('0.0.0.0', 5000) # 启动app

一个简单的Flask服务已经出来了，现在就可以通过浏览器访问 `http://localhost:5000` 看到我们的Flask app

现在增加一个测试接口 `add` ，用来给数据库添加数据！

在 `main.py` 中添加：

@app.route('/add/<name>')
    def add(name):
        database.db.session.add(
            # 给 test 表添加 JSON 数据
            database.Test(
                json_data = { 'name': name},
            )
        )
        
        # 提交
        database.db.session.commit()
        print('添加了一个数据：', name)
        
        return jsonify({ 'message': '已经添加', 'name': name, 'status': 1})

添加完毕后，现在我们可以通过浏览器访问这个地址： `http://localhost:5000/add/xiaoming`

接着查看数据库，就可以看到我们已经成功在 `test` 表中的 `json_data` 字段添加了一个 `{'name': 'xiaoming'}` 的数据！

--------------------

**如何用 Flask-SQLAlchemy 查询数据？**

现在添加我们的第二个接口：`get_all` ，一样是在 `main.py` 中：

@app.route('/get_all')
        def getAll(name):
            # 从数据库的 test 表中取出所有对象
            data = database.Test.query.all()
            print(data)
            # 上面取出的 data 是一个列表，由于我们使用的是ORM，所以里面装着 test 表中的数据对象
            # 需要将对象的数据转换成 dict 格式，才能直接用 jsonify 输出
            # 我们先前就在 database.py 中添加了 getDict 属性，所以在这里可以直接使用
            return jsonify('data': [i.getDict for i in data], 'status': 1)

好了，现在访问 `http://localhost:5000/get_all` 试试，数据库表 `test` 中的所有数据都将被显示出来

并且，取出的数据直接就是 JSON 格式的

--------------------

**如何进行条件查询？**

上面的例子我们直接获取了数据库表的所有数据，实际上我们常常都不能这么做，因此需要继续了解条件查询。

刚才用到的查询代码是：

database.Test.query.all()

而条件查询，只用加上一个 `filter_by`：

# 取出 test 表中， id = 1 的所有数据
    database.Test.filter_by(id=1).all()

然而在这个表中，id是主键，所以可以这样直接使用主键查询：

# 取出 test 表中， id = 1 的所有数据
    database.Test.get(1)

不过，由于主键实际上是唯一的，所以使用主键作为条件不需要加`all()`

--------------------

**但是问题来了，我想查询 `name` 这个属性，而这个属性又位于 `json_data` 这个JSON类型的字段中，该怎么办？**

答案是：使用 `filter`，而不是 `filter_by`

# 查询 test 表中的 json_data 字段中，属性 "name" 为 "xiaoming" 的所有数据
    database.Test.query.filter(database.Test.json_data['name'] == 'xiaoming').all()

这样就可以在 `json_data` 字段中找到所有这样的数据：`{"name": "xiaoming"}`

--------------------

**老师，我还想…**

好吧，如果要动态 `filter` ，可以使用 `set()`，像这样：

# 建一个空的set
    filters = set()
    
    if 需要按字段查询:
        # 查询条件跟刚才例子中的一样
        # 就是刚才例子中的这个：database.Test.json_data['name'] == 'xiaoming'
        filters.add(查询条件) 
    
    # 像这样即可
    database.Test.query.filter(*filters).all()

这样就可以实现动态查询了，也就是当 `filters` 为空 `set` 时，可以获得所有数据，而不会进行条件查询

--------------------

理论讲完了，那么现在继续测试，在 `main.py` 中添加接口 `search`：

@app.route('/search')
        def search():
            filters = set()
            # 获取发来的请求所携带的请求参数，格式为 dict
            # 如：http://localhost:5000/search?name=xiaoming
            # 就会得到这样的 args：{'name': 'xiaoming'}
            args = request.args
    
            # 如果有 name 参数，对 json_data 字段中的 name 属性进行查询
            if args.get('name'):
                filters.add(database.Test.json_data['name'] == args.get('name'))
        
            # 进行条件查询，如果 filters 为空 set，则不会进行条件过滤，而是取出所有数据
            data = database.Test.query.filter(*filters).all()
            return jsonify('data': [i.getDict for i in data], 'status': 1)

进行测试，访问这两个地址试试吧：

`http://localhost:5000/search?name=xiaoming`  
`http://localhost:5000/search`

已经实现了动态 `filter`

--------------------

结束