root/pyyaml-legacy/tags/PyYaml_0.32/query.py

import yaml
from test import assertEquals
import re

def Query(query, doc):
    results = []
    for row in yaml.ypath(query['from'],doc):
        data = {}
        if Where(row, query):
            for field in query['select']:
                data[field] = yaml.ypath(field,row).next()
            if query.has_key('order'):
                data['__row__'] = row
            results.append(data)
    return Order(results, query)

def Where(row, query):
    if query.has_key('where'):
        where = query['where']
        return WhereRow(row, query['where'])
    return 1

def WhereRow(row, where):
    if type(where) == type([]):
        return And(row, where, WhereRow)
    if type(where) == type(''):
        return WhereString(row, where)
    return And(row, where.keys(), 
        lambda row, field: where[field] == row[field])

def And(row, conds, condFunc):
    for cond in conds:
        if not condFunc(row, cond):
            return 0
    return 1

def WhereString(row, where):
    where = re.sub("\[(.*?)\]", r"row['\1']", where)
    return eval(where)

def Order(results, query):
    if not query.has_key('order'):
        return results
    compare = lambda x,y: Compare(x,y,query['order'])
    results.sort(compare)
    for result in results:
        del result['__row__']
    return results

def Compare(x, y, order):
    for field in order:
        val = cmp(
            yaml.ypath(field,x['__row__']).next(),
            yaml.ypath(field,y['__row__']).next()
        )
        if val != 0: return val

def TestQuery(query, expected, data):
    assertEquals(Query(query, data), expected, query)    

doc = yaml.loadFile("query.yml").next()
for test in doc['tests']:
    if not test.has_key('ignore'):
        TestQuery(test['query'], test['expected'], doc['data'])