Skip to content

Instantly share code, notes, and snippets.

Created March 5, 2013 03:18
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save anonymous/ad636439c54596f8a6f3 to your computer and use it in GitHub Desktop.
Save anonymous/ad636439c54596f8a6f3 to your computer and use it in GitHub Desktop.
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