"""Abacus Event Model. Python abstraction of an abacus_event, the normalized event that is sent between abacus nodes. """ import json class AbacusEvent: """Abacus Event class.""" def __init__(self, **kwargs): """Set object properties from kwargs.""" self.abacus_event_id = kwargs.get('abacus_event_id', None) self.event_date = kwargs.get('event_date', None) self.event_name = kwargs.get('event_name', None) self.previous_abacus_event_id = kwargs.get('previous_abacus_event_id', None) self.rolled_back_at = kwargs.get('rolled_back_at', None) self.statement_period_id = kwargs.get('statement_period_id', None) self.target_id = kwargs.get('target_id', None) self.target_type = kwargs.get('target_type', None) self.user_id = kwargs.get('user_id', None) self.user_type = kwargs.get('user_type', None) @classmethod def from_params(cls, params): """Get the AbacusEvent from the params passed by ows-abacus-event.""" abacus_event_properties = {} conf = params.get('conf') if conf: abacus_event_properties = json.loads(conf) return AbacusEvent(**abacus_event_properties) @classmethod def from_json(cls, event_json): """Get the AbacusEvent from a JSON string.""" return AbacusEvent(**json.loads(event_json)) def to_json(self): """Get the AbacusEvent as a JSON string.""" return json.dumps(self.to_dict()) def to_dict(self): """Get the AbacusEvent as a python dict.""" return { 'abacus_event_id': self.abacus_event_id, 'event_date': self.event_date, 'event_name': self.event_name, 'previous_abacus_event_id': self.previous_abacus_event_id, 'rolled_back_at': self.rolled_back_at, 'statement_period_id': self.statement_period_id, 'target_id': self.target_id, 'target_type': self.target_type, 'user_id': self.user_id, 'user_type': self.user_type }