taskdeps

#!/usr/bin/env python3
# coding: utf-8
# pylint: disable=F0401,C0111,W0232,E1101,E1103,C0301,W0614

# Copyright 2015 by Leipzig University Library, http://ub.uni-leipzig.de
#                   The Finc Authors, http://finc.info
#                   Martin Czygan, <martin.czygan@uni-leipzig.de>
#
# This file is part of some open source application.
#
# Some open source application is free software: you can redistribute
# it and/or modify it under the terms of the GNU General Public
# License as published by the Free Software Foundation, either
# version 3 of the License, or (at your option) any later version.
#
# Some open source application is distributed in the hope that it will
# be useful, but WITHOUT ANY WARRANTY; without even the implied warranty
# of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Foobar.  If not, see <http://www.gnu.org/licenses/>.
#
# @license GPL-3.0+ <http://spdx.org/licenses/GPL-3.0+>

"""
Usage: taskdeps TASKNAME

Example output:

$ taskdeps JstorIntermediateSchema
  └─ JstorIntermediateSchema(date=2018-05-03)
      └─ AMSLService(date=2018-05-03, name=outboundservices:discovery)
      └─ JstorCollectionMapping(date=2018-05-03)
      └─ JstorIntermediateSchemaGenericCollection(date=2018-05-03)
          └─ Executable(name=span-import, message=http://git.io/vI8NV)
          └─ JstorXML(date=2018-05-03)
              └─ JstorLatestMembers(date=2018-05-03, version=2)
                  └─ JstorMembers(date=2018-05-03)
                      └─ JstorPaths(date=2018-05-03)
                          └─ FTPMirror(host=example.com, ...)
                              └─ Executable(name=lftp, message=http://lftp.yar.ru/)

"""

from __future__ import print_function

import collections
import re
import sys
from io import StringIO

from luigi.cmdline_parser import CmdlineParser
from luigi.parameter import MissingParameterException
from luigi.task import Register
from luigi.task_register import TaskClassNotFoundException

from siskin.sources import *
from siskin.workflows import *

g = collections.defaultdict(set)  # node -> [deps] dictionary


def sanitize(s):
    """
    Slightly sanitize string.
    """
    s = re.sub(r"host=[^ ,]+", "host=example.com", s, 0)
    s = re.sub(r"username=[^ ,]+", "username=xxxx", s, 0)
    s = re.sub(r"password=[^ ,]+", "password=xxxx", s, 0)
    return s


def dump(root=None, indent=0, output=None):
    """
    Taken from https://git.io/vp2zB.
    """
    if indent == 0:
        output.write(u'%s ── %s\n' % ('    ' * indent, root))
    else:
        output.write(u'%s └─ %s\n' % ('    ' * indent, root))
    for dep in g[root]:
        dump(root=dep, indent=indent + 1, output=output)


if __name__ == '__main__':
    try:
        parser = CmdlineParser(sys.argv[1:])
        root_task = parser.get_task_obj()

        queue = [root_task]

        while len(queue) > 0:
            task = queue.pop()
            for dep in task.deps():
                g[task].add(dep)
                queue.append(dep)

        output = StringIO()
        dump(root=root_task, output=output)
        print(sanitize(output.getvalue()))

    except MissingParameterException as err:
        print('missing parameter: %s' % err, file=sys.stderr)
        sys.exit(1)
    except TaskClassNotFoundException as err:
        print(err, file=sys.stderr)
        sys.exit(1)
    except BrokenPipeError:
        pass