phulpy 1.0.10

Install:

$ pip install phulpy

Create your phulpyfile.py (the configuration file, that describes all your tasks):

from phulpy import task, Output

@task
def default(phulpy):
    def print_file(file):
        Output.out(Output.colorize(
            file.relative_path,
            'green'
        ))

    def print_src_class(src):
        Output.out(src.__class__.__name__)

    def if_phulpy_file(file):
        return 'phulpyfile.py' in file.name

    phulpy.src(['./*'], read=True) \
        .pipe(phulpy.iterate(print_file)) \
        .pipe(print_src_class) \
        .pipe(phulpy.filter(if_phulpy_file)) \
        .pipe(phulpy.iterate(print_file)) \
        .pipe(phulpy.dest('./var'))


@task
def clean(phulpy):
    phulpy.src(['./var/*']) \
        .pipe(phulpy.clean())

@task
def exec_1(phulpy):
    def __on_stdout__(tick, stdin):
        # Output.out(tick)

        if '[yes]?' in tick:
            stdin.write('yes\n')

    def __on_stderr__(tick, stdin):
        # Output.err(tick)
        pass

    def __on_finish__(sigint, stdout, stderr):
        pass

    command = phulpy.execute(
        'ls -lah',
        quite=False,  # write on stdout and stdin, default True, needs to be handled
        sync=False,  # default True
        on_stdout=__on_stdout__,  # handle stdout
        on_stderr=__on_stderr__,  # handle stderr
        on_finish=__on_finish__  # handle sigint
    )

    command.kill() # force stop command


@task
def exec_2(phulpy):
    # when its is sync return a command with stdout, stderr, and sigint
    command = phulpy.execute('ls -lah')

    sigint = command.sigint
    stdout = command.stdout
    stderr = command.stderr

@task
def do_nothing():
    pass

Run:

Run the phulpy over the phulpyfile.py directory

$ phulpy --help
$ phulpy # Will run the `default` task
$ phulpy mytask # Will run the `mytask` task
$ phulpy --threads 4 mytask1 mytask2  # Will run the tasks simultaneously

By:

@jefersondaniel <https://github.com/jefersondaniel>

@reisraff <https://github.com/reisraff>