import click


def _call_in_source(dag, method_name, message, kwargs=None):
    """
    Execute method on each task.source in dag, passing kwargs
    """
    kwargs = kwargs or {}
    files = []
    results = []

    for task in dag.values():
        try:
            method = getattr(task.source, method_name)
        except AttributeError:
            pass
        else:
            results.append(method(**kwargs))
            files.append(str(task.source._path))

    files_ = '\n'.join((f'    {f}' for f in files))
    click.echo(f'{message}:\n{files_}')

    return results