Python Serialization as JSON or YAML exploiting YAML TAGS

Python enables you to easily serialize objects as either JSON or YAML: very often it is very convenient to leverage on these features exploiting them to enhance your own object. YAML serialization comes almost for free if you derive your classes from the YAML object, whereas automatic instantiation of objects from a YAML document requires a little bit of work and leverages on YAML tags. As for JSON serialization, it is bloody useful and enables you to quickly and easily serialize the contents of your objects into JSON documents that can be exploited for example to interact with or to develop a REST API.

This post, Python Serialization as JSON or YAML exploiting YAML TAGS, shows you how to develop a Python package that provides such kinds of objects. As usual we take particularly care of the code style, using a very clean and portable design and adhering best practices.

The operating environment used in this post is Red Hat Enterprise Linux 9 or Rocky Linux 9 - using a different environment may lead to having to adapt or even change things.

Overview

As usual I want to show things in action, so in this post we will develop a trivial Python Flask application that is configured using a complex YAML manifest. Anyway, mind that since the goal of this post is about YAML and Python serialization, the use of Flask here is very trivial: giving Flask all the room it deserves here risks to confuse people and bring them away from the main topic. Anyway I think I will write something on Flask sooner or later.

In this post we will develop the "fooapp" service scoped within the "carcano" namespace: it consists of a Python package made of several modules  and of course of a very small driver program to test everything.

Create the Workspace

Let's begin by creating the workspace for the project - simply create a directory with the name of the project:

mkdir ~/fooapp
cd ~/fooapp

then let's create the "src" directory we use to store all the source files.

mkdir src

Creating the "src" directory is a best practice: this way you keep source files grouped together, and you can create directories to group other files by purpose. Examples of purposes can be providing documentation (group them in a directory called "docs"), samples (group them in a directory called "samples"), packaging (for example, when dealing with RPM, you can create a directory called RPM) . Mind that a clean directory structure is the very first thing to have when developing, especially when you have to deal with massive projects. For more information on this topic, please refer to the post Python Full-Featured Project.

The fooapp Python Package

It is now time to create the "carcano.fooapp" Python package directory tree - just create the following directories:

mkdir -p src/carcano/fooapp

then create an empty "__init__.py" file beneath "src/carcano/fooapp" directory - mind that it is said "empty" since it has no directives out of the initial line and a few metadata:

# __init__.py
__author__ = "Marco Antonio Carcano"
__version__ = '1.0.0'

Implement Models

Let's start describing the "carcano.fooapp" Python Package we are about to develop - we breakdown its overall configuration into the following classes:

• Service
• Monitoring
• Policy
• Rule
• Grant

Since the above objects are actually models, we group them beneath "carcano.fooapp.models"
Besides the classes, we also provide a YAML loader function that automatically calls the YAML constructor of each class when needed, automatically instantiating them on the fly.

Since these objects are actually models, we group them together beneath the "models" directory - create it as follows:

mkdir -p src/carcano/fooapp/models

then, same way as we already did within the "carcano/fooapp" directory,  we create an empty "__init__.py" file beneath "src/carcano/fooapp/models" directory:

# __init__.py
__author__ = "Marco Antonio Carcano"
__version__ = '1.0.0'

Now that we have created the "src/carcano/fooapp/models" package, we must add the actual objects. The best practice is to have a dedicated file to each of the objects, 

Grant object

Besides being the more nested object of the hierarchy, the Grant object is also quite simple so it is a good candidate to start the implementation from.

Its purpose is storing a generic grant that binds a list of users or a list of groups to a specific access level. For the user's convenience, it also provides an optional description field that can be used to explain the meaning of the specific grant.

We begin by defining the Grant object as a Python class, exploiting the standard Python logging facility and decorators to implement setters and getters method that wrap the attributes of the class. The straightforward advantage is that this way we can easily validate the input and raise exceptions if necessary.

In my experience, the very most of the problems of software comes by developers postponing the implementation of exception handling. Please always implement exception handling immediately when you are developing  any specific part of your code, don't postpone it. If you wait until things work even in very basic ways, it is very likely indeed that whoever manages the project won't give you the time to implement exception handling or code cleanup since "There is no problem: it's working, so we can tick it as done." This kind of attitude is very dangerous: it dumps your responsibility on service operations guys, causing them to spend much more time to figure out the cause of problems when incidents arise. And a delay into service recovery means loss of money, and sometimes also loss of trustability for the corporate. I prefer not to disclose the adjectives I use to define this kind of  project managers (that by the way are the very most, ... as claimed by the 1st law of the masterpiece of Carlo M. Cipolla - "The basic laws of human stupidity". Have a look to the great definition of stupid and of bandit he provides, and think about how many of them have you found so far :O).

Create the "src/carcano/fooapp/models/grant.py" file with the following contents:

__author__ = "Marco Antonio Carcano"
__version__ = '0.1.0'

import sys, logging

log = logging.getLogger(__name__)
log.addHandler(logging.NullHandler())

class Grant():
    """ Grant object """ 

    def __init__(self, access_level, description=None, groups=[], users=[]):
        """
        Initialization

        :param access_level:     granted access level
        :param description:      optional descriptive text of the grant
        :param groups:           a list of groups the access level is granted
        :param users:            a list of users the access level is granted
        """
        self.access_level = access_level
        self._description=description
        self.groups = groups
        self.users = users

    def __del__(self):
       """
       destructor
       """
       self._groups.clear()
       self._users.clear()

    @property
    def access_level(self):
        """
        granted access level
        """
        return self._access_level;
    
    @access_level.setter
    def access_level(self, access_level):

        if (not isinstance(access_level, int)) :
            raise ValueError("'access_level' must be an integer")
        self._access_level = access_level
    
    @property
    def description(self):
        """
        optional descriptive text of the grant
        """
        return self._description;
    
    @description.setter
    def description(self, description):
        if (len(description) == 0) :
            log.error('Package: {} - HTTP {} - "description" cannot be empty'
                .format(__name__))
            raise ValueError("'description' cannot be empty")
        self._description = description

    @property
    def groups(self):
        """
        a list of groups the access level is granted
        """
        return self._groups;

    @groups.setter
    def groups(self, groups):
        if not isinstance(groups, list):
            raise ValueError("'groups' must be of type 'list'")
        self._groups = groups

    @property
    def users(self):
        """
        a list of users the access level is granted
        """
        return self._users

    @users.setter
    def users(self, users):
        if not isinstance(users, list):
            raise ValueError("'users' must be of type 'list'")
        self._users = users

Just a few comments of the above code:

• setup Python standard logging facility (lines 6-7)
• initialise the Grant class (lines 12-24)
• implement a destructor to free the list when the instance of the Gran object is not used anymore (lines 26-31)
• implement setters and getters methods (lines 33-86) using decorators

If any part of the above code would not be clear to you, please read first the post Python Full-Featured Project: it clearly explains the use of  Python docstrings, of decorators to implement attribute setters and getters method and so on. I strongly advise you reading it before going on, also because it explains how to cleanly structure a Python application, how to package it using Python standard facilities and push it to PyPI, and how to package it as an RPM. It is a must read.

The above code is a Python Object that can already be used, but it is still missing serialisation facilities: it would be a pity leaving it like that.

JSON Serialisation

To address the need of representing the contents of the objects in JSON format, we implement the "toJSON()" method: just add the following content to the end of the "grant.py" file.

    def toJSON(self):
        """
        returns a JSON serialisation of the Grant object
        """
        return dict( access_level=self._access_level,
            description=self._description, groups=self._groups,
            users=self._users )

as you see, the above code just defines a function that creates a dictionary on the fly with the contents of every attribute of the class.

YAML Serialisation

PyYAML provides a convenient and easy way not only to serialise an object as YAML, but also to automatically instance an object from a YAML document, supporting also nested objects.

The very first thing to do is obviously to import the "yaml" Python package. Modify the import statement at the beginning of the "grant.py" file as follows:

import yaml, sys, logging

The mandatory step to serialise contents as YAML is deriving the Grant object from the YAMLObject as follows:

class Grant(yaml.YAMLObject):
    """ Grant object """ 

In compliance with encapsulation best practices, we declared the attributes as private (anyway remember that in Python this is only a hint - if you don't know why, again, read the Python Full-Featured Project post).

Because of this, we need to implement the "__getstate__" dundee so that PyYAML knows how to derive the proper names of the attributes from the private attributes of the class.

    def __getstate__(self):
        """
        returns the state of the Grant object
        """
        return dict( access_level=self._access_level,
            description=self._description, groups=self._groups,
            users=self._users )

YAML Constructor

As we said, PyYAML lets you easily instantiate even nested objects reading their definition from YAML documents. To do this magic, it leverages on Python Tags, so the first thing to do is add the "yaml_tag" attribute to the Grant object - put it at the very beginning of the class.

    yaml_tag = u'!Grant'

If you are one of my loyal readers, you certainly remember that I wrote the YAML in a nutshell post , that thoroughly explains YAML format and so also YAML tags. Haven't you read it so far? It's a pity, believe me.

Then we must develop a constructor function that can be added to the PyYAML loader, explaining what object to instantiate when a specific tag is found: add the following lines to the end of the "grant.py" file:

def grant_constructor(loader: yaml.SafeLoader,
    node: yaml.nodes.MappingNode) -> Grant:
    """ 
    constructs a Grant object. Can be added to yaml.loader
    """
    try:
        return Grant(**loader.construct_mapping(node))
    except TypeError as e:
        print(e, file=sys.stderr)
        exit(1)

in the above code, the function constructor called "grant_constructor" marks the return annotation (the "->" arrow) to the Grant object that gets instantiated after unpacking the dictionary  (the ** ) provided by 'loader.construct_mapping(node)' - this latter is the method that parses the YAML node of a dictionary.

If you need to parse a YAML node of kind sequence, use 'loader.construct_sequence(node)' instead.

Mind that constructors must be functions, not a method of classes. 

For your convenience, this is how the whole "src/carcano/fooapp/models/grant.py" file must looks like:

__author__ = "Marco Antonio Carcano"
__version__ = '0.1.0'

import yaml, sys, logging

log = logging.getLogger(__name__)
log.addHandler(logging.NullHandler())

class Grant(yaml.YAMLObject):
    """ Grant object """ 

    yaml_tag = u'!Grant'

    def __init__(self, access_level, description=None, groups=[], users=[]):
        """
        Initialization

        :param access_level:     granted access level
        :param description:      optional descriptive text of the grant
        :param groups:           a list of groups the access level is granted
        :param users:            a list of users the access level is granted
        """
        self.access_level = access_level
        self._description=description
        self.groups = groups
        self.users = users

    def __del__(self):
       """
       destructor
       """
       self._groups.clear()
       self._users.clear()

    @property
    def access_level(self):
        """
        granted access level
        """
        return self._access_level;
    
    @access_level.setter
    def access_level(self, access_level):

        if (not isinstance(access_level, int)) :
            raise ValueError("'access_level' must be an integer")
        self._access_level = access_level
    
    @property
    def description(self):
        """
        optional descriptive text of the grant
        """
        return self._description;
    
    @description.setter
    def description(self, description):
        if (len(description) == 0) :
            log.error('Package: {} - HTTP {} - "description" cannot be empty'
                .format(__name__))
            raise ValueError("'description' cannot be empty")
        self._description = description

    @property
    def groups(self):
        """
        a list of groups the access level is granted
        """
        return self._groups;

    @groups.setter
    def groups(self, groups):
        if not isinstance(groups, list):
            raise ValueError("'groups' must be of type 'list'")
        self._groups = groups

    @property
    def users(self):
        """
        a list of users the access level is granted
        """
        return self._users

    @users.setter
    def users(self, users):
        if not isinstance(users, list):
            raise ValueError("'users' must be of type 'list'")
        self._users = users

    def toJSON(self):
        """
        returns a JSON serialization of the Grant object
        """
        return dict( access_level=self._access_level,
            description=self._description, groups=self._groups,
            users=self._users )

    def __getstate__(self):
        """
        returns the state of the Grant object
        """
        return dict( access_level=self._access_level,
            description=self._description, groups=self._groups,
            users=self._users )

def grant_constructor(loader: yaml.SafeLoader,
    node: yaml.nodes.MappingNode) -> Grant:
    """ 
    constructs a Grant object. Can be added to yaml.loader
    """
    try:
        return Grant(**loader.construct_mapping(node))
    except TypeError as e:
        print(e, file=sys.stderr)
        exit(1)

now add to the "__init__.py" file in the "src/carcano/fooapp/models" directory the necessary code to load the Grant class:

from .grant import Grant

Since I already explained what part of the code are the ones necessary to serialise to JSON and YAML and to instantiate from a YAML, by now I put the code of an object in a single block of code.

Rule object

The purpose of the Rule object is storing a generic rule that provides a list of instances of the Grant object. For the user's convenience, it also provides an optional description field that can be used to explain the meaning of the specific Rule.

Create the "src/carcano/fooapp/models/rule.py" file with the following contents:

__author__ = "Marco Antonio Carcano"
__version__ = '0.1.0'

import yaml, sys, logging

log = logging.getLogger(__name__)
log.addHandler(logging.NullHandler())

class Rule(yaml.YAMLObject):
    """ Rule object"""

    yaml_tag = u'!Rule'

    _grants = []

    def __init__(self, name, description=None, grants=[]):
        """
        Initialization

        :param name:            a string with the name of the rule
        :param description:     optional descriptive text of the rule
        :param grants:          optional list of Grant objects
        """
        self.name = name
        self._description = description
        self.grants = grants

    def __del__(self):
        """
        destructor
        """
        self._grants.clear()

    @property
    def name(self):
        """
        a string with the name of the rule
        """
        return self._name;

    @name.setter
    def name(self, name):
        if (len(name) == 0) :
            raise ValueError("'name' cannot be empty")
        self._name = name

    @property
    def description(self):
        """
        optional descriptive text of the rule
        """
        return self._description;

    @description.setter
    def description(self, description):
        if (len(description) == 0) :
            raise ValueError("'description' cannot be empty")
        self._description = description

    @property
    def grants(self):
        """
        optional list of Grant objects
        """
        return self._grants;

    @grants.setter
    def grants(self, grants=[]):
        if not isinstance(grants, list):
            raise ValueError("'grants' must be of type 'list'")
        self._grants = grants[:]

    def toJSON(self):
        """
        returns a JSON serialization of the Rule object
        """
        jgrants = []
        for policy in self._grants:
            jgrants.append(policy.toJSON())
        return dict(name=self._name, description=self._description,
            grants=jgrants)

    def __getstate__(self):
        """
        returns the state of the Rule object
        """
        return dict(name=self._name, description=self._description,
            grants=self._grants)

def rule_constructor(loader: yaml.SafeLoader,
    node: yaml.nodes.MappingNode) -> Rule:
    """ 
    constructs a Rule object. Can be added to yaml.loader
    """
    try:
        return Rule(**loader.construct_mapping(node, deep=True))
    except TypeError as e:
        print(e, file=sys.stderr)
        exit(1)

A few comments on the above code:

• setup Python standard logging facility (lines 6-7)
• initialise the Rule class (lines 16-26)
• implement a destructor to free the list when the instance of the Rule object is not used anymore (lines 28-32)
• implement setters and getters methods (lines 34-71) using decorators
• implement the JSON serialization method (lines 73-81)
• implement the "__getstate__" dundee (lines 83-88)
• implement the YAML constructor (lines 90-99)

now add to the "__init__.py" file in the "src/carcano/fooapp/models" directory the necessary code to load the Rule class:

from .rule import Rule

Policy object

The purpose of the Policy object is storing a generic policy that matches a specific criteria along with lists of instances of the Rule objects specific for the "foo" feature and for the "bar" feature. For the user's convenience, it also provides a boolean flag field that can be used to toggle the policy on or off.

Create the "src/carcano/fooapp/models/policy.py" file with the following contents:

__author__ = "Marco Antonio Carcano"
__version__ = '0.1.0'

import yaml, sys, logging

log = logging.getLogger(__name__)
log.addHandler(logging.NullHandler())

class Policy(yaml.YAMLObject):
    """ Policy object"""

    yaml_tag = u'!Policy'

    _feature_foo_rules = {}
    _feature_bar_rules= {}

    def __init__(self, matching_criteria, feature_foo_rules=None,
        feature_bar_rules=None, enabled="False"):
        """
        Initialization

        :param matching_criteria:  the criteria that must match so to apply the
                                   policy
        :param feature_foo_rules:  a list of Rule objects for the foo feature
        :param feature_bar_rules:  a list of Rule objects for the foo feature
        :param enabled             a boolean to toggle the policy on or off
        """
        self.matching_criteria = matching_criteria
        self._feature_foo_rules=feature_foo_rules
        self._feature_bar_rules=feature_bar_rules
        self.enabled=enabled

    @property
    def matching_criteria(self):
        """
        the criteria that must match so to apply the policy
        """
        return self._matching_criteria;

    @matching_criteria.setter
    def matching_criteria(self, matching_criteria):
        if (len(matching_criteria) == 0) :
            raise ValueError("'matching_criteria' cannot be empty")
        self._matching_criteria = matching_criteria

    @property
    def feature_foo_rules(self):
        """
        a list of Rule objects for the foo feature
        """
        return self._feature_foo_rules;

    @feature_foo_rules.setter
    def feature_foo_rules(self, feature_foo_rules):
        self._feature_foo_rules = feature_foo_rules

    @property
    def feature_bar_rules(self):
        """
        a list of Rule objects for the foo feature
        """
        return self._feature_bar_rules;

    @feature_bar_rules.setter
    def feature_bar_rules(self, feature_bar_rules):
        self._feature_bar_rules = feature_bar_rules

    @property
    def enabled(self):
        """
        a boolean to toggle the policy on or off
        """
        return self._enabled;

    @enabled.setter
    def enabled(self, enabled):
        if( isinstance(enabled, bool) ):
            self._enabled = enabled
        else:
            if not enabled.lower() in ("yes", "true", "t", "1", "no", "false",
                "f", "0") :
                raise ValueError("'enabled' must be be any of 'yes', 'true', "\
                    "'t', '1', 'no', 'false', 'f', '0'")
            if enabled.lower() in ("yes", "true", "t", "1") :
                self._enabled = True
            else:
                self._enabled = False

    def toJSON(self):
        """
        returns a JSON serialization of the Policy object
        """
        return dict(matching_criteria=self._matching_criteria,
            enabled=self._enabled,
            feature_foo_rules=self._feature_foo_rules.toJSON(),
            feature_bar_rules=self._feature_bar_rules.toJSON())

    def __getstate__(self):
        """
        returns the state of the Policy object
        """
        return dict(matching_criteria=self._matching_criteria,
            feature_foo_rules=self._feature_foo_rules,
            feature_bar_rules=self._feature_bar_rules, enabled=self._enabled)

def policy_constructor(loader: yaml.SafeLoader,
    node: yaml.nodes.MappingNode) -> Policy:
    """ 
    constructs a Policy object. Can be added to yaml.loader
    """
    try:
        return Policy(**loader.construct_mapping(node))
    except TypeError as e:
        print(e, file=sys.stderr)
        exit(1)

A few comments on the above code:

• setup Python standard logging facility (lines 6-7)
• initialise the Policy class (lines 17-31)
• implement setters and getters methods (lines 33-87) using decorators
• implement the JSON serialization method (lines 89-96)
• implement the "__getstate__" dundee (lines 98-104)
• implement the YAML constructor (lines 106-115)

now add to the "__init__.py" file in the "src/carcano/fooapp/models" directory the necessary code to load the Policy class:

from .policy import Policy

Monitoring object

The purpose of the Monitoring object is storing monitoring specific settings, such as to describe how to connect the log aggregator service.

Create the "src/carcano/fooapp/models/monitoring.py" file with the following contents:

__author__ = "Marco Antonio Carcano"
__version__ = '0.1.0'

import yaml, sys, logging

log = logging.getLogger(__name__)
log.addHandler(logging.NullHandler())

class Monitoring(yaml.YAMLObject):
    """ Monitoring object """

    yaml_tag = u'!Monitoring'

    def __init__(self, driver, endpoint, token=None):
        """
        Initialization

        :param driver:       a string with the driver to use for the monitoring
                             facility
        :param endpoint:     the URL of the endpoint of the monitoring service
        :param token:        an optional authentication token to get access
                             to the monitoring service (if necessary)
        """
        self.driver = driver
        self.endpoint = endpoint
        self._token = token

    @property
    def driver(self):
        """
        a string with the driver to use for the monitoring
        """
        return self._driver;

    @driver.setter
    def driver(self, driver):
        if (len(driver) == 0) :
            raise ValueError("'driver' cannot be empty")
        self._driver = driver

    @property
    def endpoint(self):
        """
        the URL of the endpoint of the monitoring service
        """
        return self._endpoint;

    @endpoint.setter
    def endpoint(self, endpoint):
        if (len(endpoint) == 0) :
            raise ValueError("'endpoint' cannot be empty")
        self._endpoint = endpoint

    @property
    def token(self):
        """
        an optional authentication token to get access to the monitoring
        service (if necessary)
        """
        return self._token;

    @token.setter
    def token(self, token):
        if (len(token) == 0) :
            raise ValueError("'token' cannot be empty")
        self._token = token

    def toJSON(self):
        """
        returns a JSON serialization of the Monitoring object
        """
        return dict(driver=self._driver, endpoint=self._endpoint,
            token=self._token)

    def __getstate__(self):
        """
        returns the state of the Monitoring object
        """
        return dict(driver=self._driver, endpoint=self._endpoint,
            token=self._token)

def monitoring_constructor(loader: yaml.SafeLoader,
    node: yaml.nodes.MappingNode) -> Monitoring:
    """ 
    constructs a Monitoring object. Can be added to yaml.loader
    """
    try:
        return Monitoring(**loader.construct_mapping(node))
    except TypeError as e:
        print(e, file=sys.stderr)
        exit(1)

A few comments on the above code:

• setup Python standard logging facility (lines 6-7)
• initialise the Monitoring class (lines 14-26)
• implement setters and getters methods (lines 28-66) using decorators
• implement the JSON serialization method (lines 68-73)
• implement the "__getstate__" dundee (lines 75-80)
• implement the YAML constructor (lines 82-91)

now add to the "__init__.py" file in the "src/carcano/fooapp/models" directory the necessary code to load the Monitoring class:

from .monitoring import Monitoring

Service object

We eventually come to the Service class, that stores the main part of the YAML configuration file  of fooapp: it provides settings such as a list of addresses to bind to, the listening port and a lot of other settings.

Create the "src/carcano/fooapp/models/service.py" file with the following contents:

__author__ = "Marco Antonio Carcano"
__version__ = '0.1.0'

import yaml, sys, logging

log = logging.getLogger(__name__)
log.addHandler(logging.NullHandler())

from .monitoring import Monitoring

class Service(yaml.YAMLObject):
    """ Service object """

    yaml_tag = u'!Service'

    _policies = []
    _bind_addresses = []
    _tags = []
    _metadata = {}

    def __init__(self, name, tags=[], metadata={}, bind_addresses=None,
        bind_port=0, disabled="False", policies=[], monitoring=None):
        """
        Initialization

        :param name:            name of the service
        :param tags:            a list of tags
        :param metadata:        a dictionary of metadata
        :param bind_port:       the bind port
        :param disabled:        a boolean to toggle the service enabled or 
                                disabled
        :param monitoring:      an instance of the Monitoring object
        :param bind_addresses:  a list of addresses to bind the service to
        :param policy:          a list of Policy objects

        """
        self.name = name
        self.tags = tags
        self.metadata = metadata
        self.bind_port = bind_port
        self.disabled = disabled
        self.monitoring=monitoring
        if bind_addresses == None:
            self.bind_addresses = [ '127.0.0.1' ]
        else:
            self.bind_addresses = bind_addresses
        self.policies = policies

    def __del__(self):
        """
        destructor
        """
        self._policies.clear()
        self._bind_addresses.clear()
        self._tags.clear()

    @property
    def name(self):
        """
        name of the service
        """
        return self._name;

    @name.setter
    def name(self, name):
        if (len(name) == 0) :
            raise ValueError("'name' cannot be empty")
        self._name = name
    @property

    def bind_port(self):
        """
        the bind port
        """
        return self._bind_port;

    @bind_port.setter
    def bind_port(self, bind_port):
        if( not isinstance(bind_port, int) ):
            raise ValueError("'bind_port' must be a number")
        self._bind_port = bind_port

    @property
    def disabled(self):
        """
        a boolean to toggle the service enabled or disabled
        """
        return self._disabled;

    @disabled.setter
    def disabled(self, disabled):
        if( isinstance(disabled, bool) ):
            self._disabled = disabled
        else:
            if not disabled.lower() in ("yes", "true", "t", "1", "no",
                "false", "f", "0") :
                raise ValueError("'disabled' must be be any of 'yes', 'true',"\
                    " 't', '1', 'no', 'false', 'f', '0'")
            if disabled.lower() in ("yes", "true", "t", "1") :
                self._disabled = True
            else:
                self._disabled = False

    @property
    def monitoring(self):
        """
        an instance of the Monitoring object
        """
        return self._monitoring;

    @monitoring.setter
    def monitoring(self, monitoring):
        if monitoring!= None: 
            if( not isinstance(monitoring, Monitoring) ):
                raise ValueError("'monitoring' must be of kind " \
                    "'ServiceLevelAgreement'")
        self._monitoring = monitoring

    @property
    def bind_addresses(self):
        """
        a list of addresses to bind the service to
        """
        return self._bind_addresses

    @bind_addresses.setter
    def bind_addresses(self, bind_addresses):
        if not isinstance(bind_addresses, list):
            raise ValueError("'bind_addresses' must be of type 'list'")
        self._bind_addresses = bind_addresses

    @property
    def tags(self):
        """
        a list of tags
        """
        return self._tags

    @tags.setter
    def tags(self, tags):
        if not isinstance(tags, list):
            raise ValueError("'tags' must be of type 'list'")
        self._tags = tags

    @property
    def metadata(self):
        """
        a dictionary of metadata
        """
        return self._metadata

    @metadata.setter
    def metadata(self, metadata):
        if not isinstance(metadata, dict):
            raise ValueError("'metadata' must be of type 'dictionary'")
        self._metadata = metadata

    @property
    def policies(self):
        """
        a list of Policy objects
        """
        return self._policies;

    @policies.setter
    def policies(self, policies):
        if not isinstance(policies, list):
            raise ValueError("'policies' must be of type 'list'")
        self._policies = policies

    def toJSON(self):
        """
        returns a JSON serialization of the Service object
        """
        jpolicies = []
        for policies in self.policies:
            jpolicies.append(policies.toJSON())
        return dict(name=self._name, disabled=self._disabled,
            bind_port=self._bind_port, monitoring=self._monitoring.toJSON(),
            bind_addresses=self._bind_addresses, policies=jpolicies,
            tags=self._tags, metadata=self._metadata)

    def __getstate__(self):
        return dict(name=self._name, bind_port=self._bind_port,
            disabled=self._disabled, monitoring=self._monitoring,
            bind_addresses=self._bind_addresses, policies=self._policies,
            tags=self._tags, metadata=self._metadata)
        
def service_constructor(loader: yaml.SafeLoader,
    node: yaml.nodes.MappingNode) -> Service:
    """Construct a Service."""
    try:
        return Service(**loader.construct_mapping(node))
    except TypeError as e:
        print(e, file=sys.stderr)
        exit(1)

A few comments on the above code:

• setup Python standard logging facility (lines 6-7)
• initialise the Service class (lines 21-47)
• implement a destructor to free the list when the instance of the Service object is not used anymore (lines 49-55)
• implement setters and getters methods (lines 57-169) using decorators
• implement the JSON serialization method (lines 171-181)
• implement the "__getstate__" dundee (lines 183-187)
• implement the YAML constructor (lines 189-196)
• handle the TypeError exception (line 194-195)

now add to the "__init__.py" file in the "src/carcano/fooapp/models" directory the necessary code to load the Service class:

from .service import Service

eventually  add to the "__init__.py" file in the "src/carcano/fooapp" directory the necessary code to load the contents of the "models" package:

from .models import *

YAML Loader Function

Now that every model has been properly developed, deriving each object from the YAML Object, we need to define a loader function that adds to the YAML loader each constructor of the models we developed so far.

Are you enjoying these high quality free contents on a blog without annoying banners? I like doing this for free, but I also have costs so, if you like these contents and you want to help keeping this website free as it is now, please put your tip in the cup below:

Even a small contribution is always welcome!

Create "src/carcano/fooapp/yaml.py" file with the following contents:

__author__ = "Marco Antonio Carcano"
__version__ = '0.1.0'

from .models.grant import grant_constructor
from .models.rule import rule_constructor
from .models.policy import policy_constructor
from .models.monitoring import monitoring_constructor
from .models.service import service_constructor

def loader(loader=None):
    """
    This function can be passed as "Loader" argument to yaml.load to have
    objects automatically instantiated from a YAML document.
    
    The object type is guessed relying on YAML TAGs, enabling to select the
    proper constructor.

    :param loader: the YAML loader to use. If omitted, it defaults to
                    yaml.SafeLoader
    
    :returns:      the instantiated object 

    Example code snippet:
    
        from carcano.fooapp.models import *
        from carcano.fooapp.yaml import loader
    
        manifest=yaml.load(open(manifest_file_path, "rb"), 
            Loader=loader(yaml.SafeLoader))
    
    """
    if loader == None:
        loader = yaml.SafeLoader
    loader.add_constructor("!Grant", grant_constructor)
    loader.add_constructor("!Rule", rule_constructor)
    loader.add_constructor("!Policy", policy_constructor)
    loader.add_constructor("!Service", service_constructor)
    loader.add_constructor("!Monitoring", monitoring_constructor)
    return loader

There's not much to comment on the above code: we are just adding the constructors of each of the classes we previously defined to the "yaml.Safeloader" object.

Implement The Core

Now that we developed of the necessary models, we can start developing the core of our application - we group them together beneath the "core" directory:

mkdir -p src/carcano/fooapp/core

and of course we must create an empty "__init__.py" file beneath "src/carcano/fooapp/core" directory:

# __init__.py
__author__ = "Marco Antonio Carcano"
__version__ = '1.0.0'

Service Configuration Function

Now we have everything we need to develop a function that loads the configuration of the "fooapp" application from a YAML manifest file, automatically instantiating classes relying on the YAML tags found.

Create "src/carcano/fooapp/core/configuration.py" file with the following contents:

__author__ = "Marco Antonio Carcano"
__version__ = '0.1.0'

import sys, yaml
from pathlib import Path
from carcano.fooapp.yaml import loader

def load_config(manifest_file_path):
    """
    load the configuration of fooapp from the specified file

    :param manifest_file_path: path to the YAML manifest with the configuration

    :returns:                  the YAML document with the configuration
    """
    try:
        return yaml.load(open(manifest_file_path, "rb"),
            Loader=loader(yaml.SafeLoader))
    except FileNotFoundError:
        print("File {} not found".format(manifest_file_path),
             file=sys.stderr)
        exit(1)
    except PermissionError:
        print("Insufficient permission to read {}".format(manifest_file_path),
            file=sys.stderr)
        exit(1)
    except IsADirectoryError:
        print("{} is a directory".format(manifest_file_path), file=sys.stderr)

A few comments on the above code:

• open the specified file and load its contents using "yaml.load" (lines 17-18)
• handle exceptions that can happen while loading the file (lines 19-22)

now add to the "__init__.py" file in the "src/carcano/fooapp/core" directory the necessary code to load every function in the module:

from .configuration import *

eventually  add to the "__init__.py" file in the "src/carcano/fooapp" directory the necessary code to load the contents of the "core" package:

from .core import *

The Driver Program

We finally have everything is needed to develop the actual "fooapp" application. The best practice is putting it into the "src/bin" directory, so let's create and change into it:

mkdir src/bin
cd src/bin

since the application needs to import the Python package we have just developed, to avoid explicitly set the module search path, we can just create a symbolic link to the directory containing the package:

ln -s ../carcano carcano

finally we create the "src/bin/fooapp.py" file with the following contents:

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import yaml, sys, os, argparse, logging, logging.config

from carcano.fooapp import *

from flask import Flask

app = Flask(__name__)

config=None

def meta():
    return { 'metadata': config.metadata, 'tags': config.tags}

@app.route("/api/v1/meta", methods=['GET'])
def get_meta():
    logging.info('get_health(): processing new request')
    return meta(), 200

if __name__ == "__main__":
    from waitress import serve

    current_version='0.1.0'
    config_file_path='./config.yml'
    logconfig_file_path='./logging.conf'

    parser = argparse.ArgumentParser(description='fooapp service')
    parser.add_argument("-c", "--config-file", dest="config_file_path", help="path to the config file - defaults to '{}'".format(config_file_path), default=config_file_path)
    parser.add_argument("-l", "--logconfig", dest="logconfig_file_path", help="path to the configuration file of the Python logging facility - defaults to '{}'".format(logconfig_file_path), default=logconfig_file_path)
    parser.add_argument("-v", "--version", help="print the version and exit", action="store_true")

    args = parser.parse_args()
    
    if args.version:
        print(current_version)
        exit(0)
    if args.config_file_path:
        config_file_path=args.config_file_path
    if args.logconfig_file_path:
        logconfig_file_path=args.logconfig_file_path
    
    log_enabled = False
    if os.path.isfile(logconfig_file_path):
        log_enabled = True
    if log_enabled is True:
        logging.config.fileConfig(fname=logconfig_file_path, disable_existing_loggers=False)
        logger = logging.getLogger(__name__)
        logging.info('{}: started'.format(os.path.basename(__file__)))
    
    config=load_config(config_file_path)

    try:
       serve(app, host=config.bind_addresses[0],port=config.bind_port)
    except OSError as e:
        if(e.errno == 13):
            print("Unable to open the specified port - please try with a port higher than 1024", file=sys.stderr)
        else:
             print(e, file=sys.stderr)
    if log_enabled is True:
        logging.info('{}: finished'.format(os.path.basename(__file__)))

In the above code, the YAML file gets loaded and instantiated as "config" using the specific classes guessed from the YAML tags. If you want to dump the instantiated object to have a look to its contents, just use the following statement:

yaml.dump(config, sys.stdout, default_flow_style=False)

A few comments on the above code:

• initialise the Python standard logging facility
• instantiate "argparse" to parse command line arguments
• load the configuration from the YAML file
• instantiate Flask using "waitress"
• add the route "/api/vi/meta" to Flask and bind it to the "meta" function

of course, in order to be able to run it we must grant the "fooapp.py" file execution permission first:

chmod 755 fooapp.py

since the application supports the Python standard logging facility, create the "src/bin/logging.conf" configuration file with the following contents:

[loggers]
keys=root,fooapp

[handlers]
keys=console,syslog,file

[formatters]
keys=jsonFormatter

[logger_root]
level=INFO
handlers=file
propagate=0

[logger_fooapp]
level=INFO
handlers=file
qualname=carcano.*
propagate=0

[handler_null]
class = logging.NullHandler
formatter = default
args = ()

[handler_console]
class=StreamHandler
level=DEBUG
formatter=jsonFormatter
args=(sys.stderr,)

[handler_syslog]
class = handlers.SysLogHandler
args = ('/dev/log', handlers.SysLogHandler.LOG_USER)
level=DEBUG
formatter=jsonFormatter

[handler_rotatingfile]
class=handlers.RotatingFileHandler
level=DEBUG
formatter=jsonFormatter
args=('fooapp.log','a',1024,50)

[handler_file]
class=FileHandler
level=DEBUG
formatter=jsonFormatter
args=('fooapp.log','a')

[formatter_jsonFormatter]
format={"time": "%(asctime)s", "logger": "%(name)s", "level": "%(levelname)s", "pid": "%(process)d", "src": "%(pathname)s", "line": "%(lineno)d", "msg": "%(message)s"}

we are still missing the configuration file of "fooapp" itself - create the "src/bin/config.yml" file with the following contents:

!Service
name: foo service
bind_port: 8080
tags:
  - 24x7
  - foo_suite
metadata:
  service_owner: Marco Carcano
  cost_centre: Digital
monitoring:
  !Monitoring
  driver: splunk
  endpoint: https://splunk.carcano.local/services/collector/event
  token: gdr-573-93619-yjdweod
policies:
  - !Policy
    matching_criteria: .*
    feature_foo_rules:
      - !Rule
        name: MyRule
        description: boh
        grants:
        - !Grant
          access_level: 20
          description: Maintainers
          groups: []
          users: []

Mind that there are requirements that needs to be satisfied before being able to run "fooapp," such as install PyYAML, Flask and waitress.  To install them, just type:

pip3 install flask
pip3 install waitress
pip3 install pyyaml

we can finally have a go as follows:

./fooapp.py

to see it working, open http://127.0.0.1:8080/api/v1/meta in your browser - you must get the following reply:

{"metadata":{"cost_centre":"Digital","service_owner":"Marco Carcano"},"tags":["24x7","foo_suite"]}

Footnotes

Here it ends this post dedicated to showing how to serialize Python objects as either JSON or YAML: as you see it is something that can be very easily accomplished with a small effort, but with an impressive outcome. In the hope of making the post more interesting, and so to play with something that is not purely theoretical, the post also shows how to use the serializable objects developed as the configuration source for a Flask application.

I hate blogs with pop-ups, ads and all the (even worse) other stuff that distracts from the topics you're reading and violates your privacy. I want to offer my readers the best experience possible for free, ... but please be wary that for me it's not really free: on top of the raw costs of running the blog, I usually spend on average 50-60 hours writing each post. I offer all this for free because I think it's nice to help people, but if you think something in this blog has helped you professionally and you want to give concrete support, your contribution is very much appreciated: you can just use the above button.