python类where()的实例源码

def sign(user, delay=None):
    bar_rows = bar_table.search(where('user') == user.eid)
    exp = 0

    for bar_row in bar_rows:
        exp += sign_bar(user, Bar(bar_row['kw'], bar_row['fid']))
        if delay is not None:
            time.sleep(delay)

    print('{name}\'s {count} bars was signed, exp +{exp}.'.format(name=user.name, count=len(bar_rows),
                                                                  exp=exp))
    return exp

def modify(user, bduss):
    user_table.update({'bduss': bduss}, where('name') == user.name)

def test_delete(db):
    db.update(delete('int'), where('char') == 'a')
    assert 'int' not in db.get(where('char') == 'a')

def test_increment(db):
    db.update(increment('int'), where('char') == 'a')
    assert db.get(where('char') == 'a')['int'] == 2

def test_decrement(db):
    db.update(decrement('int'), where('char') == 'a')
    assert db.get(where('char') == 'a')['int'] == 0

def test_insert(db):
    db.purge()
    db.insert({'int': 1, 'char': 'a'})

    assert db.count(where('int') == 1) == 1

    db.purge()

    db.insert({'int': 1, 'char': 'a'})
    db.insert({'int': 1, 'char': 'b'})
    db.insert({'int': 1, 'char': 'c'})

    assert db.count(where('int') == 1) == 3
    assert db.count(where('char') == 'a') == 1

def test_remove(db):
    db.remove(where('char') == 'b')

    assert len(db) == 2
    assert db.count(where('int') == 1) == 2

def test_remove_multiple(db):
    db.remove(where('int') == 1)

    assert len(db) == 0

def test_remove_returns_ids(db):
    assert db.remove(where('char') == 'b') == [2]

def test_update(db):
    assert db.count(where('int') == 1) == 3

    db.update({'int': 2}, where('char') == 'a')

    assert db.count(where('int') == 2) == 1
    assert db.count(where('int') == 1) == 2

def test_update_returns_ids(db):
    db.purge()
    assert db.insert({'int': 1, 'char': 'a'}) == 1
    assert db.insert({'int': 1, 'char': 'a'}) == 2

    assert db.update({'char': 'b'}, where('int') == 1) == [1, 2]

def test_update_ids(db):
    db.update({'int': 2}, eids=[1, 2])

    assert db.count(where('int') == 2) == 2

def test_search(db):
    assert not db._query_cache
    assert len(db.search(where('int') == 1)) == 3

    assert len(db._query_cache) == 1
    assert len(db.search(where('int') == 1)) == 3  # Query result from cache

def test_get(db):
    item = db.get(where('char') == 'b')
    assert item['char'] == 'b'

def test_count(db):
    assert db.count(where('int') == 1) == 3
    assert db.count(where('char') == 'd') == 0

def test_contains(db):
    assert db.contains(where('int') == 1)
    assert not db.contains(where('int') == 0)

def test_unique_ids(tmpdir):
    """
    :type tmpdir: py._path.local.LocalPath
    """
    path = str(tmpdir.join('db.json'))

    # Verify ids are unique when reopening the DB and inserting
    with TinyDB(path) as _db:
        _db.insert({'x': 1})

    with TinyDB(path) as _db:
        _db.insert({'x': 1})

    with TinyDB(path) as _db:
        data = _db.all()

        assert data[0].eid != data[1].eid

    # Verify ids stay unique when inserting/removing
    with TinyDB(path) as _db:
        _db.purge()
        _db.insert_multiple({'x': i} for i in range(5))
        _db.remove(where('x') == 2)

        assert len(_db) == 4

        ids = [e.eid for e in _db.all()]
        assert len(ids) == len(set(ids))

def test_unicode_memory(db):
    """
    Regression test for issue #28
    """
    unic_str = 'ß'.decode('utf-8')
    byte_str = 'ß'

    db.insert({'value': unic_str})
    assert db.contains(where('value') == byte_str)
    assert db.contains(where('value') == unic_str)

    db.purge()
    db.insert({'value': byte_str})
    assert db.contains(where('value') == byte_str)
    assert db.contains(where('value') == unic_str)

def test_unicode_json(tmpdir):
    """
    Regression test for issue #28
    """
    unic_str1 = 'a'.decode('utf-8')
    byte_str1 = 'a'

    unic_str2 = 'ß'.decode('utf-8')
    byte_str2 = 'ß'

    path = str(tmpdir.join('db.json'))

    with TinyDB(path) as _db:
        _db.purge()
        _db.insert({'value': byte_str1})
        _db.insert({'value': byte_str2})
        assert _db.contains(where('value') == byte_str1)
        assert _db.contains(where('value') == unic_str1)
        assert _db.contains(where('value') == byte_str2)
        assert _db.contains(where('value') == unic_str2)

    with TinyDB(path) as _db:
        _db.purge()
        _db.insert({'value': unic_str1})
        _db.insert({'value': unic_str2})
        assert _db.contains(where('value') == byte_str1)
        assert _db.contains(where('value') == unic_str1)
        assert _db.contains(where('value') == byte_str2)
        assert _db.contains(where('value') == unic_str2)

def test_eids_json(tmpdir):
    """
    Regression test for issue #45
    """

    path = str(tmpdir.join('db.json'))

    with TinyDB(path) as _db:
        _db.purge()
        assert _db.insert({'int': 1, 'char': 'a'}) == 1
        assert _db.insert({'int': 1, 'char': 'a'}) == 2

        _db.purge()
        assert _db.insert_multiple([{'int': 1, 'char': 'a'},
                                    {'int': 1, 'char': 'b'},
                                    {'int': 1, 'char': 'c'}]) == [1, 2, 3]

        assert _db.contains(eids=[1, 2])
        assert not _db.contains(eids=[88])

        _db.update({'int': 2}, eids=[1, 2])
        assert _db.count(where('int') == 2) == 2

        el = _db.all()[0]
        assert _db.get(eid=el.eid) == el
        assert _db.get(eid=float('NaN')) is None

        _db.remove(eids=[1, 2])
        assert len(_db) == 1