Clever Geek Handbook

Lie to the state

So this is kluge. I am working with luigi and sciluigi.

The check isinstance

is in the sciluigi package and I would prefer kluge, then you need to fork all the sciluigi for this little problem :)

Simply put, I had to subclass one of the package classes (luigi.LocalTarget) - add additional functionality. This functionality works great BUT there is an object check in the sciluigi package ...

sciluigi.dependencies.DependencyHelpers._parse_outputitem()

... which causes the run to fail simply because the string is isinstance

set to check for TargetInfo objects only.

What I would like to do is just tell my child class "false" isinstance

to represent an object TargetInfo

and pass: D

Forgiveness is asked in advance: D

def _parse_outputitem(self, val, targets):
    '''
    Recursively loop through lists of TargetInfos, or
    callables returning TargetInfos, or lists of ...
    (repeat recursively) ... and return all targets.
    '''
    if callable(val):
        val = val()
    if isinstance(val, TargetInfo):
        targets.append(val.target)
    elif isinstance(val, list):
        for valitem in val:
            targets = self._parse_outputitem(valitem, targets)
    elif isinstance(val, dict):
        for _, valitem in iteritems(val):
            targets = self._parse_outputitem(valitem, targets)
    else:
        raise Exception('Input item is neither callable, TargetInfo, nor list: %s' % val)
    return targets

Error message:

2017-04-06 22:26:09,753 - PipeineTest1 - DEBUG - RunSubprocess:Traceback (most recent call last):
2017-04-06 22:26:09,754 - PipeineTest1 - DEBUG - RunSubprocess:  File "/Library/Python/2.7/site-packages/luigi/worker.py", line 305, in check_complete
2017-04-06 22:26:09,754 - PipeineTest1 - DEBUG - RunSubprocess:    is_complete = task.complete()
2017-04-06 22:26:09,754 - PipeineTest1 - DEBUG - RunSubprocess:  File "/Library/Python/2.7/site-packages/luigi/task.py", line 482, in complete
2017-04-06 22:26:09,754 - PipeineTest1 - DEBUG - RunSubprocess:    outputs = flatten(self.output())
2017-04-06 22:26:09,754 - PipeineTest1 - DEBUG - RunSubprocess:  File "/Library/Python/2.7/site-packages/sciluigi/dependencies.py", line 99, in output
2017-04-06 22:26:09,754 - PipeineTest1 - DEBUG - RunSubprocess:    return self._output_targets()
2017-04-06 22:26:09,755 - PipeineTest1 - DEBUG - RunSubprocess:  File "/Library/Python/2.7/site-packages/sciluigi/dependencies.py", line 111, in _output_targets
2017-04-06 22:26:09,755 - PipeineTest1 - DEBUG - RunSubprocess:    output_targets = self._parse_outputitem(attrval, output_targets)
2017-04-06 22:26:09,755 - PipeineTest1 - DEBUG - RunSubprocess:  File "/Library/Python/2.7/site-packages/sciluigi/dependencies.py", line 132, in _parse_outputitem
2017-04-06 22:26:09,755 - PipeineTest1 - DEBUG - RunSubprocess:    raise Exception('Input item is neither callable, TargetInfo, nor list: %s' % val)
2017-04-06 22:26:09,755 - PipeineTest1 - DEBUG - RunSubprocess:Exception: Input item is neither callable, TargetInfo, nor list: <Bioproximity.common.luigi_extensions.local_target.ToppasLocalTarget object at 0x110e48190>

... Unfortunately, this is 100% of the error traces that Sciluigi provides as output.

sciluigi.dependencies.TargetInfo (object)

class TargetInfo(object):
    '''
    Class to be used for sending specification of which target, from which
    task, to use, when stitching workflow tasks' outputs and inputs together.
    '''
    task = None
    path = None
    target = None

    def __init__(self, task, path, format=None, is_tmp=False):
        self.task = task
        self.path = path
        self.target = luigi.LocalTarget(path, format, is_tmp)

    def open(self, *args, **kwargs):
        '''
        Forward open method, from luigi target class
        '''
        return self.target.open(*args, **kwargs)

# ==============================================================================
+3


source to share


2 answers


It looks like you just need a subclass TargetInfo

for your object to pass validation isinstance

. You can do it like this:

class Foo(<whatever other base classes you have>, TargetInfo):
    ...


If you have TargetInfo

a child as root, it should not interfere with the functionality of the class, because other base classes will override any conflicting methods.

+1


source


I think you need to create a subclass TargetInfo

in addition to the subclass LocalTarget

you already have. It looks like you are currently trying to use your current subclass as the first one when it is an instance of the latter. Passing an instance of your custom class doesn't work, because passing a regular LocalTarget

in the same place won't work either.

Try something like this:

class MyTargetInfo(TargetInfo):            # pick your own name
    def __init__(self, task, path, *args): # you might want to explicitly name the args here
        self.task = task
        self.path = path
        self.target = ToppasLocalTarget(*args)


You want to pass an instance of this class to the function that was giving you errors when you provided it with an instance of your subclass LocalTarget

. As I commented, you should give the class a better name, and perhaps name (and possibly give default values) the arguments to be passed to another class (instead of using *args

).

If constructing target

internally is MyTargetInfo.__init__

not suitable for your custom class needs (for example, you need to create it ahead of time or reuse the same instance multiple times), you can pass an already existing one LocalTarget

to the constructor and just assign it self.target

instead of creating a new object. I don't know enough about the library you are using to judge if this is a good idea or not.

+2


source






More articles:


All Articles