Python sqlalchemy.exc 模块，StatementError() 实例源码

我们从Python开源项目中，提取了以下30个代码示例，用于说明如何使用sqlalchemy.exc.StatementError()。

项目：aardvark 作者：Netflix-Skunkworks | 项目源码 | 文件源码

def sqlite_regex_match(element, compiler, **kw):
    """Compile the SQL expression representing a regular expression match
    for the SQLite engine.
    """
    # determine the name of a custom SQLite function to use for the operator
    operator = element.operator.opstring
    try:
        func_name, _ = SQLITE_REGEX_FUNCTIONS[operator]
    except (KeyError, ValueError), e:
        would_be_sql_string = ' '.join((compiler.process(element.left),
                                        operator,
                                        compiler.process(element.right)))
        raise exc.StatementError(
            "unknown regular expression match operator: %s" % operator,
            would_be_sql_string, None, e)

    # compile the expression as an invocation of the custom function
    regex_func = getattr(func, func_name)
    regex_func_call = regex_func(element.left, element.right)
    return compiler.process(regex_func_call)

项目：rucio 作者：rucio01 | 项目源码 | 文件源码

def list_rule_history(rule_id, session=None):
    """
    List the rule history of a rule.

    :param rule_id: The id of the rule.
    :param session: The database session in use.
    :raises:        RucioException
    """

    query = session.query(models.ReplicationRuleHistoryRecent.updated_at,
                          models.ReplicationRuleHistoryRecent.state,
                          models.ReplicationRuleHistoryRecent.locks_ok_cnt,
                          models.ReplicationRuleHistoryRecent.locks_stuck_cnt,
                          models.ReplicationRuleHistoryRecent.locks_replicating_cnt).filter_by(id=rule_id).order_by(models.ReplicationRuleHistoryRecent.updated_at)

    try:
        for rule in query.yield_per(5):
            yield {'updated_at': rule[0], 'state': rule[1], 'locks_ok_cnt': rule[2], 'locks_stuck_cnt': rule[3], 'locks_replicating_cnt': rule[4]}
    except StatementError:
        raise RucioException('Badly formatted input (IDs?)')

项目：rucio 作者：rucio01 | 项目源码 | 文件源码

def list_associated_rules_for_file(scope, name, session=None):
    """
    List replication rules a file is affected from.

    :param scope:   Scope of the file.
    :param name:    Name of the file.
    :param session: The database session in use.
    :raises:        RucioException
    """

    query = session.query(models.ReplicationRule).\
        with_hint(models.ReplicaLock, "INDEX(LOCKS LOCKS_PK)", 'oracle').\
        join(models.ReplicaLock, models.ReplicationRule.id == models.ReplicaLock.rule_id).\
        filter(models.ReplicaLock.scope == scope, models.ReplicaLock.name == name).distinct()
    try:
        for rule in query.yield_per(5):
            d = {}
            for column in rule.__table__.columns:
                d[column.name] = getattr(rule, column.name)
            yield d
    except StatementError:
        raise RucioException('Badly formatted input (IDs?)')

项目：rucio 作者：rucio01 | 项目源码 | 文件源码

def get_rule(rule_id, session=None):
    """
    Get a specific replication rule.

    :param rule_id: The rule_id to select.
    :param session: The database session in use.
    :raises:        RuleNotFound if no Rule can be found.
    """

    try:
        rule = session.query(models.ReplicationRule).filter_by(id=rule_id).one()
        d = {}
        for column in rule.__table__.columns:
            d[column.name] = getattr(rule, column.name)
        return d

    except NoResultFound:
        raise RuleNotFound('No rule with the id %s found' % (rule_id))
    except StatementError:
        raise RucioException('Badly formatted rule id (%s)' % (rule_id))

项目：rucio 作者：rucio01 | 项目源码 | 文件源码

def get_subscription_by_id(subscription_id, session=None):
    """
    Get a specific subscription by id.

    :param subscription_id: The subscription_id to select.
    :param session: The database session in use.
    :raises: SubscriptionNotFound if no Subscription can be found.
    """

    try:
        subscription = session.query(models.Subscription).filter_by(id=subscription_id).one()
        result = {}
        for column in subscription.__table__.columns:
            result[column.name] = getattr(subscription, column.name)
        return result

    except NoResultFound:
        raise SubscriptionNotFound('No subscription with the id %s found' % (subscription_id))
    except StatementError:
        raise RucioException('Badly formatted subscription id (%s)' % (subscription_id))

项目：sqlalchemy_aio 作者：RazerM | 项目源码 | 文件源码

def test_close(engine):
    conn = await engine.connect()
    assert not conn.closed

    result = await conn.execute(select([1]))
    assert await result.scalar() == 1

    await conn.close()
    assert conn.closed

    with pytest.raises(StatementError) as exc:
        await conn.execute(select([1]))
    assert "This Connection is closed" in str(exc)