Serialization: JSON (for now) [3]

Today's post is almost entirely code, after a very brief run-down of what a typical implementation-pattern of serialization functionality looks like.

IsJSONSerializable Subclassing Basics

Because Python is a dynamic programming language, abstraction requirements aren't checked, and so won't fail, until the code that is affected is executed. That is, if you define a class that derives from another class that has abstract properties or methods, but don't implement those abstract members in the derived class, nothing will happen until your code tries to make an instance of the derived class. For example if the following code were executed:

#!/usr/bin/env python
"""bad-class-demo.py: Demonstrates an abstract-class requirement that fails 
because those requirements aren't implemented."""

from serialization import IsJSONSerializable

class MySerializableClass( IsJSONSerializable ):
    pass

instance = MySerializableClass()

It yields this:

Traceback (most recent call last):
  File "bad-class-demo.py", line 10, in <>module>
    instance = MySerializableClass()
TypeError: Can't instantiate abstract class 
  MySerializableClass with abstract methods 
  GetSerializationDict

As soon as the required method exists, even if there is no real implementation, that error will go away, but this implementation is still not complete.

The bare minimum of what's necessary to successfully implement a complete subclass of IsJSONSerializable boils down to:

• import the IsJSONSerializable abstract class in whatever fashion makes sense in the context of the code;
• Define a class that is derived from IsJSONSerializable;
• Make sure that the defined class provides a non-empty _jsonPublicFields class-attribute;
• Make sure to call IsJSONSerializable.__init__ in the __init__ of the defined class (this is not technically necessary at this point, since there isn't anything happening in IsJSONSerializable.__init__ that, for example, sets default properties, but it's a good habit to get into, I think);
• Implement the GetSerializationDict method — the stub-code below feels like a good starting-point to me, but you may prefer to take a different approach;
• Implement the FromDict class-method; and
• Register the defined class as JSON-loadable with [ClassName].RegisterLoadable().

All together, that looks like this:

from serialization import IsJSONSerializable

class JSONSerializableStub( IsJSONSerializable ):

  _jsonPublicFields = ( 'FieldName',  )

  def __init__( self ):
    IsJSONSerializable.__init__( self )
    # TODO: Whatever other initialization needs to happen

  def GetSerializationDict( self, sanitize=False ):
      # TODO: Actually implement code that reads the instance state 
      # and drops it into the "result" dict
      result = {}
      if sanitize:
        return self.SanitizeDict( result )
      return result

  @classmethod
  def FromDict( cls, data={}, **properties ):
    # Merge the properties and the data
    data.update( properties )
    # Create the instance to be returned
    result = cls()
    # TODO: Populate the instance with state-data from "data"
    # Return the instance
    return result

JSONSerializableStub.RegisterLoadable()

A Deeper Model

I'm going to build out some classes that model a hypothetical application that uses JSON-serialized files to store configuration-data. That Application has both simple configuration-values (strings only in this case, but numbers and booleans would also be feasible) and some complex values that are represented by instances of classes aggregated into the core Application structure. The example configuration-file that I'll be starting with looks like this:

{
   "Name":"Application",
   "UUID":"00000000-0000-0000-0000-000000000000",
   "Authentication":[
      {
         "Datastore":"ApplicationReader"
         "Type":"Database",
      },
      {
         "GUID":"The GUID of the provider",
         "Server":"ldap.company.com",
         "Type":"LDAP",
         "Name":"The name of the LDAP configuration",
         "NameFilter":"(&(objectClass=user)(objectCategory=Person))",
         "SearchBase":"DC=server,DC=company,DC=com"
      }
   ],
   "Datastores":{
      "ApplicationReader":{
         "Database":"ApplicationDatabase",
         "Host":"dbserver.company.com",
         "Password":"ApplicationReadOnlyUserPassword",
         "Type":"MySQL",
         "User":"ApplicationReadOnlyUser"
      },
      "ApplicationWriter":{
         "Database":"ApplicationDatabase",
         "Host":"dbserver.company.com",
         "Password":"ApplicationWriterUserPassword",
         "Type":"MySQL",
         "User":"ApplicationWriterUser"
      },
      "ExternalReadOnly":{
         "Database":"ExternalDatabase",
         "Host":"dbserver.external.com",
         "Password":"ExternalReadOnlyUserPassword",
         "Type":"ODBC",
         "User":"ExternalReadOnlyUser"
      }
   }
}

The entire JSON-object relates to a given Application instance, and its Authentication and Datastores properties to collections of instances of related classes whose relationship looks like this:

For simplicity, none of these classes have any methods, though I will probably build some out for various puropses as I work through the example, but I won't update this class-digram as I do so. Also, I won't go through my full template-based development cycle, since this example is pretty much throw-away code, only usable for demonstration purposes.

I'm going to demonstrate all of the functionality that IsJSONSerializable provides on this structure as follows:

1. Creating a complete instance of the Application class by reading it from the JSON configuration file above, using Application.FromJSON;
2. Showing that the entire application-structure is present by printing it (using a custom __str__ method that will render instances as an XML-like string);
3. Show that json.dumps executes against the nested-object instance as needed to serialize the entire object to JSON;
4. Show that the SanitizedJSON of the object strips out the fields that aren't listed in the class' _jsonPublicFields attribute;
5. Show that json.dump also executes as needed agains the nested-object structure by writing a file, then showing it's written JSON structure;
6. Use that output to unserialize an instance with json.load from the file; and
7. Do te same with the content of the file read in as a string, and processed with json.loads.

The code for all that is pretty simple:

#!/usr/bin/env python
"""app-demo.py: Demonstrates the functionality made available by 
IsJSONSerializable."""

import json
import os

from AppClasses import *

#------------------------------------------------------------------------------#
# Create an Application instance from the example.json file                    #
#------------------------------------------------------------------------------#
fp = open( 'example.json', 'r' )
appFromGenericFile = Application.FromJSON( fp.read() )
fp.close()

#------------------------------------------------------------------------------#
# Show that the entire Application structure has been loaded by printing it    #
#------------------------------------------------------------------------------#
print ( '# -- Printing application structure ' ).ljust( 79, '-' ) + '#'
print appFromGenericFile

#------------------------------------------------------------------------------#
# Show that json.dumps works as expected                                       #
#------------------------------------------------------------------------------#
print '\n' + ( '# -- Printing json.dumps results ' ).ljust( 79, '-' ) + '#'
print json.dumps( appFromGenericFile, indent=2 )

#------------------------------------------------------------------------------#
# Show that SanitizedJSON works as expected                                    #
#------------------------------------------------------------------------------#
print '\n' + ( '# -- appFromGenericFile.SanitizedJSON ' ).ljust( 79, '-' ) + '#'
print appFromGenericFile.SanitizedJSON

#------------------------------------------------------------------------------#
# Show that json.dump works as expected                                        #
#------------------------------------------------------------------------------#
fp = 'aFGF.json'
print '\n' + ( '# -- Printing json.dump results to %s ' % fp ).ljust( 79, '-' ) + '#'
ofp = open( fp, 'w' )
json.dump( appFromGenericFile, ofp, indent=2 )
ofp.close()
ifp = open( fp, 'r' )
for line in ifp.read().split( '\n' ):
    print line
ifp.close()
# os.unlink( fp )

#------------------------------------------------------------------------------#
# Show that json.load works as expected                                        #
#------------------------------------------------------------------------------#
print '\n' + ( '# -- Printing json.load results from %s ' % fp ).ljust( 79, '-' ) + '#'
ifp = open( fp, 'r' )
secondApp = json.load( ifp )
ifp.close()
print secondApp

#------------------------------------------------------------------------------#
# Show that json.loads works as expected                                        #
#------------------------------------------------------------------------------#
print '\n' + ( '# -- Printing json.loads results from %s ' % fp ).ljust( 79, '-' ) + '#'
ifp = open( fp, 'r' )
rawJSON = ifp.read()
ifp.close()
thirdApp = json.loads( rawJSON )
print thirdApp

The implementation of the relevant classes:

#!/usr/bin/env python
"""AppClasses.py: The classes for demonstrating the functionality made 
available by IsJSONSerializable."""

import abc
import json
import uuid

import os, sys

from serialization import IsJSONSerializable

__all__ = []

class Application( IsJSONSerializable, object ):

  _jsonPublicFields = ( 'Authentication', 'Datastores', 'Name' )

  def __init__( self ):
    IsJSONSerializable.__init__( self )
    self.Authentication = []
    self.Datastores = {}
    self.Name = None
    self.UUID = str( uuid.uuid4() )

  def GetSerializationDict( self, sanitize=False ):
      # TODO: Actually implement code that reads the instance state 
      # and returns a dictionary representation of all of it.
      result = {
        'Authentication':[ 
          authenticator.GetSerializationDict( sanitize ) 
          for authenticator in self.Authentication
          ],
        'Datastores':dict(
          [
            ( dsname, self.Datastores[ dsname ].GetSerializationDict( sanitize ) )
            for dsname in self.Datastores
          ]
        ),
        'Name':self.Name,
        'UUID':self.UUID,
      }
      if sanitize:
        return self.SanitizeDict( result )
      return result

  def __str__( self ):
    result = '<Application uuid="%s" name="%s"' % ( self.UUID, self.Name )
    if self.Authentication or self.Datastores:
      result += '>\n'
      if self.Authentication:
        result += '  <Authentication>\n'
        for authenticator in self.Authentication:
          result += '    %s\n' % authenticator
        result += '  </Authentication>\n'
      if self.Datastores:
        result += '  <Datastores>\n'
        for datastore in self.Datastores:
          result += '    %s\n' % self.Datastores[ datastore ]
        result += '  <Datastores>\n'
      result += '</Application>'
    else:
      result += ' />'
    return result

  @classmethod
  def FromDict( cls, data={}, **properties ):
    # Merge the properties and the data
    data.update( properties )
    # Create the instance to be returned
    result = cls()
    # Populate the instance
    result.Name = data[ 'Name' ]
    result.UUID = data[ 'UUID' ]
    authenticators = data.get( 'Authentication' )
    if authenticators:
      for authenticator in authenticators:
        if authenticator[ 'Type' ] == 'Database':
          result.Authentication.append( 
            DatabaseAuthenticator.FromDict( authenticator )
          )
        if authenticator[ 'Type' ] == 'LDAP':
          result.Authentication.append( 
            LDAPAuthenticator.FromDict( authenticator )
          )
    datastores = data.get( 'Datastores' )
    if datastores:
      for dsname in datastores:
        datastore = datastores[ dsname ]
        if datastore[ 'Type' ] == 'MySQL':
          result.Datastores[ dsname ] = MySQLDatastore.FromDict( datastore )
        elif datastore[ 'Type' ] == 'ODBC':
          result.Datastores[ dsname ] = ODBCDatastore.FromDict( datastore )
    # Return the instance
    return result

Application.RegisterLoadable()

__all__.append( 'Application' )

class BaseAuthenticator( IsJSONSerializable, object ):
    __metaclass__ = abc.ABCMeta

    def __init__( self ):
        if self.__class__ == BaseAuthenticator:
            raise NotImplementedError( 'BaseAuthenticator is '
                'intended to be an abstract class, NOT to be '
                'instantiated.' )
        self.Type = None

__all__.append( 'BaseAuthenticator' )

class BaseDatastore( object ):

  __metaclass__ = abc.ABCMeta

  _jsonPublicFields = ( 'Database', )

  def __init__( self ):
    if self.__class__ == BaseDatastore:
      raise NotImplementedError( 'BaseDatastore is '
        'intended to be an abstract class, NOT to be '
        'instantiated.' )
    self.Database = None
    self.Host = None
    self.Password = None
    self.Type = None
    self.User = None

  def __str__( self ):
    return ( '<%s database="%s" host="%s" password="%s" '
      'type="%s" user="%s" />' % ( self.__class__.__name__, self.Database, 
      self.Host, self.Password, self.Type, self.User ) )

  def GetSerializationDict( self, sanitize=False ):
    result = {
        'Database':self.Database,
        'Host':self.Host,
        'Password':self.Password,
        'Type':self.Type,
        'User':self.User,
        }
    if sanitize:
      return self.SanitizeDict( result )
    return result

  @classmethod
  def FromDict( cls, data={}, **properties ):
    # Merge the properties and the data
    data.update( properties )
    result = cls()
    result.Database = data[ 'Database' ]
    result.Host = data[ 'Host' ]
    result.Password = data[ 'Password' ]
    result.Type = data[ 'Type' ]
    result.User = data[ 'User' ]
    return result

__all__.append( 'BaseDatastore' )

class DatabaseAuthenticator( BaseAuthenticator, object ):

  _jsonPublicFields = ( 'Datastore', 'Type' )

  def __init__( self ):
    BaseAuthenticator.__init__( self )
    self.Datastore = None
    self.Type = 'Database'

  def GetSerializationDict( self, sanitize=False ):
    result = {
        'Datastore':self.Datastore,
        'Type':self.Type,
        }
    if sanitize:
      return self.SanitizeDict( result )
    return result

  def __str__( self ):
    return ( '<%s datastore="%s" type="%s"  />' % ( 
        self.__class__.__name__, self.Datastore, self.Type ) )

  @classmethod
  def FromDict( cls, data={}, **properties ):
    # Merge the properties and the data
    data.update( properties )
    result = cls()
    result.Datastore = data[ 'Datastore' ]
    result.Type = 'Database'
    return result

DatabaseAuthenticator.RegisterLoadable()

__all__.append( 'DatabaseAuthenticator' )

class LDAPAuthenticator( BaseAuthenticator, object ):

  _jsonPublicFields = ( 'Name', 'Type' )

  def __init__( self ):
    BaseAuthenticator.__init__( self )
    self.GUID = None
    self.Server = None
    self.Type = 'LDAP'
    self.Name = None
    self.NameFilter = None
    self.SearchBase = None

  def GetSerializationDict( self, sanitize=False ):
    result = {
        'GUID':self.GUID,
        'Server':self.Server,
        'Type':self.Type,
        'Name':self.Name,
        'NameFilter':self.NameFilter,
        'SearchBase':self.SearchBase,
        }
    if sanitize:
      return self.SanitizeDict( result )
    return result

  def __str__( self ):
    return ( '<%s guid="%s" server="%s" type="%s" '
      'name="%s" namefilter="%s" searchbase="%s" />' % ( 
        self.__class__.__name__, self.GUID, self.Server, self.Type, 
        self.Name, self.NameFilter, self.SearchBase ) )

  @classmethod
  def FromDict( cls, data={}, **properties ):
    # Merge the properties and the data
    data.update( properties )
    result = cls()
    result.GUID = data[ 'GUID' ]
    result.Server = data[ 'Server' ]
    result.Type = 'LDAP'
    result.Name = data[ 'Name' ]
    result.NameFilter = data[ 'NameFilter' ]
    result.SearchBase = data[ 'SearchBase' ]
    return result

LDAPAuthenticator.RegisterLoadable()

__all__.append( 'LDAPAuthenticator' )

class MySQLDatastore( BaseDatastore, IsJSONSerializable, object ):

  def __init__( self ):
    BaseDatastore.__init__( self )
    IsJSONSerializable.__init__( self )

MySQLDatastore.RegisterLoadable()

__all__.append( 'MySQLDatastore' )

class ODBCDatastore( BaseDatastore, IsJSONSerializable, object ):

  def __init__( self ):
    BaseDatastore.__init__( self )
    IsJSONSerializable.__init__( self )

ODBCDatastore.RegisterLoadable()

__all__.append( 'ODBCDatastore' )

When this code is run, the _dumps call raises the warning:

app-demo.py:27: UnsanitizedJSONWarning: 
Application is an instance derived from IsJSONSerializable, 
and has "sanitized" JSON available in its SanitizedJSON property.
  print json.dumps( appFromGenericFile, indent=2 )

and generates the following output:

# -- Printing application structure -------------------------------------------#
<Application uuid="00000000-0000-0000-0000-000000000000" name="Application">
  <Authentication>
    <DatabaseAuthenticator datastore="ApplicationReader" type="Database"  />
    <LDAPAuthenticator guid="The GUID of the provider" server="ldap.company.com" type="LDAP" name="The name of the LDAP configuration" namefilter="(&(objectClass=user)(objectCategory=Person))" searchbase="DC=server,DC=company,DC=com" />
  </Authentication>
  <Datastores>
    <MySQLDatastore database="ApplicationDatabase" host="dbserver.company.com" password="ApplicationReadOnlyUserPassword" type="MySQL" user="ApplicationReadOnlyUser" />
    <MySQLDatastore database="ApplicationDatabase" host="dbserver.company.com" password="ApplicationWriterUserPassword" type="MySQL" user="ApplicationWriterUser" />
    <ODBCDatastore database="ExternalDatabase" host="dbserver.external.com" password="ExternalReadOnlyUserPassword" type="ODBC" user="ExternalReadOnlyUser" />
  <Datastores>
</Application>

# -- Printing json.dumps results ----------------------------------------------#
{
  "Datastores": {
    "ApplicationReader": {
      "Host": "dbserver.company.com", 
      "Password": "ApplicationReadOnlyUserPassword", 
      "Type": "MySQL", 
      "User": "ApplicationReadOnlyUser", 
      "Database": "ApplicationDatabase"
    }, 
    "ApplicationWriter": {
      "Host": "dbserver.company.com", 
      "Password": "ApplicationWriterUserPassword", 
      "Type": "MySQL", 
      "User": "ApplicationWriterUser", 
      "Database": "ApplicationDatabase"
    }, 
    "ExternalReadOnly": {
      "Host": "dbserver.external.com", 
      "Password": "ExternalReadOnlyUserPassword", 
      "Type": "ODBC", 
      "User": "ExternalReadOnlyUser", 
      "Database": "ExternalDatabase"
    }
  }, 
  "Authentication": [
    {
      "Datastore": "ApplicationReader", 
      "Type": "Database"
    }, 
    {
      "Name": "The name of the LDAP configuration", 
      "Server": "ldap.company.com", 
      "NameFilter": "(&(objectClass=user)(objectCategory=Person))", 
      "SearchBase": "DC=server,DC=company,DC=com", 
      "GUID": "The GUID of the provider", 
      "Type": "LDAP"
    }
  ], 
  "__namespace": "AppClasses.Application", 
  "Name": "Application", 
  "UUID": "00000000-0000-0000-0000-000000000000"
}

# -- appFromGenericFile.SanitizedJSON -----------------------------------------#
{
  "Datastores": {
    "ApplicationReader": {
      "Database": "ApplicationDatabase"
    }, 
    "ApplicationWriter": {
      "Database": "ApplicationDatabase"
    }, 
    "ExternalReadOnly": {
      "Database": "ExternalDatabase"
    }
  }, 
  "Authentication": [
    {
      "Datastore": "ApplicationReader", 
      "Type": "Database"
    }, 
    {
      "Type": "LDAP", 
      "Name": "The name of the LDAP configuration"
    }
  ], 
  "Name": "Application"
}

# -- Printing json.dump results to aFGF.json ----------------------------------#
{
  "Datastores": {
    "ApplicationReader": {
      "Host": "dbserver.company.com", 
      "Password": "ApplicationReadOnlyUserPassword", 
      "Type": "MySQL", 
      "User": "ApplicationReadOnlyUser", 
      "Database": "ApplicationDatabase"
    }, 
    "ApplicationWriter": {
      "Host": "dbserver.company.com", 
      "Password": "ApplicationWriterUserPassword", 
      "Type": "MySQL", 
      "User": "ApplicationWriterUser", 
      "Database": "ApplicationDatabase"
    }, 
    "ExternalReadOnly": {
      "Host": "dbserver.external.com", 
      "Password": "ExternalReadOnlyUserPassword", 
      "Type": "ODBC", 
      "User": "ExternalReadOnlyUser", 
      "Database": "ExternalDatabase"
    }
  }, 
  "Authentication": [
    {
      "Datastore": "ApplicationReader", 
      "Type": "Database"
    }, 
    {
      "Name": "The name of the LDAP configuration", 
      "Server": "ldap.company.com", 
      "NameFilter": "(&(objectClass=user)(objectCategory=Person))", 
      "SearchBase": "DC=server,DC=company,DC=com", 
      "GUID": "The GUID of the provider", 
      "Type": "LDAP"
    }
  ], 
  "__namespace": "AppClasses.Application", 
  "Name": "Application", 
  "UUID": "00000000-0000-0000-0000-000000000000"
}

# -- Printing json.load results from aFGF.json --------------------------------#
<Application uuid="00000000-0000-0000-0000-000000000000" name="Application">
  <Authentication>
    <DatabaseAuthenticator datastore="ApplicationReader" type="Database"  />
    <LDAPAuthenticator guid="The GUID of the provider" server="ldap.company.com" type="LDAP" name="The name of the LDAP configuration" namefilter="(&(objectClass=user)(objectCategory=Person))" searchbase="DC=server,DC=company,DC=com" />
  </Authentication>
  <Datastores>
    <MySQLDatastore database="ApplicationDatabase" host="dbserver.company.com" password="ApplicationReadOnlyUserPassword" type="MySQL" user="ApplicationReadOnlyUser" />
    <MySQLDatastore database="ApplicationDatabase" host="dbserver.company.com" password="ApplicationWriterUserPassword" type="MySQL" user="ApplicationWriterUser" />
    <ODBCDatastore database="ExternalDatabase" host="dbserver.external.com" password="ExternalReadOnlyUserPassword" type="ODBC" user="ExternalReadOnlyUser" />
  <Datastores>
</Application>

# -- Printing json.loads results from aFGF.json -------------------------------#
<Application uuid="00000000-0000-0000-0000-000000000000" name="Application">
  <Authentication>
    <DatabaseAuthenticator datastore="ApplicationReader" type="Database"  />
    <LDAPAuthenticator guid="The GUID of the provider" server="ldap.company.com" type="LDAP" name="The name of the LDAP configuration" namefilter="(&(objectClass=user)(objectCategory=Person))" searchbase="DC=server,DC=company,DC=com" />
  </Authentication>
  <Datastores>
    <MySQLDatastore database="ApplicationDatabase" host="dbserver.company.com" password="ApplicationReadOnlyUserPassword" type="MySQL" user="ApplicationReadOnlyUser" />
    <MySQLDatastore database="ApplicationDatabase" host="dbserver.company.com" password="ApplicationWriterUserPassword" type="MySQL" user="ApplicationWriterUser" />
    <ODBCDatastore database="ExternalDatabase" host="dbserver.external.com" password="ExternalReadOnlyUserPassword" type="ODBC" user="ExternalReadOnlyUser" />
  <Datastores>
</Application>

All of the code above, plus a stripped-down copy of serialization.py is in the download-package for today's post — Everything needed to run the code is all there.

serialization-demo.zip

90.2kB

This sort of testing — writing a program to show that pieces of functionality are doing what they're supposed to — is something that most developers do from time to time at least, I expect. It's often just easier to work through issues that arise when that sort of test-code is run. That said, it's not all that thorough a test-process, so starting in my next post, I'll dig in to some unit-testing thoughts and processes.