从子查询中SQLAlchemy的

生成布尔表达式

PENDING_STATE = 'pending' 
COMPLETE_STATE = 'success' 
ERROR_STATE = 'error' 

class Assessment(db.Model): 
    __tablename__ = 'assessments' 

    id = db.Column(db.Integer, primary_key=True) 
    state = db.Column(
     db.Enum(PENDING_STATE, COMPLETE_STATE, ERROR_STATE, 
       name='assessment_state'), 
     default=PENDING_STATE, 
     nullable=False, 
     index=True) 

    test_results = db.relationship("TestResult") 


class TestResult(db.Model): 
    __tablename__ = 'test_results' 

    name = db.Column(db.String, primary_key=True) 
    state = db.Column(
     db.Enum(PENDING_STATE, COMPLETE_STATE, ERROR_STATE, 
       name='test_result_state_state'), 
     default=PENDING_STATE, 
     nullable=False, 
     index=True) 

    assessment_id = db.Column(
     db.Integer, 
     db.ForeignKey(
      'assessments.id', onupdate='CASCADE', ondelete='CASCADE'), 
     primary_key=True)

我想实现的逻辑来更新评估，错误状态，如果任何测试结果是在错误状态如果所有测试结果都处于成功状态，则将评估更新为成功状态。

我可以写原始SQL这样的：

SELECT 'error' 
FROM assessments 
WHERE assessments.state = 'error' OR 'error' IN (
    SELECT test_results.state 
    FROM test_results 
    WHERE test_results.assessment_id = 1);

但我不知道如何翻译成SQLAlchemy的。我认为，子查询会是这样的：

(select([test_results.state]).where(test_results.assessment_id == 1)).in_('error')

，但我无法找到任何方式对文字比较查询结果就像我在原始SQL正在做。我发誓我一定会错过一些东西，但我只是没有看到写返回布尔表达式的查询的方式，我认为这是根本上我所反对的。只是一些简单：

SELECT 'a' = 'b'

似乎是从文档缺席。

有关如何在SQLAlchemy中表示此状态更改的任何想法？如果看起来像我以愚蠢的方式讨论这个问题，我也会很乐意重新思考我的模式。

谢谢！

来源

2014-09-10 Snakes McGee

以下查询应该为error检查。请记住，如果它不是一个错误，则不会返回任何行。

q = (db.session.query(literal_column("'error'")) 
    .select_from(Assessment) 
    .filter(Assessment.id == sid) 
    .filter(or_(
     Assessment.state == ERROR_STATE, 
     Assessment.test_results.any(TestResult.state == ERROR_STATE), 
    )))

如果您希望为success做类似的检查，你会发现，如果有任何TestResult这是不一个success和否定布尔结果。

来源

2014-09-11 20:26:18 van

我实际上最终用postgres触发器来做这件事，这可能是处理状态更新的更好方法。因此，对于错误的情况下，我有：

sqlalchemy.event.listen(TestResult.__table__, 'after_create', sqlalchemy.DDL(""" 
CREATE OR REPLACE FUNCTION set_assessment_failure() RETURNS trigger AS $$ 
BEGIN 
    UPDATE assessments 
    SET state='error' 
    WHERE id=NEW.assessment_id; 
    RETURN NEW; 
END; 
$$ LANGUAGE 'plpgsql'; 

CREATE TRIGGER assessment_failure 
    AFTER INSERT OR UPDATE OF state ON test_results 
    FOR EACH ROW 
    WHEN (NEW.state = 'error') 
    EXECUTE PROCEDURE set_assessment_failure();"""))

以及对于在其中我算测试结果VS的成功测试结果的数量的数量的“成功”的情况下，类似的东西。

虽然我问了一个问题，但可以信赖面包车回答我的问题！谢谢，我没有碰过任何关系。

来源

2014-09-12 16:44:24

感谢您分享您的解决方案。请记住，在这种情况下，您的'sqlalchemy''会话'可能与数据库不同步，所以无论何时提交'TestResult'状态的更新，都需要确保其'Assessment'无效并将从数据库中重新获取。否则它可能仍会显示与其数据库行不同步的状态。 – van 2014-09-12 19:11:21

是的，我想我慢慢说服自己，我根本不需要ORM层。 – 2014-09-12 19:22:37

我明白了。当然，使用任何有意义的东西 – van 2014-09-12 19:23:56

从子查询中SQLAlchemy的

回答

相关问题