-
-
Save anonymous/ad636439c54596f8a6f3 to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import types | |
class AuditTrail(object): | |
_borg_dict = {} | |
def __init__(self, msg=None): | |
self.__dict__ = self._borg_dict | |
if not hasattr(self, '_log'): | |
self._log = [] | |
if msg: | |
self._log.append(msg) | |
def reset(self): | |
self._log = [] | |
def set_log(self, value): | |
AuditTrail(value) | |
def get_log(self): | |
return self._log | |
log = property(get_log, set_log) | |
class Rule(object): | |
_ARG_COUNT = 0 | |
def __init__(self, *args): | |
if len(args) != self._ARG_COUNT: | |
raise TypeError('Rule %s expected %d arguments, got %d' \ | |
% (self.__class__.__name__, self._ARG_COUNT, len(args))) | |
self.args = list(args) | |
def __call__(self, **kwargs): | |
self._map_kwarg_instances_to_args(**kwargs) | |
run = self.run_filter(**kwargs) | |
self.record_match() | |
return run | |
def _map_kwarg_instances_to_args(self, **kwargs): | |
for index, target in enumerate(self.args): | |
if isinstance(target, types.StringTypes): | |
target_parts = target.split('.') | |
if target_parts[0] in kwargs: | |
target_inst = kwargs[target_parts[0]] | |
try: | |
for prop_or_meth in target_parts[1:]: | |
target_inst = getattr(target_inst, prop_or_meth) | |
if callable(target_inst): | |
self.args[index] = target_inst() | |
else: | |
self.args[index] = target_inst | |
except AttributeError: | |
pass | |
def record_match(self): | |
AuditTrail('%s%s' % (self.__class__.__name__, self.args)) | |
def run_filter(self): | |
raise NotImplementedError | |
class Conditional(object): | |
def __init__(self, *rule_list): | |
self.rule_list = rule_list | |
class and_(Conditional): | |
def evaluate(self, result_set): | |
return all(result_set) | |
class or_(Conditional): | |
def evaluate(self, result_set): | |
return any(result_set) | |
class Filter(object): | |
def __init__(self, selection, *rule_list): | |
self._condition = None | |
self.selection = selection | |
self.rule_list = rule_list | |
self._validate_rule_list() | |
self.match_miss_log = [] | |
def _validate_rule_list(self): | |
if self.rule_list: | |
if isinstance(self.rule_list[0], Conditional): | |
self._condition = self.rule_list[0] | |
self.rule_list = self.rule_list[0].rule_list | |
for rule in self.rule_list: | |
if not isinstance(rule, Rule): | |
raise Exception('Filter expected list of Rule instances, ' | |
'got %s' % type(rule)) | |
def __call__(self, **kwargs): | |
if self._condition: | |
return self._condition.evaluate( | |
[rule(**kwargs) for rule in self.rule_list]) | |
for rule in self.rule_list: | |
result = rule(**kwargs) | |
if result: | |
return True | |
class MatchingFilterNotFoundError(Exception): | |
pass | |
class FilterSet(object): | |
def __init__(self, filter_list): | |
self.filter_list = filter_list | |
self._validate_filter_list() | |
AuditTrail().reset() | |
def _validate_filter_list(self): | |
for _filter in self.filter_list: | |
if not isinstance(_filter, Filter): | |
raise Exception('FilterSet expected list of Filter instances, ' | |
'got %s' % type(_filter)) | |
def __call__(self, **kwargs): | |
for _filter in self.filter_list: | |
if _filter(**kwargs): | |
print AuditTrail().log | |
return _filter.selection | |
raise MatchingFilterNotFoundError('RuleSet filters exhausted ' | |
'without a match') | |
############## rules | |
class Default(Rule): | |
def run_filter(self, **kwargs): | |
return True | |
class IsEqual(Rule): | |
_ARG_COUNT = 2 | |
def run_filter(self, **kwargs): | |
if self.args[0] == self.args[1]: | |
return True | |
######### settings.py | |
choice1 = 'choice1' | |
choice2 = 'choice2' | |
MYCHOICE_FILTERS = ( | |
Filter(choice1, | |
and_(IsEqual(1, 1), | |
IsEqual(2, 2) | |
), | |
), | |
# .... more filters... | |
) | |
########## app code | |
# decide | |
route = FilterSet(MYCHOICE_FILTERS)(myobjects=myobjects) | |
print route | |
'choice1' |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment