Task scheduler & executor with text-based user interface, able to:
- Declare task & dependency as a graph
- Execute task graph
- Show task status, stdout & log
- Provide a nice user interface
I always have a lot of scripts to run. I want to execute them parallelly in a certain order and be able to check their status, output & log in an easy way. So I built this package.
$ pip install gtui
Note: it's not supported on windows, you need to use cygwin or wsl.
Let's say we have some helloworld tasks & their dependencies like this:
We need to create Task and add them to TaskGraph:
import time
import logging
from gtui import Task, TaskGraph
def foo(n):
logging.info('foo(%s) is called', n)
print('Start to sleep %s seconds.' % n)
time.sleep(n)
print('Hello World!')
t1 = Task('t1', func=foo, args=[0.1])
t2 = Task('t2', func=foo, args=[1])
t3 = Task('t3', func=foo, args=[1])
t4 = Task('t4', func=foo, args=[0.5])
g = TaskGraph()
g.add_task(t1)
g.add_task(t2, waiting_for=t1)
g.add_task(t3, waiting_for=t1)
g.add_task(t4, waiting_for=[t2, t3])
g.run()TaskGraph.run starts the text user interface, and you can navigate through tasks, see their status, output and logs:
Keybindings:
- t : toggle tail -f mode, will follow text when enabled
- tab : switch between output & log
- j/k : select previous/next task
- h/l : page up/down
- ↑/↓ : scroll up/down one line
- y : copy text
- q : exit
Task defines what to do and has an unique name:
# foo(*args, **kwargs) will be called
t = Task(name='foo', func=foo, args=[1, 2], kwargs={'foo': 'bar'})TaskGraph defines execution order of a set of tasks, it provides method to declare task & dependency:
g = TaskGraph()
g.add_task(t1) # added task t1
g.add_task(t2, waiting_for=t1) # added task t2, t2 runs after t1 finishes
g.add_tasks([t3, t4]) # added task t3, t4
g.add_dependency(t3, waiting_for=[t1, t2]) # declare t3 to run after t1 & t2 finishWhen TaskGraph contains a cycle denpendency, run method will throw a ValueError. You can also use has_cycle to
check:
> g = TaskGraph()
> g.add_tasks([t1, t2])
> g.add_dependency(t1, waiting_for=t2)
> g.add_dependency(t2, waiting_for=t1)
> g.has_cycle()
[t1, t2, t1]TaskGraph.run provides some options:
g.run(
title='Demo', # Text shown at the left bottom corner
callback=None, # A function called when execution fail or succeed
log_formatter=None, # An instance of logging.Formatter, to specify the log format
exit_on_success=False # whether exit tui when execution succeed
)callback can be used to notify the execution result, it will be called with an boolean indicating whether execution succeed. gtui.callback has some common callbacks:
# emit a desktop notification, use osascript on mac and notify-send on linux
from gtui import callback
g.run(callback.desktop_nofity(title='Plz See Here!', success_msg='Success', fail_msg='Fail'))Writing to stdout will break the TUI display. gtui runs each task in a new thread with sys.stdout replaced so functions like print will just work fine. When creating a new thread inside a task, gtui.IORedirectedThread can be used to achieve the same result:
from gtui import IORedirectedThread
t = IORedirectedThread(target=print, args=['hello world'])
t.start()
t.join()
content = t.get_stdout_content()However, gtui doesn't try to deal with other cases so you should take care of it by yourself.