<s> """<STR_LIT>""" <EOL> from functools import partial <EOL> from zope . interface import Interface , implementer <EOL> from twisted . python . constants import NamedConstant , Names <EOL> from . _levels import InvalidLogLevelError , LogLevel <EOL> from . _observer import ILogObserver <EOL> class PredicateResult ( Names ) : <EOL> """<STR_LIT>""" <EOL> yes = NamedConstant ( ) <EOL> no = NamedConstant ( ) <EOL> maybe = NamedConstant ( ) <EOL> class ILogFilterPredicate ( Interface ) : <EOL> """<STR_LIT>""" <EOL> def __call__ ( event ) : <EOL> """<STR_LIT>""" <EOL> def shouldLogEvent ( predicates , event ) : <EOL> """<STR_LIT>""" <EOL> for predicate in predicates : <EOL> result = predicate ( event ) <EOL> if result == PredicateResult . yes : <EOL> return True <EOL> if result == PredicateResult . no : <EOL> return False <EOL> if result == PredicateResult . maybe : <EOL> continue <EOL> raise TypeError ( "<STR_LIT>" . format ( result ) ) <EOL> return True <EOL> @ implementer ( ILogObserver ) <EOL> class FilteringLogObserver ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( <EOL> self , observer , predicates , <EOL> negativeObserver = lambda event : None <EOL> ) : <EOL> """<STR_LIT>""" <EOL> self . _observer = observer <EOL> self . _shouldLogEvent = partial ( shouldLogEvent , list ( predicates ) ) <EOL> self . _negativeObserver = negativeObserver <EOL> def __call__ ( self , event ) : <EOL> """<STR_LIT>""" <EOL> if self . _shouldLogEvent ( event ) : <EOL> if "<STR_LIT>" in event : <EOL> event [ "<STR_LIT>" ] . append ( ( self , self . observer ) ) <EOL> self . _observer ( event ) <EOL> else : <EOL> self . _negativeObserver ( event ) <EOL> @ implementer ( ILogFilterPredicate ) <EOL> class LogLevelFilterPredicate ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , defaultLogLevel = LogLevel . info ) : <EOL> """<STR_LIT>""" <EOL> self . _logLevelsByNamespace = { } <EOL> self . defaultLogLevel = defaultLogLevel <EOL> self . clearLogLevels ( ) <EOL> def logLevelForNamespace ( self , namespace ) : <EOL> """<STR_LIT>""" <EOL> if not namespace : <EOL> return self . _logLevelsByNamespace [ None ] <EOL> if namespace in self . _logLevelsByNamespace : <EOL> return self . _logLevelsByNamespace [ namespace ] <EOL> segments = namespace . split ( "<STR_LIT:.>" ) <EOL> index = len ( segments ) - <NUM_LIT:1> <EOL> while index > <NUM_LIT:0> : <EOL> namespace = "<STR_LIT:.>" . join ( segments [ : index ] ) <EOL> if namespace in self . _logLevelsByNamespace : <EOL> return self . _logLevelsByNamespace [ namespace ] <EOL> index -= <NUM_LIT:1> <EOL> return self . _logLevelsByNamespace [ None ] <EOL> def setLogLevelForNamespace ( self , namespace , level ) : <EOL> """<STR_LIT>""" <EOL> if level not in LogLevel . iterconstants ( ) : <EOL> raise InvalidLogLevelError ( level ) <EOL> if namespace : <EOL> self . _logLevelsByNamespace [ namespace ] = level <EOL> else : <EOL> self . _logLevelsByNamespace [ None ] = level <EOL> def clearLogLevels ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _logLevelsByNamespace . clear ( ) <EOL> self . _logLevelsByNamespace [ None ] = self . defaultLogLevel <EOL> def __call__ ( self , event ) : <EOL> eventLevel = event . get ( "<STR_LIT>" , None ) <EOL> namespace = event . get ( "<STR_LIT>" , None ) <EOL> namespaceLevel = self . logLevelForNamespace ( namespace ) <EOL> if ( <EOL> eventLevel is None or <EOL> namespace is None or <EOL> LogLevel . _priorityForLevel ( eventLevel ) < <EOL> LogLevel . _priorityForLevel ( namespaceLevel ) <EOL> ) : <EOL> return PredicateResult . no <EOL> return PredicateResult . maybe </s>
