Problem Statement: I am writing Job Scheduling software in Python. I expect users of this software to provide Job definition in XML like format. For example:

    job_name: "myjob"
        command_name: "mycommand"
        command_exec_string : "echo hello"

One of the component of "Job Scheduling" software is "Job Execution Engine", this takes care of executing the job once scheduled. The input to job execution engine is job definition XML file. Job execution engine has to load this XML file in Python objects. In a first step, I am converting XML file to python dictionary using https://github.com/etingof/apacheconfig parser. Then I want to convert this python dictionary returned by the parser to application specific domain objects.

Now for these application specific domain objects I am trying to create a custom type system in Python, with a way to inject validations per fields, support for aliasing field names, providing default values to fields, support to convert the field types etc. I have created prototype for this and want this to be reviewed by Python experts.

from collections import Iterable
import six

class Field(object):
    Specification of a field

    def __init__(self,
        self._type = prop_type
        self._required = required
        self._repeated = repeated
        if choices is not None:
            if not isinstance(choices, Iterable):
                raise IllegalTypeError("Expected iterable got '%s' (%r)" % (type(choices), choices))
            choices = frozenset(choices)
        self._choices = choices
        self._default = default
        self._validators = validators or ()
        self._alias = alias or None
        self._converter = converter

    def is_required(self):
        return self._required

    def is_repeated(self):
        return self._repeated

    # Workaround for not having __set_name__ hook in Python versions before 3.6
    def set_name(self, name):
        Upon class creation, this method is called on all the attribute defined in the class with name of the attribute.
        self._name = name

    def _convert_value(self, value):
        # run converter only if value is not already of defined type
        value_to_store = value
        if self._converter:
            if self.is_repeated() and isinstance(value_to_store, (list, set)):
                value_to_store = [
                    self._converter(val) if (not isinstance(val, self._dtype())) else val for val in value_to_store]
            elif not isinstance(value_to_store, self._dtype()):
                value_to_store = self._converter(value_to_store)
        return value_to_store

    def _store_value(self, owner, value):
        Owner is the object that holds the corresponding field
        value_to_store = value
        if value_to_store:
            value_to_store = self._convert_value(value_to_store)
            if self._repeated:
                if isinstance(value_to_store, (list, set)):
                    value_to_store = [self._run_validators(item) for item in value_to_store]
                    value_to_store = [self._run_validators(value_to_store)]
                value_to_store = self._run_validators(value_to_store)
        owner._values[self._name] = value_to_store

    def _retrieve_value(self, owner):
        return owner._values.get(self._name)

    def _run_validators(self, value):
        if value:
            for validator in self._validators:
                # validator should raise the exception if value is invalid
        return value

    def _get_for_dict(self, owner):
        return self._retrieve_value(owner)

    def check_type(self, value):
        if not isinstance(value, self._dtype()):
            raise IllegalTypeError("Expected '%s' got '%s' (%r) for the field %s" % (self._dtype(),

    def check_choices(self, value):
        if self._choices is not None:
            if value not in self._choices:
                raise IllegalChoiceError("Value provided is outside the set of choices for the field : %s" % self._name)

    def _check_required_field(self, value):
        return self._required and value is None

    def _dtype(self):
        return self._type

    def __get__(self, owner, value):
        if owner is None:
            return self
        return self._retrieve_value(owner)

    def __set__(self, owner, value):
        self._store_value(owner, value)

class ComplexField(Field):
    Specification of a complex field
    def __init__(self, model_class, **kwargs):
        super(ComplexField, self).__init__(model_class, **kwargs)
        self._model_class = model_class

    def __getattr__(self, name):
        return self._model_class._properties.get(name)

    def _get_for_dict(self, owner):
        value = self._retrieve_value(owner)
        if self._repeated:
            value = [item._to_dict() for item in value]
            value = value._todict()
        return value

class BaseModel(type):
    """ """

    def __init__(cls, name, bases, classdict):
        def make_default(field, value):
            def default(self):
                if callable(value):
                    return value(self)
                return value
            setattr(cls, 'default_%s' % field, default)
        super(BaseModel, cls).__init__(name, bases, classdict)
        cls._properties = {}
        cls._alias_to_property_map = {}
        for name in set(dir(cls)):
            attr = getattr(cls, name, None)
            if isinstance(attr, Field):
                cls._properties[attr._name] = attr
                if not getattr(cls, 'default_%s' % attr._name, None):
                    make_default(attr._name, attr._default)
                # Add alias to field class map
                if attr._alias:
                    cls._alias_to_property_map[attr._alias] = attr._name

    def __repr__(cls):
        props = []
        for _, prop in sorted(cls._properties.items()):
            props.append('%s=%r' % (prop._name, prop))
        return '%s<%s>' % (cls.__name__, ', '.join(props))

class Model(object):
    """ """

    def __init__(self, **kwds):
        # add checks for required fields
        self._values = {}

    def _set_attributes(self, kwds):
        cls = self.__class__
        for name, value in kwds.items():
            prop = getattr(cls, name)
            if not isinstance(prop, Field):
                raise IllegalTypeError('Cannot set non-field "%s" value' % name)
            prop._store_value(self, value)

    def _to_dict(self):
        values = {}
        for _, prop in self._properties.items():
            name = prop._name
            value = prop._get_for_dict(self)
            values[name] = value
        return values

    asdict = _to_dict

    def _populate(self):
        for name, prop in self._properties.items():
            if getattr(self, name) is None:
                prop._store_value(self, getattr(self, "default_%s" % name)())

    def __post_init__(self):

    def _check_required_fields(self):

    def __iter__(self):
        values = self._to_dict()
        for key, value in values.items():
            yield (key, value)

    def __getitem__(self, key):
        return getattr(self, key)

    def __setitem__(self, key, value):
        cls = self.__class__
        prop = getattr(cls, key)
        if not isinstance(prop, Field):
            raise IllegalTypeError('Cannot set non-field "%s" value' % key)
        return setattr(self, key, value)

    def __len__(self):
        return 1

    def __post__init__(self):

    def create_object_from_dict(cls, input_dict):
        """ """

        # Step 1: Validate
        # Check if all dictionary fields are defined fields 
        # Check if all required fields are present
        # Convert the aliases 
        # Execute converters

        current_fields = cls.validate_and_convert_input_dictionary(input_dict)
        for key, value in current_fields.items():
            field_class = cls.get_field_class(key)
            is_repeated = cls.is_field_repeated(key)

            if not cls.is_domain_class(field_class):
                val = cls.create_simple_field(value, field_class, is_repeated)

            elif isinstance(value, dict):
                val = cls.create_complex_field_from_dict(field_class, value)

            elif isinstance(value, list):
                # currently list is assumed to be list of dicts this will be changed accordingly
                val = [cls.create_complex_field_from_dict(field_class, v) for v in value]

                raise IllegalTypeError('Improper type of value for field %s' % key)

            current_fields[key] = val

        return cls(**current_fields)

    def is_field_repeated(cls, field_name):
        return cls._properties[field_name].is_repeated()

    def create_simple_field(cls, value, field_class, is_repeated):
        val = None
        # if we have multiple arguments unwrap it and pass to the constructor
        if isinstance(value, dict):
            val = field_class(**value)
        elif is_repeated:
            if isinstance(value, (list, tuple, set)):
                val = [field_class(v) for v in value]
                val = [field_class(value)]
            if isinstance(value, (list, tuple, set)):
                val = field_class(*value)
                val = field_class(value)
        return val

    def get_field_class(cls, field_name):
        # first we check if class is present in the field class map
        class_map = getattr(cls, "field_class_map", None)
        if class_map and field_name in class_map:
            return class_map[field_name]
        return cls._properties[field_name]._type

    def check_if_all_fields_valid(cls, input_dict):  # checks if all the fields defined are already in cls definition
        return all([k in cls._properties for k, v in input_dict.items()])

    def check_if_all_required_fields_present(cls, input_dict):
        return all([k2 in input_dict for k2 in [k for k, v in cls._properties.items() if v.is_required()]])

    def create_complex_field_from_dict(cls, field_class, input_dict):
        return field_class.create_object_from_dict(input_dict)

    def is_domain_class(cls, field_class):
        return issubclass(field_class, Model)

    def convert_alias(cls, input_dict):
        return {
            cls._alias_to_property_map[k] if k in cls._alias_to_property_map else k: v for k, v in input_dict.items()}

    def apply_converters(cls, input_dict):
        current_dict = {}
        for key, val in input_dict.items():
            if val:
                val = cls._properties[key]._convert_value(val)
            current_dict[key] = val
        return current_dict

    def validate_and_convert_input_dictionary(cls, input_dict):
        current_dict = cls.convert_alias(input_dict)

        if not cls.check_if_all_fields_valid(current_dict):
            raise IllegalFieldError("Fields present are outside the fields allowed")

        if not cls.check_if_all_required_fields_present(current_dict):
            raise RequiredFieldNotPresentError("All required fields are not present in the dictionary ")

        return cls.apply_converters(current_dict)

# Example field classes

class CommandSpec(Model):
    command_name = Field(str, default=None)
    command_exec_string = Field(str, default=None) 

class JobSpec(Model):
    job_name = Field(str, default=None)
    command_spec = Field(CommandSpec, repeated = True, default=None)

# Test
job_definition = {"job_name": "myjob", "command_spec": {"command_name":"mycommand", "command_exec_string":"echo hello"}}
job_spec = JobSpec.create_object_from_dict(job_definition)
for key, value in job_spec.__dict__.items():
    print(key, value)

I'd like general comments on how I could improve the coding style. If I am reinventing the wheel, I am using bad coding practices, is there some better way to resolve some of the aspects, am I violating some of the design principles, is code not pythonic etc.

  • 1
    \$\begingroup\$ @TobySpeight since they are using six I guess it's both Python 2 and Python 3. \$\endgroup\$ – Mathias Ettinger Feb 8 at 13:00

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.