SQLAlchemy - 如何在事务中立即提交一些 INSERT？答案

【问题标题】：SQLAlchemy - how can I commit some INSERTs immediately within a transaction?SQLAlchemy - 如何在事务中立即提交一些 INSERT？
【发布时间】：2021-05-27 18:02:18
【问题描述】：

我的 (CLI) 应用使用 SQLAlchemy 1.3。它必须执行的一项工作是查询大量记录（> 300k），然后对这些记录进行一些计算并根据处理结果插入新记录。该应用程序还将其活动写入日志表，因此我可以看到它在长时间运行的工作中当前正在做什么。我有各种管道（任务），所以有一个“管道”表，与“log_messages”表有 1:many 的关系。

我使用的是 ORM 风格，这里省略了模型类。我认为这无关紧要，但如果我应该添加更多详细信息，请告诉我。

所以一般流程是这样的：

def perform_task():
    with session_scope() as session:
        # get a pipeline record for our log messages to link to
        pipeline = session.query(Pipelines).filter(Pipelines.name=='some_name')

        # log the start of the work
        pipeline.append(LogMessage( text="started work")

        # query the records we are working on (>300k) 
        job_input_all = session.query(SomeModel).filter(SomeModel.is_of_interest = True ).all()
        for job_input in job_input_all():
            job_input.append(SomeOtherModel( something_calculated = _do_calculation(job_input, pipeline)))
        
        pipeline.append(LogMessage( text="finished work")        
            
            
def _do_calculation( job_input, pipeline ):
    # of course this isn't the real calcualtion, just illsutrating that "something happens here"
    # the real stuff is complex and takes a lot of time to compute
    # and we need to write log messages from time to time
    calculated_value = job_input.value * 1000
    if calculated_value > 100000:
        pipeline.append(LogMessage( text=f'input value {job_input.value} resulted in bad output {calculated_value}'))

如果我这样做，在会话范围结束之前不会出现任何日志消息，这会提交所有内容。由于这项工作需要很长时间，因此实时更新日志非常重要，这样我就可以看到发生了什么。我该怎么做？如果我在创建每个管道日志消息后提交，我将使 job_input_all 中的行结果对象无效（并强制重新查询），这会很糟糕。更成问题的是：我无法在 _do_calculation() 中提交日志，因为我还不想提交所有计算过的东西。

我以前使用过其他语言的 ORM，但我对 SQLA（和 Python 相关）还是新手，所以我可能在这里遗漏了一些基本的东西。感谢您的帮助！

【问题讨论】：

我想我可能会在两个地方登录：我会将日志保存在数据库中，因为它允许我将日志消息链接到各种表中的记录，我目前正在这样做。但除此之外，我将使用“logger”包将相同的日志消息写入文件。这样，我就可以实时获得应用程序正在执行的操作的即时反馈，并且稍后我会在查询中获得使用日志消息。将很容易添加到真正的应用程序中，因为日志消息的写入已经由一个类中的方法处理，该方法无论如何都会执行所有数据库交互。

标签： python sqlalchemy

【解决方案1】：

我的建议是不要以这种方式写日志。如果您改为将日志写入 ELK（Elasticsearch/Logstash/Kibana）堆栈，您可以使其独立于您的会话，并拥有 Kibana 为您提供的所有不错的内置日志相关功能，在一个易于使用的 web-图形用户界面。

【讨论】：

感谢您让我意识到这一点。不过，出于各种原因，我认为这对我来说不是一个选择。另外：如果我要在其他地方写日志，我将失去将日志条目与应用程序中的其他记录链接的能力，这是我需要的。
您的另一个选择是将会话创建为 Session = sessionmaker(expire_on_commit=False) 这将停止在提交时使您的行对象无效。缺点是这可能会以意想不到的方式影响您的代码。
听起来我应该去看看。进入计算繁重步骤的较大行结果集将永远不会在数据库中更新，因此不使它们过期似乎是一个不错的选择
我想我可能会在两个地方登录：我会将日志保存在数据库中，因为它允许我将日志消息链接到各种表中的记录，我目前正在这样做。但除此之外，我将使用“logger”包将相同的日志消息写入文件。这样，我就可以实时获得应用程序正在执行的操作的即时反馈，并且稍后我会在查询中获得使用日志消息。将很容易添加到真正的应用程序中，因为日志消息的写入已经由一个类中的方法处理，该方法无论如何都会执行所有数据库交互。