"""Set up custom fields for marshmallow.""" import dateutil.parser from dateutil import tz from flask_marshmallow import Marshmallow from marshmallow.validate import OneOf, Range from abacus_common_logic.constants import error from abacus_common_logic.constants.constants import DATE_FORMAT, DATETIME_FORMAT from abacus_common_logic.marshalling.custom_validations import ( check_decimal_precision, check_file_path, not_blank, ) from abacus_common_logic.marshalling.helpers import create_custom_field from abacus_common_logic.utils.dates import is_iso_datetime ma = Marshmallow() ma.Field.default_error_messages['required'] = error.ERROR_FIELD_MISSING def create_enum(options, **kwargs): """Create a custom enum field.""" return create_custom_field( base_class=ma.String, custom_validations=[OneOf(options, error=error.ERROR_INVALID_OPTION)], )(**kwargs) class FormattedDate(ma.Date): """Represents a date.""" default_error_messages = {'invalid': error.ERROR_INVALID_DATE} def __init__(self, **options): """Init method.""" super().__init__(**options, format=DATE_FORMAT) def _serialize(self, value, attr, obj, **kwargs): try: if value is None: return None return value.strftime(DATE_FORMAT) except AttributeError: print('date:', value) return None class FormattedDateTime(ma.DateTime): """Represents a datetime.""" default_error_messages = {'invalid': error.ERROR_INVALID_DATETIME} def __init__(self, **options): """Init method.""" super().__init__(**options, format=DATETIME_FORMAT) def _serialize(self, value, attr, obj, **kwargs): try: if value is None: return None return value.strftime(DATETIME_FORMAT) except AttributeError: print('datetime:', value) return None def _deserialize(self, value, attr, data, **kwargs): if value and is_iso_datetime(value): datetime_value = dateutil.parser.parse(value) if not datetime_value.tzinfo: datetime_value = datetime_value.replace(tzinfo=tz.tzutc()) value = datetime_value.strftime(DATETIME_FORMAT) return super()._deserialize(value, attr, data, **kwargs) ma.Enum = create_enum ma.FormattedDate = FormattedDate ma.FormattedDateTime = FormattedDateTime ma.MoneyAmount = create_custom_field( base_class=ma.Number, custom_validations=[check_decimal_precision], error_messages={'invalid': error.ERROR_INVALID_DECIMAL}, ) ma.NonemptyString = create_custom_field( base_class=ma.String, custom_validations=[not_blank] ) ma.NonNegativeInteger = create_custom_field( base_class=ma.Int, custom_validations=[Range(min=0)], error_messages={'invalid': error.ERROR_MUST_BE_INT}, ) ma.IntegerId = ma.NonNegativeInteger ma.Percentage = create_custom_field( base_class=ma.Number, custom_validations=[ Range(0, 100, error=error.ERROR_PERCENTAGE_NOT_IN_RANGE), check_decimal_precision, ], error_messages={'invalid': error.ERROR_INVALID_DECIMAL}, ) ma.FilePath = create_custom_field( base_class=ma.String, custom_validations=[check_file_path], error_messages={'invalid': error.ERROR_INVALID_FILE_PATH}, ) class TruncatedString(ma.NonemptyString): """TruncatedString serializes a NonemptyString to a truncated length.""" def __init__(self, *args, **kwargs): """Init method. metadata (dict): KV containing the key `truncate` truncate: Desired length of serialized string. If no value or negative value is provided, string is not truncated. """ self.metadata = kwargs.get('metadata', {}) self.truncate = self.metadata.get('truncate', 0) if self.truncate < 0: self.truncate = 0 super().__init__(*args, **kwargs) def _bind_to_schema(self, field_name, parent): super()._bind_to_schema(field_name, parent) def _deserialize(self, value, *args, **kwargs): return super()._deserialize(value, *args, **kwargs) def _serialize(self, value, attr, obj, **kwargs): value = super()._serialize(value, attr, obj, **kwargs) if isinstance(value, str) and self.truncate: value = value[: self.truncate] return value ma.TruncatedString = TruncatedString