Source code for gavo.svcs.customcore

User-defined cores, either with an external file or with code within
the RD.

#c Copyright 2008-2023, the GAVO project <>
#c This program is free software, covered by the GNU GPL.  See the
#c COPYING file in the source distribution.

from gavo import base
from gavo import rsc
from gavo import rscdef
from gavo import utils
from gavo.svcs import core

[docs]class CustomCore(core.Core): """A wrapper around a core defined in a module. This core lets you write your own cores in modules. The module must define a class Core. When the custom core is encountered, this class will be instantiated and will be used instead of the CustomCore, so your code should probably inherit core.Core. See `Writing Custom Cores`_ for details. """ name_ = "customCore" _module = rscdef.ResdirRelativeAttribute("module", description="Path to the module containing the core definition.", copyable=True, default=base.Undefined)
[docs] def completeElement(self, ctx): overrideInput = self.inputTable is not base.NotGiven overrideOutput = self.outputTable is not base.NotGiven super().completeElement(ctx) try: loaded_module, _ = utils.loadPythonModule(self.module) except Exception as msg: raise base.StructureError("Cannot load custom core %s: %s"%( self.module, msg)) innerCore = base.makeStruct(loaded_module.Core, parent_=self.parent) = # let people overwrite inputTable and outputTable from the RD # (this may be a questionable feature, but there you go. if overrideInput: innerCore.inputTable = self.inputTable if overrideOutput: innerCore.outputTable = self.outputTable # and we let people override properties on the inner core, too. # now replace the "empty" core from the xml in the id map and in the # parent with our inner core if ctx.idmap.get( is self: ctx.idmap[] = innerCore self.parent.feedObject(self.name_, innerCore)
[docs] def onElementComplete(self): super().onElementComplete() # we have already added the inner core in completeElement, so we # should not leave any trace here. raise base.Ignore(self)
[docs]class CoreProc(rscdef.ProcApp): """A definition of a pythonCore's functionalty. This is a procApp complete with setup and code; you could inherit between these. coreProcs see the embedding service, the input table passed, and the query metadata as service, inputTable, and queryMeta, respectively. The core itself is available as self. """ name_ = "coreProc" requiredType = "coreProc" formalArgs = "self, service, inputTable, queryMeta" additionalNamesForProcs = { "rsc": rsc }
[docs]class PythonCore(core.Core): """A core doing computation using a piece of python. See `Python Cores instead of Custom Cores`_ in the reference. """ name_ = "pythonCore" _computer = base.StructAttribute("coreProc", default=base.Undefined, childFactory=CoreProc, description="Code making the outputTable from the inputTable.", copyable=True)
[docs] def expand(self, s): # macro expansion should ideally take place in the service, # but that's impossible in general because a core could be # in use by several services. Hence, we go ask the RD return self.rd.expand(s)
[docs] def run(self, service, inputTable, queryMeta): return self.coreProc.compile()(self, service, inputTable, queryMeta)