rule.py 文件源码

def list_rules(filters={}, session=None):
    """
    List replication rules.

    :param filters: dictionary of attributes by which the results should be filtered.
    :param session: The database session in use.
    :raises:        RucioException
    """

    query = session.query(models.ReplicationRule)
    if filters:
        for (key, value) in filters.items():
            if key == 'created_before':
                query = query.filter(models.ReplicationRule.created_at <= str_to_date(value))
                continue
            elif key == 'created_after':
                query = query.filter(models.ReplicationRule.created_at >= str_to_date(value))
                continue
            elif key == 'updated_before':
                query = query.filter(models.ReplicationRule.updated_at <= str_to_date(value))
                continue
            elif key == 'updated_after':
                query = query.filter(models.ReplicationRule.updated_at >= str_to_date(value))
                continue
            elif key == 'state':
                if isinstance(value, basestring):
                    value = RuleState.from_string(value)
                else:
                    try:
                        value = RuleState.from_sym(value)
                    except ValueError:
                        pass
            elif key == 'did_type' and isinstance(value, basestring):
                value = DIDType.from_string(value)
            elif key == 'grouping' and isinstance(value, basestring):
                value = RuleGrouping.from_string(value)
            query = query.filter(getattr(models.ReplicationRule, key) == value)

    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?)')