Flask 教程

Flask SqlAlchemy 操作 MySQL 插入数据时,如何使用 ON DUPLICATE KEY UPDATE

Flask 数据库框架及基本操作示例 Flask 数据库框架及基本操作示例


通过 Flask SqlAlchemy 针对 MySQL db 进行插入操作时,若有主键或唯一键值冲突则修改指定字段内容,具体代码如何写?

推荐方式

作者的各个库版本如下:

flask==1.1.2
flask_sqlalchemy==2.5.1
SQLAlchemy==1.4.14

具体代码如下:

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.dialects.mysql import insert


app = Flask(__name__)

# 绑定多个数据库
app.config['SQLALCHEMY_BINDS'] = {
    'a': 'mysql+pymysql://user:password@host:port/db_name',
    'b': 'mysql+pymysql://user:password@host:port/db_name',
}
# 如果是单个数据库
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://user:password@host:port/db_name'

app.config['SQLALCHEMY_ENGINE_OPTIONS'] = {'pool_recycle': 7200, 'pool_pre_ping': True}

db = SQLAlchemy(app)


class TbContent(db.Model):
    # 若上面是多个数据库配置,需要指定库
    __bind_key__ = 'a'
    # 定义表名
    __tablename__ = 'tb_content'
    # 定义字段
    id = db.Column(db.BIGINT, primary_key=True, autoincrement=True)
    uuid = db.Column(db.String(32), nullable=False, unique=True, comment='内容唯一值')
    content = db.Column(db.TEXT)


@app.route('/xxx', methods=['POST'])
def flask_api():
    ...
    # 入库
    dct = {'uuid': uuid, 'content': content}
    insert_on_duplicate_key_update = insert(TbContent).values(dct)\
        .on_duplicate_key_update({'content': content})
    # 若上面配置多个数据库
    res = db.get_engine(app, bind='a').execute(insert_on_duplicate_key_update)
    # 若配置单个数据库
    res = db.engine.execute(insert_on_duplicate_key_update)
    app.logger.info('insert or update id ===> %s, res.lastrowid)
    ...

如上完整的代码所述,需要用底层的sqlalchemy库提供的insert api进行操作。