如何使SQLAlchemy的插入操作与Postgres多进程安全的Upsert触发器配合使用?

3

我有一个需要使用upsert(插入,如果存在则更新)功能的多进程应用程序。

我决定使用触发器解决upsert问题。(您为每个启用upsert的表添加名为is_upsert的附加列,并在触发器中检查此字段。如果为false,则执行正常插入操作;但如果为true,则执行upsert逻辑-尝试更新,如果因记录不存在而失败,则尝试插入)。

这里是触发器逻辑:

CREATE OR REPLACE FUNCTION upsert_trigger_function_{table}()
RETURNS TRIGGER AS $upsert_trigger_function$
    DECLARE
    row record;
    BEGIN
    RAISE NOTICE 'upsert trigger fired, upsert is %%', NEW.{upsert_column};
    IF NEW.{upsert_column} THEN
        NEW.{upsert_column} := false;
        LOOP
            UPDATE {table} SET
                {update_set}
            WHERE
                {update_where}
            ;
            IF found THEN
                RETURN NULL;
            END IF;
            BEGIN
                INSERT INTO {table} SELECT NEW.*;
                RETURN NULL;
            EXCEPTION WHEN unique_violation THEN
                -- loop
            END;
        END LOOP;
        RETURN NULL;
    ELSE
        RETURN NEW;
    END IF;
    END;
$upsert_trigger_function$ LANGUAGE plpgsql;

测试对象,(add_upsert 只是使上面的触发器被安装):

class SimpleItem(PipelinesBase):
    __tablename__ = 'simple_item'

    id = Column(BigInteger, primary_key=True)
    item_type = Column(String, nullable=False, unique=True)
    quantity = Column(Integer, nullable=False)
    price = Column(Float, nullable=False)
    in_stock = Column(Boolean, nullable=False)
    arrived = Column(Date)
    sys_time = Column(
    TSTZRANGE,
    nullable=False,
    server_default=text("TSTZRANGE(now(), null)"),
    )
    _upsert = Column(Boolean, nullable=False, server_default=text('false'))
    _type_identifier = 1400
add_upsert(SimpleItem, ['item_type'])

测试脚本

from sqlalchemy.engine import create_engine
from pipelines.settings_proxy import TEST_DB
from sqlalchemy.orm.session import sessionmaker
from test_pipelines.test_persistence.mock_items import SimpleItem
from test_pipelines.test_persistence.helpers import random_simple_item

def main():
    engine = create_engine(TEST_DB)

    values = random_simple_item(_upsert=True)

    session = sessionmaker(engine)()

    si = SimpleItem(**values)
    session.add(si)
    session.commit()

    si = SimpleItem(**values)
    si.price = 1
    session.merge(si)
    session.commit()

使用SQL语句时它能正常工作,但是当我将其与SQLAlchemy ORM添加对象一起使用时,会出现问题。

Traceback (most recent call last):
  File "pipelines/persistence/experiment_with_upsert_field.py", line 59, in <module>
    main()
  File "pipelines/persistence/experiment_with_upsert_field.py", line 27, in main
    session.commit()
  File "/home/sebastian/local/virtualenvs/perception/lib/python3.4/site-packages/sqlalchemy/orm/session.py", line 801, in commit
    self.transaction.commit()
  File "/home/sebastian/local/virtualenvs/perception/lib/python3.4/site-packages/sqlalchemy/orm/session.py", line 392, in commit
    self._prepare_impl()
  File "/home/sebastian/local/virtualenvs/perception/lib/python3.4/site-packages/sqlalchemy/orm/session.py", line 372, in _prepare_impl
    self.session.flush()
  File "/home/sebastian/local/virtualenvs/perception/lib/python3.4/site-packages/sqlalchemy/orm/session.py", line 2019, in flush
    self._flush(objects)
  File "/home/sebastian/local/virtualenvs/perception/lib/python3.4/site-packages/sqlalchemy/orm/session.py", line 2137, in _flush
    transaction.rollback(_capture_exception=True)
  File "/home/sebastian/local/virtualenvs/perception/lib/python3.4/site-packages/sqlalchemy/util/langhelpers.py", line 60, in __exit__
    compat.reraise(exc_type, exc_value, exc_tb)
  File "/home/sebastian/local/virtualenvs/perception/lib/python3.4/site-packages/sqlalchemy/util/compat.py", line 184, in reraise
    raise value
  File "/home/sebastian/local/virtualenvs/perception/lib/python3.4/site-packages/sqlalchemy/orm/session.py", line 2101, in _flush
    flush_context.execute()
  File "/home/sebastian/local/virtualenvs/perception/lib/python3.4/site-packages/sqlalchemy/orm/unitofwork.py", line 373, in execute
    rec.execute(self)
  File "/home/sebastian/local/virtualenvs/perception/lib/python3.4/site-packages/sqlalchemy/orm/unitofwork.py", line 532, in execute
    uow
  File "/home/sebastian/local/virtualenvs/perception/lib/python3.4/site-packages/sqlalchemy/orm/persistence.py", line 174, in save_obj
    mapper, table, insert)
  File "/home/sebastian/local/virtualenvs/perception/lib/python3.4/site-packages/sqlalchemy/orm/persistence.py", line 800, in _emit_insert_statements
    execute(statement, params)
  File "/home/sebastian/local/virtualenvs/perception/lib/python3.4/site-packages/sqlalchemy/engine/base.py", line 914, in execute
    return meth(self, multiparams, params)
  File "/home/sebastian/local/virtualenvs/perception/lib/python3.4/site-packages/sqlalchemy/sql/elements.py", line 323, in _execute_on_connection
    return connection._execute_clauseelement(self, multiparams, params)
  File "/home/sebastian/local/virtualenvs/perception/lib/python3.4/site-packages/sqlalchemy/engine/base.py", line 1010, in _execute_clauseelement
    compiled_sql, distilled_params
  File "/home/sebastian/local/virtualenvs/perception/lib/python3.4/site-packages/sqlalchemy/engine/base.py", line 1159, in _execute_context
    result = context._setup_crud_result_proxy()
  File "/home/sebastian/local/virtualenvs/perception/lib/python3.4/site-packages/sqlalchemy/engine/default.py", line 828, in _setup_crud_result_proxy
    self._setup_ins_pk_from_implicit_returning(row)
  File "/home/sebastian/local/virtualenvs/perception/lib/python3.4/site-packages/sqlalchemy/engine/default.py", line 893, in _setup_ins_pk_from_implicit_returning
    for col in table.primary_key
  File "/home/sebastian/local/virtualenvs/perception/lib/python3.4/site-packages/sqlalchemy/engine/default.py", line 891, in <listcomp>
    for col, value in [
TypeError: 'NoneType' object is not subscriptable

在sqlalchemy.engine.default的深处引发错误。我相信这是因为我的触发器在执行UPSERT时返回NULL,而SQLAlchemy试图使用RETURNING语句传播具有插入ID的对象。显然会失败,因为无法从下属的INSERT / UPDATE中获取正确的ID,同时阻止正常的插入。
请注意,我已经测试了upsert作为特殊函数,但对于更新与其他项目存在关系(具有关系)的复杂项目,它对我没有用。
所以我的问题是:如何告诉SQLAlchemy避免加载插入对象的ID?

你无法使用PostgreSQL 9.5吗?能否同时发布完整的堆栈跟踪和您如何从Python调用pl/pgsql函数的方法? - univerio
无法使用9.5版本。以下是更多细节。然而,可能无法避免使用ORM进行加载。我通过使用SQLAlchemy核心解决了这个问题。 - omikron
1个回答

1
经过研究和测试,我发现在SQLAlchemy ORM中无法实现。但是,在SQLAlchemy Core中可以通过将inline关键字参数设置为True来实现:
engine.execute(
    SimpleItem.__table__.insert(inline=True),
    values
)

values['price'] = 1
engine.execute(
    SimpleItem.__table__.insert(inline=True),
    values
)

网页内容由stack overflow 提供, 点击上面的
可以查看英文原文,
原文链接