如何在flask_sqlalchemy中使用PostgreSQL的"INSERT ... ON CONFLICT"(UPSERT)功能？

Question

语句中的PostgreSQL ON CONFLICT子句INSERT提供"upsert"功能(即更新现有记录,或者如果不存在此类记录则插入新记录).此功能在SQLAlchemy中通过支持on_conflict_do_nothing和on_conflict_do_updatePostgreSQL的方言对方法的Insert对象(如描述这里):

from sqlalchemy.dialects.postgresql import insert

insert_stmt = insert(my_table).values(
    id='some_existing_id',
    data='inserted value'
)

do_nothing_stmt = insert_stmt.on_conflict_do_nothing(
    index_elements=['id']
)

conn.execute(do_nothing_stmt)

do_update_stmt = insert_stmt.on_conflict_do_update(
    constraint='pk_my_table',
    set_=dict(data='updated value')
)

conn.execute(do_update_stmt)

我正在使用flask_sqlalchemy,它管理SQLAlchemy的引擎,会话和连接.要向数据库添加元素,我创建模型的实例,将其添加到数据库会话,然后调用commit,如下所示:

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
db = SQLAlchemy(app)

class MyTable(db.Model):
    id = db.Column(UUID, primary_key=True)
    data = db.Column(db.String)

relation = MyTable(id=1, data='foo')
db.session.add(relation)
db.session.commit()

因此,Insert对象被完全包裹和遮挡flask_sqlalchemy.

如何访问PostgreSQL特定的方言方法来执行upsert？我是否需要绕过flask_sqlalchemy并创建自己的会话？如果我这样做,我怎么能确保没有冲突？

Answer 1

事实证明，您可以在上执行较低级别的语句db.session。因此，解决方案如下所示：

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.dialects.postgresql import insert as pg_insert

app = Flask(__name__)
db = SQLAlchemy(app)

class MyTable(db.Model):
    id = db.Column(UUID, primary_key=True)
    data = db.Column(db.String)

    def __init__(self, _id, *args, **kwargs):
        self.id = _id
        self.data = kwargs['data']

    def as_dict(self):
        return {'id': self.id, 'data': self.data}

    def props_dict(self):
        d = self.as_dict()
        d.pop('id')
        return d

relation = MyTable(id=1, data='foo')
statement = pg_insert(MyTable)\.
    values(**relation.as_dict()).\
    on_conflict_do_update(constraint='id',
                          set_=relation.props_dict())

db.session.execute(statement)
db.session.commit()

我的模型类中的as_dict()和props_dict()方法允许我使用构造函数从传入的HTTP请求中过滤掉不需要的属性。