长时间运行过程中的 SQLAlchemy 会话管理答案

【问题标题】：SQLAlchemy session management in long-running process长时间运行过程中的 SQLAlchemy 会话管理
【发布时间】：2009-09-14 13:18:43
【问题描述】：

场景：

基于 .NET 的应用服务器 (Wonderware IAS/System Platform) 托管与工厂车间的各种设备通信的自动化对象。
CPython 托管在此应用程序服务器中（使用 Python for .NET）。
自动化对象具有内置的脚本功能（使用基于 .NET 的自定义语言）。这些脚本调用 Python 函数。

Python 函数是跟踪工厂车间工作进度的系统的一部分。系统的目的是沿着流程跟踪生产的小部件，确保小部件以正确的顺序通过流程，并检查在流程中是否满足某些条件。小部件生产历史和小部件状态存储在关系数据库中，这就是 SQLAlchemy 发挥作用的地方。

例如，当一个小部件通过扫描仪时，自动化软件会触发以下脚本（以应用服务器的自定义脚本语言编写）：

' wiget_id and scanner_id provided by automation object
' ExecFunction() takes care of calling a CPython function
retval = ExecFunction("WidgetScanned", widget_id, scanner_id);
' if the python function raises an Exception, ErrorOccured will be true
' in this case, any errors should cause the production line to stop.
if (retval.ErrorOccured) then
    ProductionLine.Running = False;
    InformationBoard.DisplayText = "ERROR: " + retval.Exception.Message;
    InformationBoard.SoundAlarm = True
end if;

脚本调用WidgetScannedpython函数：

# pywip/functions.py
from pywip.database import session
from pywip.model import Widget, WidgetHistoryItem
from pywip import validation, StatusMessage
from datetime import datetime

def WidgetScanned(widget_id, scanner_id):
    widget = session.query(Widget).get(widget_id)
    validation.validate_widget_passed_scanner(widget, scanner) # raises exception on error

    widget.history.append(WidgetHistoryItem(timestamp=datetime.now(), action=u"SCANNED", scanner_id=scanner_id))
    widget.last_scanner = scanner_id
    widget.last_update = datetime.now()

    return StatusMessage("OK")

# ... there are a dozen similar functions

我的问题是：在这种情况下我如何最好地管理 SQLAlchemy 会话？ 应用程序服务器是一个长时间运行的进程，通常在重新启动之间运行几个月。应用服务器是单线程的。

目前，我的做法如下：

我将装饰器应用于我使应用程序服务器可用的功能：

# pywip/iasfunctions.py
from pywip import functions

def ias_session_handling(func):
    def _ias_session_handling(*args, **kwargs):
        try:
            retval = func(*args, **kwargs)
            session.commit()
            return retval
        except:
            session.rollback()
            raise
    return _ias_session_handling

# ... actually I populate this module with decorated versions of all the functions in pywip.functions dynamically
WidgetScanned = ias_session_handling(functions.WidgetScanned)

问题：上面的装饰器是否适合处理长时间运行的会话？我应该打电话给session.remove()吗？

SQLAlchemy 会话对象是一个作用域会话：

# pywip/database.py
from sqlalchemy.orm import scoped_session, sessionmaker

session = scoped_session(sessionmaker())

我想将会话管理排除在基本功能之外。有两个原因：

还有另一个函数族，序列函数。序列函数调用几个基本函数。一个序列函数应该等于一个数据库事务。
我需要能够在其他环境中使用该库。 a) 从 TurboGears Web 应用程序。在这种情况下，会话管理由 TurboGears 完成。 b) 从 IPython 外壳。在这种情况下，提交/回滚将是明确的。

（对于这个冗长的问题，我真的很抱歉。但我觉得我需要解释一下这个场景。也许没有必要？）

【问题讨论】：

标签： python sqlalchemy wonderware

【解决方案1】：

所描述的装饰器适用于长时间运行的应用程序，但如果您不小心在请求之间共享对象，您可能会遇到麻烦。为了使错误更早出现并且不破坏任何内容，最好使用 session.remove() 丢弃会话。

try:
    try:
        retval = func(*args, **kwargs)
        session.commit()
        return retval
    except:
        session.rollback()
        raise
finally:
    session.remove()

或者如果您可以使用with 上下文管理器：

try:
    with session.registry().transaction:
        return func(*args, **kwargs)
finally:
    session.remove()

顺便说一句，您可能希望在查询中使用.with_lockmode('update')，这样您的验证就不会在陈旧数据上运行。

【讨论】：

【解决方案2】：

请您的 WonderWare 管理员授予您对 Wonderware Historian 的访问权限，您可以通过 sqlalchemy 上的 MSSQL 调用非常轻松地跟踪标签的值，您可以经常轮询。

另一种选择是使用 archestra 工具包来监听内部标签更新，并在银河系中部署一个服务器作为平台，您可以从中监听。

【讨论】：