What would you improve at the following code, except:
- The problems I'm already aware of, marked as
TODO - Line width of at most 120 characters (I've run flake8 over it and it's fine)
- Using decorators (I know about them, they don't fulfill my requirements in terms of how new commands should be written)
I'm using Python 3.5 and I have control over the environment where this code will be run.
The code is just a POC, and it will be part of a bigger open-source project, which has tests and code coverage set up. Sample usage is at the bottom of the file.
#!/usr/bin/python
# GistID: 7a0503430cb56a118c4124789e3a4b62
"""
Example of:
* some python magic
* the command pattern
* the factory pattern
* dependency injection
* lazy initialization
Only covers a constrained scenario where you can impose some characteristics of
the code, like for instance inheriting the base class `Command` or type hinting
each command's constructor.
"""
import inspect
import collections
class SubclassContainer(object):
"""
This data descriptor protects the injected property by `SubclassInjector`.
This way, neighbours in the inheritance tree do not know about each other.
"""
def __init__(self, topBaseClass):
self.value = {}
self.topBaseClass = topBaseClass
def __get__(self, instance, cls):
if cls is self.topBaseClass and instance is None:
return self.value
return {}
def __set__(self, instance, value):
self.value = value
def __delete__(self, instance):
pass
class SubclassInjector(type):
"""
When set as a metaclass of another class C, it injects the property
C.SUBCLASSES into that class.
You will want this for some abstract base classes. Other than that, this
metaclass doesn't do a lot.
It is useful in situations when you might want to track which classes
inherit a base class, for instance, for injecting trusted plugins into
a domain-specific factory.
"""
FIELD_NAME = 'SUBCLASSES'
def __init__(cls, name, bases, attrs):
type.__init__(cls, name, bases, attrs)
topBaseClassMetaclass = SubclassInjector
topBaseClass = object
for base in inspect.getmro(cls):
if not (topBaseClassMetaclass is base.__class__):
break
topBaseClassMetaclass = base.__class__
topBaseClass = base
if not hasattr(topBaseClass, cls.FIELD_NAME):
topBaseClass.SUBCLASSES = SubclassContainer(topBaseClass)
if topBaseClass is not cls:
topBaseClass.SUBCLASSES[name] = cls
class Command(object, metaclass=SubclassInjector):
pass
class FactoryException(Exception):
pass
class CommandFactory(object):
"""
Is able to create commands based on command's NAME
Requirements:
* the command inherits `Command`
* the command has a string property `NAME`
* the `__init__` method uses type hinting; the first parameter (`self`) is
ignored
When a command has a conflicting `NAME`, it replaces the previously
registered command.
"""
namesMap = {}
baseclass = None
# TODO: instead of names as keys, also add a hash over the params used for __init__
commandInstances = {}
initCanonicalForms = {}
def __init__(self, baseclass, services=[]):
"""
Creates Command instances from their names
This factory is also able to do dependency injection.
It is able to inject other known commands and services.
See help(Command) for requirements for new commands.
"""
self.namesMap = self.__getNamesMap(baseclass)
self.baseclass = baseclass
self.services = services
def create(self, name, *args, **kwargs):
"""
Create a new command with the name `name`, feeding it with its required
dependencies `args` and `kwargs`.
For help with those arguments, see help(DesiredCommand).
"""
self.namesMap = self.__getNamesMap(self.baseclass)
if name not in self.namesMap:
raise FactoryException("No command with NAME = '{}' known"
"Known names: {}".
format(name, list(self.namesMap.keys())))
wantedClass = self.namesMap[name]
classname = wantedClass.__name__
if classname in self.commandInstances:
return self.commandInstances[classname]
(args, kwargs) = self.__mergeParams(wantedClass, args, kwargs)
wantedObject = wantedClass(*args, **kwargs)
if len(args) == 0 and len(kwargs) == 0:
self.commandInstances[classname] = wantedObject
return wantedObject
def __getNamesMap(self, baseclass):
if not hasattr(baseclass, SubclassInjector.FIELD_NAME):
raise AttributeError("Class {} cannot be created automatically "
"because it does not have the attributed {}".
format(baseclass, SubclassInjector.FIELD_NAME))
namesMap = {}
for classname, subcls in baseclass.SUBCLASSES.items():
if not hasattr(subcls, 'NAME'):
raise AttributeError("Class {} does not have the attribute "
"NAME".format(classname))
if not hasattr(subcls, 'execute'):
raise AttributeError("Class {} does not have the execute method".
format(classname))
namesMap[subcls.NAME] = subcls
return namesMap
def __mergeParams(self, wantedClass, overrideArgs, overrideKwargs):
"""
Merge parameters from different sources, preparing the __init__ call.
The primary source of parameters is overrideArgs and overrideKwargs.
If one parameter in one of these lists has the default value or the
value None, then it's ignored, otherwise it's used.
If a parameter is a command but it is missing completely, the factory
will create a new object and inject it as necessary, provided the
dependency is known to the factory.
Sources of dependees are, in order of priority:
* explicitly named kwargs by the caller
* explicitly named args by the caller
* other known commands
* services known by the factory itself at construction time
"""
finalArgs = []
finalKwargs = {}
# TODO: try catch
sources = {
"canon": self.__getParamsCanonicalForm(wantedClass),
"kwargs": self.__normalizeParamsFromKwargs(wantedClass, overrideKwargs),
"args": self.__normalizeParamsFromArgs(wantedClass, overrideArgs),
"known": self.__normalizeParamsFromKnownInstances(wantedClass, self.commandInstances),
"classes": self.__normalizeParamsFromKnownClasses(wantedClass, Command.SUBCLASSES),
"services": self.__normalizeParamsFromServices(wantedClass, self.services),
}
for paramname, paramval in sources["canon"].items():
# TODO: try catch, chain exceptions
kwargs = sources["kwargs"][paramname]
args = sources["args"][paramname]
known = sources["known"][paramname]
classes = sources["classes"][paramname]
services = sources["services"][paramname]
candidates = [kwargs, args, known, classes, services]
candidate = (el for el in candidates if el is not None)
candidate = next(candidate, None)
if candidate is None:
initMethod = self.__getInitMethod(wantedClass)
initSignature = inspect.signature(initMethod)
raise FactoryException("Could not find a candidate for param {} of method {}{}".
format(paramname, wantedClass, initSignature))
finalArgs.append(candidate)
return (finalArgs, finalKwargs)
def __getParamsCanonicalForm(self, wantedClass):
if wantedClass in self.initCanonicalForms:
return self.initCanonicalForms[wantedClass]
canonicalForm = collections.OrderedDict()
initMethod = self.__getInitMethod(wantedClass)
if initMethod:
initSignature = inspect.signature(initMethod)
iterations = 0
for paramname, paramval in initSignature.parameters.items():
if iterations == 0:
iterations = 1
continue
iterations = 1
if paramval.annotation is paramval.empty:
raise FactoryException("parameter {} for method {}{} is not annotated".
format(paramname, wantedClass, initSignature))
if paramval.kind == paramval.POSITIONAL_OR_KEYWORD:
canonicalForm[paramname] = paramval
else:
raise FactoryException("Unhandled param type {} for method {}{}".
format(paramval.kind, wantedClass, initSignature))
self.initCanonicalForms[wantedClass] = canonicalForm
return canonicalForm
def __normalizeParamsFromKwargs(self, wantedClass, kwargs):
params = collections.OrderedDict()
canonicalForm = self.__getParamsCanonicalForm(wantedClass)
for paramname, paramval in canonicalForm.items():
if paramname in kwargs and type(kwargs[paramname]) is paramval.annotation:
params[paramname] = kwargs[paramname]
else:
params[paramname] = None
return params
def __normalizeParamsFromArgs(self, wantedClass, args):
params = collections.OrderedDict()
canonicalForm = self.__getParamsCanonicalForm(wantedClass)
argsIndex = 0
for paramname, paramval in canonicalForm.items():
if argsIndex < len(args) and type(args[argsIndex]) is paramval.annotation:
params[paramname] = args[argsIndex]
else:
params[paramname] = None
argsIndex += 1
return params
def __normalizeParamsFromKnownInstances(self, wantedClass, knownInstances):
params = collections.OrderedDict()
canonicalForm = self.__getParamsCanonicalForm(wantedClass)
for paramname, paramval in canonicalForm.items():
paramtype = paramval.annotation.__name__
if paramtype in knownInstances:
params[paramname] = knownInstances[paramtype]
else:
params[paramname] = None
return params
def __normalizeParamsFromKnownClasses(self, wantedClass, knownClasses):
params = collections.OrderedDict()
canonicalForm = self.__getParamsCanonicalForm(wantedClass)
for paramname, paramval in canonicalForm.items():
paramtype = paramval.annotation.__name__
if paramtype in knownClasses:
params[paramname] = knownClasses[paramtype]
else:
params[paramname] = None
return params
def __normalizeParamsFromServices(self, wantedClass, services):
serviceClasses = {type(s): s for s in services}
params = collections.OrderedDict()
canonicalForm = self.__getParamsCanonicalForm(wantedClass)
for paramname, paramval in canonicalForm.items():
if paramval.annotation in serviceClasses:
params[paramname] = serviceClasses[paramval.annotation]
else:
params[paramname] = None
return params
def __getInitMethod(self, wantedClass):
methods = inspect.getmembers(wantedClass)
initMethod = None
for methodname, method in methods:
if methodname == '__dict__' and '__init__' in method:
initMethod = method['__init__']
break
return initMethod
class PrintService(object):
def print(self, message):
print(message)
class FooCommand(Command):
NAME = 'foo'
def execute(self):
pass
class BarCommand(Command):
NAME = 'bar'
def execute(self):
pass
class FooBarCommand(FooCommand):
NAME = 'foobar'
def __init__(self, f: FooCommand, b: BarCommand, p: PrintService):
print('FOOBAR says:', f, b, p)
def execute(self):
pass
printService = PrintService()
factory = CommandFactory(Command, [printService])
# foobar = factory.create('foobar')
foo = factory.create('foo')
bar = factory.create('bar')
foobar = factory.create('foobar', foo, bar)
foobar = factory.create('foobar', None, bar)
foobar = factory.create('foobar', foo, None)
foobar = factory.create('foobar', b=bar)
foobar = factory.create('foobar', p=printService)
foobar = factory.create("test")
# help(CommandFactory)
# help(FooBarCommand)
