pythonflaskserializationdeserializationmarshmallow

Trouble with finding right solution for serialization in marshmallow


I have two issues:

  1. I would like to make a generic lowercaseAwareSchema that can take a string as an input (not a dict, a string) and load the string along with its lowercase version in a dict
  2. I would like to add validations to the above field not where I'm defining it but where I'm using it, ex: Email() requires the string to be a valid email etc.

Current implementations:

class LowercaseAwareField(Schema):
"""
Schema for a string field with lowercase version
"""

    original = fields.String(required=True)
    lowercase = fields.String(required=True)
    
    @post_load
    def make_fields(self, data, **kwargs):
        data["lowercase"] = data["original"]
        return data

class CountrySchema(Schema):
"""
Schema for country code and name
"""

    code = fields.String(validate=[OneOf([country.name for country in Country])])
    name = fields.String(required=True)
    
    @post_load
    def make_fields(self, data, **kwargs):
        data["name"] = Country[data["code"]].value
        return data

class UserSchema(Schema):
"""
Schema for handling base user data
"""

    email = fields.Pluck(
        LowercaseAwareField, "original", required=True, validate=[Email()]
    )
    password = fields.String(required=True, load_only=True)
    username = fields.Pluck(LowercaseAwareField, "original", required=True)
    country = fields.Pluck(CountrySchema, "code", required=True)

user_schema = UserSchema()

My goal here is:

Input:

{
    "email": "A@b.com",
    "username": "Tanav",
    "password": "sdafasad",
    "country": "US"
}

Post Load:

{
    "country": {
        "code": "US",
        "name": "United States"
    },
    "email": {
        "lowercase": "A@b.com",
        "original": "A@b.com"
    },
    "password": "sdafasad",
    "username": {
        "lowercase": "Tanav",
        "original": "Tanav"
    }
}

Post Dump:

{
    "country": {
        "code": "US",
        "name": "United States"
    },
    "email": {
        "lowercase": "A@b.com",
        "original": "A@b.com"
    },
    "username": {
        "lowercase": "Tanav",
        "original": "Tanav"
    }
}

Solution

  • voila!

    from marshmallow import Schema, fields, post_load, ValidationError, validates
    from marshmallow.validate import Email, OneOf
    from enum import Enum
    
    
    # Example Country Enum
    class Country(Enum):
        US = "United States"
        CA = "Canada"
        UK = "United Kingdom"
    
    
    class LowercaseAwareField(fields.Field):
        def __init__(self, *args, **kwargs):
            self.validators = kwargs.pop("validate", [])
            super().__init__(*args, **kwargs)
    
        def _deserialize(self, value, attr, data, **kwargs):
            if not isinstance(value, str):
                raise ValidationError("Value must be a string.")
            for validator in self.validators:
                validator(value)
            return {"original": value, "lowercase": value.lower()}
    
        def _serialize(self, value, attr, obj, **kwargs):
            if isinstance(value, dict):
                return value
            return {"original": value, "lowercase": value.lower()}
    
    
    class CountrySchema(Schema):
        code = fields.String(validate=OneOf([country.name for country in Country]))
        name = fields.String()
    
        @post_load
        def populate_name(self, data, **kwargs):
            if "code" in data:
                data["name"] = Country[data["code"]].value
            return data
    
    
    class UserSchema(Schema):
        email = LowercaseAwareField(validate=[Email()], required=True)
        password = fields.String(required=True, load_only=True)
        username = LowercaseAwareField(required=True)
        country = fields.Nested(CountrySchema, only=("code",), required=True)
    
        @post_load
        def populate_country_name(self, data, **kwargs):
            if "country" in data:
                data["country"]["name"] = Country[data["country"]["code"]].value
            return data
    
    
    user_schema = UserSchema()
    input_data = {
        "email": "A@b.com",
        "username": "Tanav",
        "password": "sdafasad",
        "country": {"code": "US"}
    }
    
    loaded_data = user_schema.load(input_data)
    print("Loaded Data:")
    print(loaded_data)
    
    dumped_data = user_schema.dump(loaded_data)
    print("\nDumped Data:")
    print(dumped_data)
    

    Good luck!