/usr/share/pyshared/martian/core.py is in python-martian 0.14-0ubuntu1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 | import types, inspect
from zope.interface import implements
from martian.interfaces import IGrokker, IMultiGrokker
from martian import util, scan
from martian.components import (GrokkerBase, ClassGrokker, InstanceGrokker,
GlobalGrokker)
from martian.error import GrokError
from martian.martiandirective import component, priority
class MultiGrokkerBase(GrokkerBase):
implements(IMultiGrokker)
def register(self, grokker):
raise NotImplementedError
def grok(self, name, obj, **kw):
grokked_status = False
for g, name, obj in self.grokkers(name, obj):
grokked = g.grok(name, obj, **kw)
if grokked not in (True, False):
raise GrokError(
"%r returns %r instead of True or False." %
(g, grokked), None)
if grokked:
grokked_status = True
return grokked_status
def clear(self):
raise NotImplementedError
def grokkers(self, name, obj):
raise NotImplementedError
def _grokker_sort_key((grokker, name, obj)):
"""Helper function to calculate sort order of grokker.
"""
return priority.bind().get(grokker)
class ModuleGrokker(MultiGrokkerBase):
def __init__(self, grokker=None, prepare=None, finalize=None):
if grokker is None:
grokker = MultiGrokker()
self._grokker = grokker
self.prepare = prepare
self.finalize = finalize
def register(self, grokker):
self._grokker.register(grokker)
def clear(self):
self._grokker.clear()
def grok(self, name, module, **kw):
grokked_status = False
# prepare module grok - this can also influence the kw dictionary
if self.prepare is not None:
self.prepare(name, module, kw)
# sort grokkers by priority
grokkers = sorted(self.grokkers(name, module),
key=_grokker_sort_key,
reverse=True)
for g, name, obj in grokkers:
grokked = g.grok(name, obj, **kw)
if grokked not in (True, False):
raise GrokError(
"%r returns %r instead of True or False." %
(g, grokked), None)
if grokked:
grokked_status = True
# finalize module grok
if self.finalize is not None:
self.finalize(name, module, kw)
return grokked_status
def grokkers(self, name, module):
grokker = self._grokker
# get any global grokkers
for t in grokker.grokkers(name, module):
yield t
# try to grok everything in module
for name in dir(module):
if '.' in name:
# This must be a module-level variable that couldn't
# have been set by the developer. It must have been a
# module-level directive.
continue
obj = getattr(module, name)
if not util.defined_locally(obj, module.__name__):
continue
if util.is_baseclass(obj):
continue
for t in grokker.grokkers(name, obj):
yield t
class MultiInstanceOrClassGrokkerBase(MultiGrokkerBase):
def __init__(self):
self.clear()
def register(self, grokker):
key = component.bind().get(grokker)
grokkers = self._grokkers.setdefault(key, [])
for g in grokkers:
if g.__class__ is grokker.__class__:
return
grokkers.append(grokker)
def clear(self):
self._grokkers = {}
def grokkers(self, name, obj):
used_grokkers = set()
for base in self.get_bases(obj):
grokkers = self._grokkers.get(base)
if grokkers is None:
continue
for grokker in grokkers:
if grokker not in used_grokkers:
yield grokker, name, obj
used_grokkers.add(grokker)
class MultiInstanceGrokker(MultiInstanceOrClassGrokkerBase):
def get_bases(self, obj):
return inspect.getmro(obj.__class__)
class MultiClassGrokker(MultiInstanceOrClassGrokkerBase):
def get_bases(self, obj):
if type(obj) is types.ModuleType:
return []
return inspect.getmro(obj)
class MultiGlobalGrokker(MultiGrokkerBase):
def __init__(self):
self.clear()
def register(self, grokker):
for g in self._grokkers:
if grokker.__class__ is g.__class__:
return
self._grokkers.append(grokker)
def clear(self):
self._grokkers = []
def grokkers(self, name, module):
for grokker in self._grokkers:
yield grokker, name, module
class MultiGrokker(MultiGrokkerBase):
def __init__(self):
self.clear()
def register(self, grokker):
if isinstance(grokker, InstanceGrokker):
self._multi_instance_grokker.register(grokker)
elif isinstance(grokker, ClassGrokker):
self._multi_class_grokker.register(grokker)
elif isinstance(grokker, GlobalGrokker):
self._multi_global_grokker.register(grokker)
else:
assert 0, "Unknown type of grokker: %r" % grokker
def clear(self):
self._multi_instance_grokker = MultiInstanceGrokker()
self._multi_class_grokker = MultiClassGrokker()
self._multi_global_grokker = MultiGlobalGrokker()
def grokkers(self, name, obj):
if isinstance(obj, (type, types.ClassType)):
return self._multi_class_grokker.grokkers(name, obj)
elif isinstance(obj, types.ModuleType):
return self._multi_global_grokker.grokkers(name, obj)
else:
return self._multi_instance_grokker.grokkers(name, obj)
class MetaMultiGrokker(MultiGrokker):
"""Multi grokker which comes pre-registered with meta-grokkers.
"""
def clear(self):
super(MetaMultiGrokker, self).clear()
# bootstrap the meta-grokkers
self.register(ClassMetaGrokker(self))
self.register(InstanceMetaGrokker(self))
self.register(GlobalMetaGrokker(self))
def grok_dotted_name(dotted_name, grokker, exclude_filter=None,
ignore_nonsource=True, **kw):
module_info = scan.module_info_from_dotted_name(dotted_name, exclude_filter,
ignore_nonsource)
grok_package(module_info, grokker, **kw)
def grok_package(module_info, grokker, **kw):
grok_module(module_info, grokker, **kw)
for sub_module_info in module_info.getSubModuleInfos():
grok_package(sub_module_info, grokker, **kw)
def grok_module(module_info, grokker, **kw):
grokker.grok(module_info.dotted_name, module_info.getModule(),
module_info=module_info, **kw)
# deep meta mode here - we define grokkers that can pick up the
# three kinds of grokker: ClassGrokker, InstanceGrokker and ModuleGrokker
class MetaGrokker(ClassGrokker):
def __init__(self, multi_grokker):
"""multi_grokker - the grokker to register grokkers with.
"""
self.multi_grokker = multi_grokker
def grok(self, name, obj, **kw):
self.multi_grokker.register(obj())
return True
class ClassMetaGrokker(MetaGrokker):
component(ClassGrokker)
class InstanceMetaGrokker(MetaGrokker):
component(InstanceGrokker)
class GlobalMetaGrokker(MetaGrokker):
component(GlobalGrokker)
class GrokkerRegistry(ModuleGrokker):
def __init__(self):
super(GrokkerRegistry, self).__init__(MetaMultiGrokker())
|