<s> """<STR_LIT>""" <EOL> from datetime import datetime <EOL> from IPython . config . configurable import LoggingConfigurable <EOL> from IPython . utils . traitlets import Dict , Unicode , Instance <EOL> filters = { <EOL> '<STR_LIT>' : lambda a , b : a < b , <EOL> '<STR_LIT>' : lambda a , b : b > a , <EOL> '<STR_LIT>' : lambda a , b : a == b , <EOL> '<STR_LIT>' : lambda a , b : a != b , <EOL> '<STR_LIT>' : lambda a , b : a <= b , <EOL> '<STR_LIT>' : lambda a , b : a >= b , <EOL> '<STR_LIT>' : lambda a , b : a in b , <EOL> '<STR_LIT>' : lambda a , b : a not in b , <EOL> '<STR_LIT>' : lambda a , b : all ( [ a in bb for bb in b ] ) , <EOL> '<STR_LIT>' : lambda a , b : a % b [ <NUM_LIT:0> ] == b [ <NUM_LIT:1> ] , <EOL> '<STR_LIT>' : lambda a , b : ( b and a is not None ) or ( a is None and not b ) <EOL> } <EOL> class CompositeFilter ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , dikt ) : <EOL> self . tests = [ ] <EOL> self . values = [ ] <EOL> for key , value in dikt . items ( ) : <EOL> self . tests . append ( filters [ key ] ) <EOL> self . values . append ( value ) <EOL> def __call__ ( self , value ) : <EOL> for test , check in zip ( self . tests , self . values ) : <EOL> if not test ( value , check ) : <EOL> return False <EOL> return True <EOL> class BaseDB ( LoggingConfigurable ) : <EOL> """<STR_LIT>""" <EOL> session = Unicode ( "<STR_LIT>" ) <EOL> class DictDB ( BaseDB ) : <EOL> """<STR_LIT>""" <EOL> _records = Dict ( ) <EOL> def _match_one ( self , rec , tests ) : <EOL> """<STR_LIT>""" <EOL> for key , test in tests . items ( ) : <EOL> if not test ( rec . get ( key , None ) ) : <EOL> return False <EOL> return True <EOL> def _match ( self , check ) : <EOL> """<STR_LIT>""" <EOL> matches = [ ] <EOL> tests = { } <EOL> for k , v in check . items ( ) : <EOL> if isinstance ( v , dict ) : <EOL> tests [ k ] = CompositeFilter ( v ) <EOL> else : <EOL> tests [ k ] = lambda o : o == v <EOL> for rec in self . _records . values ( ) : <EOL> if self . _match_one ( rec , tests ) : <EOL> matches . append ( rec ) <EOL> return matches <EOL> def _extract_subdict ( self , rec , keys ) : <EOL> """<STR_LIT>""" <EOL> d = { } <EOL> d [ '<STR_LIT>' ] = rec [ '<STR_LIT>' ] <EOL> for key in keys : <EOL> d [ key ] = rec [ key ] <EOL> return d <EOL> def add_record ( self , msg_id , rec ) : <EOL> """<STR_LIT>""" <EOL> if msg_id in self . _records : <EOL> raise KeyError ( "<STR_LIT>" % ( msg_id ) ) <EOL> self . _records [ msg_id ] = rec <EOL> def get_record ( self , msg_id ) : <EOL> """<STR_LIT>""" <EOL> if msg_id not in self . _records : <EOL> raise KeyError ( "<STR_LIT>" % ( msg_id ) ) <EOL> return self . _records [ msg_id ] <EOL> def update_record ( self , msg_id , rec ) : <EOL> """<STR_LIT>""" <EOL> self . _records [ msg_id ] . update ( rec ) <EOL> def drop_matching_records ( self , check ) : <EOL> """<STR_LIT>""" <EOL> matches = self . _match ( check ) <EOL> for m in matches : <EOL> del self . _records [ m [ '<STR_LIT>' ] ] <EOL> def drop_record ( self , msg_id ) : <EOL> """<STR_LIT>""" <EOL> del self . _records [ msg_id ] <EOL> def find_records ( self , check , keys = None ) : <EOL> """<STR_LIT>""" <EOL> matches = self . _match ( check ) <EOL> if keys : <EOL> return [ self . _extract_subdict ( rec , keys ) for rec in matches ] <EOL> else : <EOL> return matches <EOL> def get_history ( self ) : <EOL> """<STR_LIT>""" <EOL> msg_ids = list ( self . _records . keys ( ) ) <EOL> return sorted ( msg_ids , key = lambda m : self . _records [ m ] [ '<STR_LIT>' ] ) </s>
