Source code for caspia.meadow.rules.base

import hashlib
import json
from collections import defaultdict
from uuid import uuid4

from caspia.meadow.serialization import Deserializable, Serializable

_rules_id_counts = defaultdict(int)


[docs]class Rule(Serializable, Deserializable): enable_duplicates = False def __init__(self, identifier=None, name=None): """ Subclasses must call super().__init__ after their initialization !!! """ self.meta = defaultdict(lambda: None) self._identifier = identifier or None self.name = name
[docs] def build_identifier(self): """ Builds unique identifier for this rule. """ serialized = self.serialize_content() if serialized is None: content_ident = str(uuid4()) else: data = json.dumps(serialized, sort_keys=True) content_ident = hashlib.md5(data.encode('utf-8')).hexdigest() if self.enable_duplicates: identifier = content_ident + ':' + str(_rules_id_counts[content_ident]) _rules_id_counts[content_ident] += 1 else: identifier = content_ident return identifier
[docs] def serialize_content(self): """ Subclasses should override this and return serialized ``self``. If the instance cannot be serialized, return None. `identifer` key must not be included. """ raise NotImplementedError
[docs] def serialize(self): serialized = self.serialize_content() if serialized is None: return serialized serialized['identifier'] = self.identifier return serialized
@property def identifier(self): if self._identifier is None: self._identifier = self.build_identifier() return self._identifier def __hash__(self): return hash(self.identifier) def __eq__(self, other): if isinstance(other, type(self)): return self.identifier == other.identifier else: return False