from abc import ABC, abstractmethod from enum import Enum from functools import reduce from atlas_um.logs import logger class ClaimValuesSerializer(ABC): is_multiple = False @abstractmethod def serialize(self, claim_values): pass # pragma: no cover @abstractmethod def represent(self, claim_values): pass # pragma: no cover class BooleanAndClaimValuesSerializer(ClaimValuesSerializer): """All claim value components are expected to be boolean values. The combining operation is a logical AND. """ def serialize(self, claim_values): if not claim_values: return None components = map((lambda a: a.components), claim_values) return reduce((lambda a, b: a and b), components) def represent(self, claim_values): if not claim_values: return "" friendly_names = map((lambda a: a.friendly), claim_values) return reduce((lambda a, b: a and b), friendly_names) class BooleanAndIdClaimValuesSerializer(ClaimValuesSerializer): """ All claim value components are expected to be boolean convertable values. The combining operation is a logical AND. """ def serialize(self, claim_values): if not claim_values: return None ids = map(lambda v: self._get_id(v), claim_values) return [reduce((lambda a, b: a and b), ids)] def represent(self, claim_values): if not claim_values: return "" friendly_names = map((lambda a: a.friendly), claim_values) return reduce((lambda a, b: a and b), friendly_names) def _get_id(self, value): try: id = int(value.external_id) except Exception: id = value.external_id if not id: logger.bind( claim_calue=value, serializer=self.__class__.__name__ ).error("Non serializable claim value") return id class ListAppendClaimValuesSerializer(ClaimValuesSerializer): """Claim value components may be any JSON value. All components are lifted into singleton arrays, then all arrays are appended together. Order of the final list dependent on the input list. """ is_multiple = True def serialize(self, claim_values): if not claim_values: return None return list(map((lambda a: a.components), claim_values)) def represent(self, claim_values): if not claim_values: return "" return ", ".join(map((lambda a: a.friendly), claim_values)) class ListAppendIdClaimValuesSerializer(ClaimValuesSerializer): """ The `external_id` field is retrieved from all claim names and appended together. For more compact result trying to cast into integers. Order of the final list dependent on the input list. """ is_multiple = True def serialize(self, claim_values): if not claim_values: return None ids = [] for value in claim_values: try: id = int(value.external_id) except Exception: id = value.external_id if id: ids.append(id) else: logger.bind( claim_calue=value, serializer=self.__class__.__name__ ).error("Non serializable claim value") return ids def represent(self, claim_values): if not claim_values: return "" return ", ".join(map((lambda a: a.friendly), claim_values)) class RegisteredSerializers(Enum): BooleanAnd = BooleanAndClaimValuesSerializer BooleanAndId = BooleanAndIdClaimValuesSerializer ListAppend = ListAppendClaimValuesSerializer ListAppendId = ListAppendIdClaimValuesSerializer @classmethod def of(cls, serializer_name, **kwargs): return RegisteredSerializers[serializer_name].value(**kwargs)