"""Secure document.""" from dynamodb_encryption_sdk.identifiers import CryptoAction from payee.connectors.secure_data.exceptions import ( SecureDocumentException, SecureDocumentFieldValidationException, ) from payee.connectors.secure_data.fields.field import Field class SecureDocument: """Secure document.""" _fields: dict[str, Field] = {} _document_type = '' _owner_type = '' created_at = None closed_at = None def __init__(self, data=None): """Init.""" self._data = {} for key, field in self._fields.items(): self._data[key] = field.clone() if data: self.set_values(data, False) def get_value(self, field_name, is_obscured: bool = False): """Get a field's original value.""" if is_obscured: return None return self._data[field_name].value @property def has_obscured_values(self): """The document has obscured values in its fields.""" for item in self._data.values(): if item.has_obscured_value: return True return False @property def fields(self): """Get a list of the document's fields.""" return self._data @property def values(self): """Get all field values as a dict.""" return {key: item.value for (key, item) in self._data.items()} def get_values(self, is_obscured: bool): """Get dict of the document field's value.""" return self.values if is_obscured else self.original_values @property def values_with_meta(self): """Return same as .values with metadata included.""" return { **self.values, 'created_at': self.created_at, 'closed_at': self.closed_at, } @property def original_values(self): """Get all original values.""" return {key: item.original_value for (key, item) in self._data.items()} def set_values(self, original_values: dict, set_empty_values: bool = True): """Set values for the documents field. If set_empty_values is True then we try to set empty values what can lead to validation errors for required fields. But we need to be able to turn this validation off when we load data from the database. """ errors = [] for key in self._data.keys(): try: field = self._data[key] if set_empty_values or not field.is_empty(original_values.get(key)): field.value = original_values.get(key, None) except SecureDocumentFieldValidationException as e: errors.append(str(e)) if errors: raise SecureDocumentException('Validation errors', '\n'.join(errors)) @values.setter def values(self, original_values: dict): """Set values on the document fields. This method will accept a dict and iteratively populate and validate each document field with the corresponding keyed value. """ self.set_values(original_values, True) def set_value(self, field_name, field_value): """Set value on the document field. This method will accept a value to provided document field with. """ try: if field_name not in self._data: raise SecureDocumentFieldValidationException( f'`{field_name}` not exists in `{self._document_type}` document.' ) self._data[field_name].value = field_value except SecureDocumentFieldValidationException as e: raise SecureDocumentException(f'Validation error: {str(e)}') def field_exists(self, field_name): """Validate is field exists in document.""" return field_name in self._data def field_with_value_exists(self, field_name): """Validate is field exists in document and its value is not None.""" return field_name in self._data and self._data[field_name].has_value @classmethod def get_field_actions(cls): """Get the crypto field actions. This will return a dict formatted in the way the encryption SDK expects. Since the default action is to encrypt, only unencrypted fields are included in the output. """ return { key: CryptoAction.DO_NOTHING for (key, item) in cls._fields.items() if not item.encrypt } # noqa @classmethod def document_type(cls): """Get document type.""" return cls._document_type @classmethod def owner_type(cls): """Get owner type.""" return cls._owner_type or cls.__name__