场景：

>基于.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;

该脚本调用WidgetScanned python函数：

# 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 Shell.在这种情况下,提交/回滚将是明确的.

(对于这个漫长的问题,我真的感到抱歉.但是,我觉得我需要解释这种情况.也许没有必要吗？)

解决方法:

所描述的装饰器适用于长时间运行的应用程序,但是如果不小心在请求之间共享对象,则可能会遇到麻烦.为了使错误更早出现并且不破坏任何内容,最好使用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’),以便您的验证不会在过时的数据上运行.