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()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:
importtheIsJSONSerializableabstract 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
_jsonPublicFieldsclass-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 inIsJSONSerializable.__init__that, for example, sets default properties, but it's a good habit to get into, I think); - Implement the
GetSerializationDictmethod — the stub-code below feels like a good starting-point to me, but you may prefer to take a different approach; - Implement the
FromDictclass-method; and - Register the defined class as JSON-loadable with
[ClassName].RegisterLoadable().
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:
I'm going to demonstrate all of the functionality that IsJSONSerializable
provides on this structure as follows:
- Creating a complete instance of the
Applicationclass by reading it from the JSONconfiguration
file above, usingApplication.FromJSON; - 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); - Show that
json.dumpsexecutes against the nested-object instance as needed to serialize the entire object to JSON; - Show that the
SanitizedJSONof the object strips out the fields that aren't listed in the class'_jsonPublicFieldsattribute; - Show that
json.dumpalso executes as needed agains the nested-object structure by writing a file, then showing it's written JSON structure; - Use that output to unserialize an instance with
json.loadfrom the file; and - Do te same with the content of the file read in as a string, and
processed with
json.loads.
#!/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#!/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' )_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.
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.
