python - SQLAlchemy - 错误 TVP 的行必须是 Sequence 对象

Question

C# devloper 玩 python，我在使用 SQLAlchemy 将行插入 SQL 数据库时遇到问题。

更新代码是这样的。

def updateDatabase(self, testSummary):
        params = quote_plus(
            "DRIVER={ODBC Driver 17 for SQL Server};SERVER=GBBED01DB01\SQLHOTEL2;DATABASE=TesterDb;trusted_connection=yes")
        engine = create_engine(
            "mssql+pyodbc:///?odbc_connect={}".format(params))

        Session = sessionmaker(bind=engine)

        session = Session()
        session.add(testSummary)

        for testStepResult in testSummary.testStepResults:   <------- Exception
            session.add(testStepResult)
            for ictResult in testStepResult.ictResults:
                session.add(ictResult)

        try:
            session.commit()
        except Exception as e:
            print("SQL failed to save\n" + e)

我已经标记了异常发生的位置，这就是异常。

(pyodbc.ProgrammingError) ("A TVP's rows must be Sequence objects.", 'HY000') [SQL: 'INSERT INTO [IctResults] ([didTestPass], [testName], component, [lowerLimit], [upperLimit], measured, [rawMeasured], [isOverRange], [isUnderRange], [isPosativeInfinity], [isNegativeInfinity], [testStepResultId]) OUTPUT inserted.id VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)'] [parameters: (1, 'Motor Resistance Check', 'Motor', (2224.0,), (2409.0,), (2292.9794921875,), (2292.9794921875,), 0, 0, 0, 0, 129649)]

表结构看起来像这样

class TestSummary(Base):
__tablename__ = 'TestSummaries'
id = Column(Integer, primary_key=True)
didTestPass = Column(Boolean)
barcode = Column(String)
testDateTime = Column(DateTime)
testerId = Column(Integer)
fixtureId = Column(Integer)
boardId = Column(Integer)
testMode = Column(Integer)
userMode = Column(Integer)
productVariantId = Column(Integer, ForeignKey('ProductVariants.id'))
cycleTimeSeconds = Column(Integer)
testStepResults = relationship("TestStepResult", backref="TestSummary", lazy='dynamic')


class TestStepResult(Base):
__tablename__ = 'TestStepResults'
id = Column(Integer, primary_key=True)
testSummaryId = Column(Integer, ForeignKey('TestSummaries.id'))
testSummary = relationship(TestSummary)
testName = Column(String)
stepTime = Column(Float, default=0)
stepResult = Column(Boolean)
ictResults = relationship("IctResult", backref="TestStepResult", lazy='dynamic')


class IctResult(Base):
__tablename__ = 'IctResults'
id = Column(Integer, primary_key=True)
didTestPass = Column(Boolean)
testName = Column(String)
component = Column(String)
lowerLimit = Column(Float)
upperLimit = Column(Float)
measured = Column(Float)
rawMeasured = Column(Float)
isOverRange = Column(Boolean, default=False)
isUnderRange = Column(Boolean, default=False)
isPosativeInfinity = Column(Boolean, default=False)
isNegativeInfinity = Column(Boolean, default=False)
testStepResultId = Column(Integer, ForeignKey('TestStepResults.id'))
testStepResult = relationship(TestStepResult)

也许我的关系设置错误，但应该是这样。
一个带有许多“TestStepResult”的“测试摘要”和一个带有许多“IctResult”的“TestStepResult”。

Answer 1

正如@ilja 指出的那样，您传递给查询的一些参数是元组，而不是标量。据我所知，SQLAlchemy 不支持开箱即用的数组数据库类型。
我有同样的问题，我的问题是因为我在赋值行的末尾留下了逗号作为复制粘贴的结果。像这样:

testSummary.lowerLimit = 2224.0,
testSummary.upperLimit = 2409.0,

这有效地将标量值转换为单值元组，SQLAlchemy 不知道如何处理它。因此 ProgrammingError异常(exception)。
您现在一定已经解决了您的问题，但我发布此信息是因为这是在网络上搜索此异常时的第一个结果。