redun.console package#

Submodules#

redun.console.app module#

class redun.console.app.MenuItem(widget: Widget, value: Any)#

Bases: ListItem

can_focus: bool = False#: Widget may receive focus.

can_focus_children: bool = True#: Widget’s children may receive focus.

class redun.console.app.MenuScreen(name: str | None = None, id: str | None = None, classes: str | None = None)#

Bases: Screen

Main menu for top-level Screens.

BINDINGS: ClassVar[list[BindingType]] = [('q', 'quit', 'Quit'), ('m', '', ''), ('escape', 'app.pop_screen()', 'Back')]#

CSS_PATH: ClassVar[CSSPathType | None] = 'style.css'#: File paths to load CSS from.

Note

This CSS applies to the whole app.

can_focus: bool = False#: Widget may receive focus.

can_focus_children: bool = True#: Widget’s children may receive focus.

compose() → Iterable[Widget]#

Called by Textual to create child widgets.

Extend this to build a UI.

Example

```python def compose(self) -> ComposeResult:

```

on_list_view_selected(message) → None#

class redun.console.app.RedunApp(scheduler: Scheduler, args: Namespace, extra_args: List[str], argv: List[str])#

Bases: App

Top-level redun console App.

BINDINGS: ClassVar[list[BindingType]] = [('q', 'quit', 'Quit'), ('m', "push_screen('MenuScreen')", 'Menu'), Binding(key='ctrl+p', action='print_screen', description='Print screen', show=False, key_display=None, priority=False)]#

CSS_PATH: ClassVar[CSSPathType | None] = 'style.css'#: File paths to load CSS from.

SCREENS: ClassVar[dict[str, Screen[Any] | Callable[[], Screen[Any]]]] = {'ExecutionsNamespaceScreen': <class 'redun.console.screens.ExecutionsNamespaceScreen'>, 'ExecutionsScreen': <class 'redun.console.screens.ExecutionsScreen'>, 'FilesScreen': <class 'redun.console.screens.FilesScreen'>, 'MenuScreen': <class 'redun.console.app.MenuScreen'>, 'ReplScreen': <class 'redun.console.screens.ReplScreen'>, 'SearchScreen': <class 'redun.console.screens.SearchScreen'>, 'TasksScreen': <class 'redun.console.screens.TasksScreen'>}#: Screens associated with the app for the lifetime of the app.

TITLE: str | None = 'redun'#

A class variable to set the default title for the application.

To update the title while the app is running, you can set the [title][textual.app.App.title] attribute. See also [the Screen.TITLE attribute][textual.screen.Screen.TITLE].

action_print_screen() → None#

close_screen() → None#: Pop and uninstall the current screen.

goto_record(record: Any) → None#: Push a Screen relevant for the given record.

goto_screen(screen_cls: Screen, name: str, args: tuple) → Screen#: Get or create a new screen.

on_mount() → None#

route(argv: List[str]) → None#: Route a path to the appropriate Screen.

route_screens: List[Type[RedunScreen]] = [<class 'redun.console.screens.ExecutionsScreen'>, <class 'redun.console.screens.ExecutionsNamespaceScreen'>, <class 'redun.console.screens.ExecutionScreen'>, <class 'redun.console.screens.JobScreen'>, <class 'redun.console.screens.TasksScreen'>, <class 'redun.console.screens.TaskVersionsScreen'>, <class 'redun.console.screens.TaskScreen'>, <class 'redun.console.screens.ValueScreen'>, <class 'redun.console.upstream_screen.UpstreamDataflowScreen'>, <class 'redun.console.screens.FilesScreen'>, <class 'redun.console.screens.SearchScreen'>]#

redun.console.parser module#

redun.console.parser.format_args(parser: ArgumentParser, args: Namespace) → List[str]#: Format a Namespace back into an argv.

redun.console.parser.get_parser_action(parser: ArgumentParser, key: str) → Action | None#: Returns the associated ArgumentParser Action for the given Namespace key.

redun.console.parser.parse_args(parser: ArgumentParser, argv: List[str] = []) → Namespace#: Parse an argv into an argparse.Namespace.

redun.console.screens module#

class redun.console.screens.ExecutionScreen(execution_id: str)#

Bases: RedunScreen

Screen for exploring a redun Execution.

BINDINGS: ClassVar[list[BindingType]] = [('q', 'quit', 'Quit'), ('escape', 'back', 'Back'), ('n', 'next', 'Next page'), ('p', 'prev', 'Previous page'), ('f', 'focus', 'Focus job'), ('g', 'unfocus', 'Unfocus job'), ('/', 'filter', 'Filter'), ('r', 'repl', 'REPL'), ('ctrl+r', 'refresh', 'Refresh')]#

action_back()#: Callback for closing screen.

action_focus() → None#: Callback for focusing on currently selected job.

action_next() → None#: Callback for next page of jobs.

action_prev() → None#: Callback for previous page of jobs.

async action_refresh() → None#

action_repl() → None#: Callback for requesting a REPL to explore the execution.

action_unfocus() → None#: Callback for unfocusing the focused job.

can_focus: bool = False#: Widget may receive focus.

can_focus_children: bool = True#: Widget’s children may receive focus.

compose() → Iterable[Widget]#

Called by Textual to create child widgets.

Extend this to build a UI.

Example

```python def compose(self) -> ComposeResult:

```

classmethod get_parser() → ArgumentParser#

get_path() → str#: Returns the path element of the argv (the first argument).

jobs#

Create a reactive attribute.

Parameters:

default – A default value or callable that returns a default.
layout – Perform a layout on change.
repaint – Perform a repaint on change.
init – Call watchers on initialize (post mount).
always_update – Call watchers even when the new value equals the old value.

async load_jobs()#

on_job_list_selected(message: Selected) → None#

on_table_selected(message: Selected) → None#

path_pattern = '^executions/(.*)$'#

watch_args() → None#: Update UI when args change.

class redun.console.screens.ExecutionsNamespaceScreen#

Bases: RedunScreen

Top-level Screen for exploring all redun Execution namespaces.

BINDINGS: ClassVar[list[BindingType]] = [('q', 'quit', 'Quit'), ('escape', 'app.pop_screen()', 'Back'), ('n', "sort('namespace')", 'Sort by namespace'), ('c', "sort('count')", 'Sort by Execution count'), ('t', "sort('time')", "Sort by 'most recent'"), ('/', 'filter', 'Filter'), ('r', 'repl', 'REPL')]#

action_repl() → None#: Callback for requesting a REPL to explore the execution namespaces.

action_sort(column: str) → None#: Callback for selecting column sorting.

can_focus: bool = False#: Widget may receive focus.

can_focus_children: bool = True#: Widget’s children may receive focus.

compose() → Iterable[Widget]#

Called by Textual to create child widgets.

Extend this to build a UI.

Example

```python def compose(self) -> ComposeResult:

```

classmethod get_parser() → ArgumentParser#

get_path() → str#: Returns the path element of the argv (the first argument).

async load_namespaces()#: Load execution namespaces list from db.

on_mount()#

on_table_selected(message: Selected) → None#: Callback for when namespace is selected.

path_pattern = '^executions/namespace$'#

watch_args()#: Update screen with new arguments.

class redun.console.screens.ExecutionsScreen#

Bases: RedunScreen

Top-level Screen for exploring all redun Executions.

BINDINGS: ClassVar[list[BindingType]] = [('q', 'quit', 'Quit'), ('escape', 'back', 'Back'), ('n', 'next', 'Next page'), ('p', 'prev', 'Previous page'), ('/', 'filter', 'Filter'), ('r', 'repl', 'REPL'), ('ctrl+r', 'refresh', 'Refresh')]#

action_back() → None#: Callback for popping the screen.

action_next() → None#: Callback for next page of executions.

action_prev() → None#: Callback for previous page of executions.

async action_refresh() → None#

action_repl() → None#: Callback for requesting a repl for the current executions list.

can_focus: bool = False#: Widget may receive focus.

can_focus_children: bool = True#: Widget’s children may receive focus.

compose() → Iterable[Widget]#

Called by Textual to create child widgets.

Extend this to build a UI.

Example

```python def compose(self) -> ComposeResult:

```

classmethod get_parser() → ArgumentParser#

get_path() → str#: Returns the path element of the argv (the first argument).

async load_executions()#: Load executions list from db.

notify_args() → None#: Notify the screen that self.args has changed.

on_execution_list_selected(message: Selected) → None#: Callback for when execution is selected.

on_screen_resume(message: Message) → None#: When resuming the screen ensure executions are loaded.

path_pattern = '^executions$'#

watch_args() → None#: Update filter when args change.

class redun.console.screens.FilesScreen#

Bases: RedunScreen

Top-level Screen for exploring all redun Files.

BINDINGS: ClassVar[list[BindingType]] = [('q', 'quit', 'Quit'), ('escape', 'app.pop_screen()', 'Back'), ('n', 'next', 'Next page'), ('p', 'prev', 'Previous page'), ('/', 'filter', 'Filter'), ('r', 'repl', 'REPL')]#

action_next() → None#: Callback for next page of jobs.

action_prev() → None#: Callback for previous page of jobs.

action_repl() → None#: Callback for requesting a REPL to explore the execution namespaces.

can_focus: bool = False#: Widget may receive focus.

can_focus_children: bool = True#: Widget’s children may receive focus.

compose() → Iterable[Widget]#

Called by Textual to create child widgets.

Extend this to build a UI.

Example

```python def compose(self) -> ComposeResult:

```

classmethod get_parser() → ArgumentParser#

get_path() → str#: Returns the path element of the argv (the first argument).

async load_files()#: Load task fullnames from db.

on_table_selected(message: Selected) → None#: Callback for when task is selected.

path_pattern = '^files$'#

watch_args() → None#: Update screen with new arguments.

class redun.console.screens.FilterScreen(parent_screen: RedunScreen)#

Bases: Screen

Generic Screen for editing the parent Screen filters.

BINDINGS: ClassVar[list[BindingType]] = [('q', 'quit', 'Quit'), ('escape', 'app.pop_screen()', 'Back')]#

CSS_PATH: ClassVar[CSSPathType | None] = 'style.css'#: File paths to load CSS from.

Note

This CSS applies to the whole app.

can_focus: bool = False#: Widget may receive focus.

can_focus_children: bool = True#: Widget’s children may receive focus.

compose() → Iterable[Widget]#

Called by Textual to create child widgets.

Extend this to build a UI.

Example

```python def compose(self) -> ComposeResult:

```

on_input_submitted(message)#

class redun.console.screens.HelpScreen(parent_screen: RedunScreen)#

Bases: Screen

Modal Help Screen that is launched from a parent Screen.

BINDINGS: ClassVar[list[BindingType]] = [('q', 'quit', 'Quit'), ('escape', 'app.pop_screen()', 'Back')]#

CSS_PATH: ClassVar[CSSPathType | None] = 'style.css'#: File paths to load CSS from.

Note

This CSS applies to the whole app.

can_focus: bool = False#: Widget may receive focus.

can_focus_children: bool = True#: Widget’s children may receive focus.

compose() → Iterable[Widget]#

Called by Textual to create child widgets.

Extend this to build a UI.

Example

```python def compose(self) -> ComposeResult:

```

class redun.console.screens.JobScreen(job_id: str)#

Bases: RedunScreen

Screen for exploring a redun Job.

BINDINGS: ClassVar[list[BindingType]] = [('q', 'quit', 'Quit'), ('escape', 'back', 'Back'), ('p', 'parent', 'Parent job'), ('c', 'children', 'Child jobs'), ('r', 'repl', 'REPL')]#

action_children() → None#: Callback for showing child jobs.

action_click_exec() → None#: Callback for clicking on Execution.

action_click_parent_job() → None#: Callback for clicking on the parent job.

action_click_task() → None#: Callback for click on a Task.

action_click_value(value_hash: str) → None#: Callback for clicking on a Value.

action_parent() → None#: Callback for showing parent job.

action_repl() → None#: Callback for requesting a repl for the current Job.

can_focus: bool = False#: Widget may receive focus.

can_focus_children: bool = True#: Widget’s children may receive focus.

compose() → Iterable[Widget]#

Called by Textual to create child widgets.

Extend this to build a UI.

Example

```python def compose(self) -> ComposeResult:

```

get_path() → str#: Returns the path element of the argv (the first argument).

path_pattern = '^jobs/(.*)$'#

class redun.console.screens.RedunScreen#

Bases: Screen

Base Textual Screen for redun.

action_back() → None#: Callback for popping the screen.

action_filter() → None#: Callback for displaying the filter screen for the current screen.

can_focus: bool = False#: Widget may receive focus.

can_focus_children: bool = True#: Widget’s children may receive focus.

compose() → Iterable[Widget]#

Called by Textual to create child widgets.

Extend this to build a UI.

Example

```python def compose(self) -> ComposeResult:

```

get_options() → str#: Returns the options as single string.

classmethod get_parser() → ArgumentParser#

get_path() → str#: Returns the path element of the argv (the first argument).

notify_args() → None#: Notify the screen that self.args has changed.

on_mount()#

parse(argv: List[str]) → None#: Parse command-line arguments and update screen.

path_pattern = ''#

watch_args() → None#: Update screen with new arguments.

class redun.console.screens.ReplScreen(locals={}, obj_id: str | None = None)#

Bases: RedunScreen

Screen for read-eval-print-loop (REPL).

BINDINGS: ClassVar[list[BindingType]] = [('q', 'quit', 'Quit'), ('escape', 'app.pop_screen()', 'Back'), ('c', 'clear', 'Clear')]#

action_clear() → None#: Clear repl output.

can_focus: bool = False#: Widget may receive focus.

can_focus_children: bool = True#: Widget’s children may receive focus.

compose() → Iterable[Widget]#

Called by Textual to create child widgets.

Extend this to build a UI.

Example

```python def compose(self) -> ComposeResult:

```

get_path() → str#: Returns the path element of the argv (the first argument).

on_command_input_complete() → None#: Callback for tab-complete action.

on_input_submitted(message) → None#: Callback for when input command is submitted.

on_mount() → None#

on_write(data: str) → None#: Callback for when the interpreter wants to write to the output log.

path_pattern = 'repl(/(.*))?'#

update(locals={}, obj_id: str | None = None) → None#: Update the repl environment with new values.

write_intro() → None#: Write intro text to log.

class redun.console.screens.SearchScreen(*args: Any, **kwargs: Any)#

Bases: RedunScreen

Screen for doing general record searching.

BINDINGS: ClassVar[list[BindingType]] = [('q', 'quit', 'Quit'), ('escape', 'app.pop_screen()', 'Back'), ('n', 'next', 'Next page'), ('p', 'prev', 'Previous page'), ('/', 'filter', 'Filter'), ('r', 'repl', 'REPL')]#

action_next() → None#: Callback for next page of results.

action_prev() → None#: Callback for previous page of results.

action_repl() → None#

can_focus: bool = False#: Widget may receive focus.

can_focus_children: bool = True#: Widget’s children may receive focus.

compose() → Iterable[Widget]#

Called by Textual to create child widgets.

Extend this to build a UI.

Example

```python def compose(self) -> ComposeResult:

```

classmethod get_parser() → ArgumentParser#

get_path() → str#: Returns the path element of the argv (the first argument).

async load_results() → None#

on_input_submitted(message: Submitted) → None#: Callback for when input command is submitted.

on_table_selected(message: Selected) → None#: Callback for pressing enter key on a result.

page#

Create a reactive attribute.

Parameters:

default – A default value or callable that returns a default.
layout – Perform a layout on change.
repaint – Perform a repaint on change.
init – Call watchers on initialize (post mount).
always_update – Call watchers even when the new value equals the old value.

path_pattern = '^search$'#

results#

Create a reactive attribute.

Parameters:

default – A default value or callable that returns a default.
layout – Perform a layout on change.
repaint – Perform a repaint on change.
init – Call watchers on initialize (post mount).
always_update – Call watchers even when the new value equals the old value.

watch_args()#: Update screen with new arguments.

watch_page() → None#

watch_results() → None#

class redun.console.screens.TaskScreen(task_hash: str)#

Bases: RedunScreen

Screen for exploring a redun Task.

BINDINGS: ClassVar[list[BindingType]] = [('q', 'quit', 'Quit'), ('escape', 'back', 'Back'), ('r', 'repl', 'REPL')]#

action_click_jobs() → None#: Callback for clicking jobs link.

action_click_namespace() → None#: Callback for clicking Task namespace.

action_click_versions() → None#: Callback for clicking on Task versions link.

action_repl() → None#: Callback for requesting a repl for the current Task.

can_focus: bool = False#: Widget may receive focus.

can_focus_children: bool = True#: Widget’s children may receive focus.

compose() → Iterable[Widget]#

Called by Textual to create child widgets.

Extend this to build a UI.

Example

```python def compose(self) -> ComposeResult:

```

get_path() → str#: Returns the path element of the argv (the first argument).

path_pattern = '^tasks/([^/]*)$'#

class redun.console.screens.TaskVersionsScreen(task_fullname: str)#

Bases: RedunScreen

Screen for exploring versions of a redun Task.

BINDINGS: ClassVar[list[BindingType]] = [('q', 'quit', 'Quit'), ('escape', 'app.pop_screen()', 'Back'), ('r', 'repl', 'REPL')]#

action_repl() → None#: Callback for requesting a REPL to explore the execution namespaces.

can_focus: bool = False#: Widget may receive focus.

can_focus_children: bool = True#: Widget’s children may receive focus.

compose() → Iterable[Widget]#

Called by Textual to create child widgets.

Extend this to build a UI.

Example

```python def compose(self) -> ComposeResult:

```

get_path() → str#: Returns the path element of the argv (the first argument).

async load_tasks()#: Load task fullnames from db.

on_table_selected(message: Selected) → None#: Callback for when task is selected.

path_pattern = '^tasks/name/([^/]*)$'#

watch_args() → None#: Update screen with new arguments.

class redun.console.screens.TasksScreen#

Bases: RedunScreen

Top-level Screen for exploring all redun Tasks.

BINDINGS: ClassVar[list[BindingType]] = [('q', 'quit', 'Quit'), ('escape', 'app.pop_screen()', 'Back'), ('/', 'filter', 'Filter'), ('r', 'repl', 'REPL')]#

action_repl() → None#: Callback for requesting a REPL to explore the execution namespaces.

can_focus: bool = False#: Widget may receive focus.

can_focus_children: bool = True#: Widget’s children may receive focus.

compose() → Iterable[Widget]#

Called by Textual to create child widgets.

Extend this to build a UI.

Example

```python def compose(self) -> ComposeResult:

```

classmethod get_parser() → ArgumentParser#

get_path() → str#: Returns the path element of the argv (the first argument).

async load_tasks()#: Load task fullnames from db.

on_table_selected(message: Selected) → None#: Callback for when task is selected.

path_pattern = '^tasks$'#

watch_args() → None#: Update screen with new arguments.

class redun.console.screens.ValueScreen(value_hash: str)#

Bases: RedunScreen

Screen for exploring a redun Value.

BINDINGS: ClassVar[list[BindingType]] = [('q', 'quit', 'Quit'), ('escape', 'back', 'Back'), ('r', 'repl', 'REPL')]#

action_click_value(value_hash: str) → None#

action_downstream_jobs() → None#

action_repl() → None#

action_upstream() → None#

action_upstream_jobs() → None#

can_focus: bool = False#: Widget may receive focus.

can_focus_children: bool = True#: Widget’s children may receive focus.

compose() → Iterable[Widget]#

Called by Textual to create child widgets.

Extend this to build a UI.

Example

```python def compose(self) -> ComposeResult:

```

get_path() → str#: Returns the path element of the argv (the first argument).

path_pattern = '^values/(.*)$'#

redun.console.screens.tree_sort_jobs(root: Job, jobs: List[Job]) → List[Job]#: Sort Jobs in preorder traversal. Annotate tree depth on each job.

redun.console.upstream_screen module#

class redun.console.upstream_screen.UpstreamDataflowScreen(value_hash: str)#

Bases: RedunScreen

Screen for exploring the upstream dataflow of a redun Value.

BINDINGS: ClassVar[list[BindingType]] = [('q', 'quit', 'Quit'), ('escape', 'back', 'Back'), ('r', 'repl', 'REPL')]#

action_click_hash(hash: str) → None#

action_click_value(value_hash: str) → None#

action_repl() → None#

can_focus: bool = False#: Widget may receive focus.

can_focus_children: bool = True#: Widget’s children may receive focus.

compose() → Iterable[Widget]#

Called by Textual to create child widgets.

Extend this to build a UI.

Example

```python def compose(self) -> ComposeResult:

```

get_path() → str#: Returns the path element of the argv (the first argument).

path_pattern = '^upstreams/(.*)$'#

redun.console.upstream_screen.display_call_node(call_node: CallNode, renames: Dict[str, str]) → str#: Formats a CallNode to a string.

redun.console.upstream_screen.display_dataflow(dom: Iterable[DataflowSectionDOM]) → Iterator[str]#: Yields for lines displaying a dataflow DOM.

redun.console.upstream_screen.display_hash(node: ArgumentValue | CallNodeValue | CallNode | Value | None) → str#: Formats hash for a DataflowNode.

redun.console.upstream_screen.display_link(hash: str | None, text: str) → str#

redun.console.upstream_screen.display_node(node: ArgumentValue | CallNodeValue | CallNode | Value | None, renames: Dict[str, str]) → Tuple[str, str]#: Formats a dataflow node to a string.

redun.console.upstream_screen.display_section(dom: DataflowSectionDOM) → Iterator[str]#: Yields lines for displaying a dataflow section DOM.

redun.console.upstream_screen.display_value(value: Value) → str#: Format a Value to a string.

redun.console.utils module#

redun.console.utils.format_arguments(args: List[Argument]) → str#

Display CallNode arguments.

For example, if args has 2 positional and 1 keyword argument, we would display that as:

redun.console.utils.format_job(job: Job) → str#: Format a redun Job into a string representation.

redun.console.utils.format_link(link_pattern: str, tags: Dict[str, Any]) → str | None#: Format a link pattern using a tag dictionary.

redun.console.utils.format_record(record: Any) → str#: Format a redun repo record (e.g. Execution, Job, etc) into a string.

redun.console.utils.format_tags(tags: List[Tag], max_length: int = 100, color='#9999cc') → str#: Format a set of tags.

redun.console.utils.format_traceback(job: Job) → str#: Format the call stack from Execution down to the given Job.

redun.console.utils.get_links(link_patterns: List[str], tags: List[Tag]) → List[str]#: Get links from a list of link patterns and redun tags.

redun.console.utils.log_write(*args: Any) → None#: Debugging function to use when developing with Textual.

redun.console.utils.style_status(status: str) → str#: Returns styled text for a job/execution status.

redun.console.widgets module#

class redun.console.widgets.CommandInput(*args, complete: bool = True, **kwargs)#

Bases: Input

Input widget with history and tab-completion.

BINDINGS: ClassVar[list[BindingType]] = [Binding(key='up', action='cursor_up', description='cursor up', show=False, key_display=None, priority=False), Binding(key='down', action='cursor_down', description='cursor down', show=False, key_display=None, priority=False)]#: Key(s) | Description |

:- | :- |

left | Move the cursor left. |

ctrl+left | Move the cursor one word to the left. |

right | Move the cursor right or accept the completion suggestion. |

ctrl+right | Move the cursor one word to the right. |

backspace | Delete the character to the left of the cursor. |

home,ctrl+a | Go to the beginning of the input. |

end,ctrl+e | Go to the end of the input. |

delete,ctrl+d | Delete the character to the right of the cursor. |

enter | Submit the current value of the input. |

ctrl+w | Delete the word to the left of the cursor. |

ctrl+u | Delete everything to the left of the cursor. |

ctrl+f | Delete the word to the right of the cursor. |

ctrl+k | Delete everything to the right of the cursor. |

class Complete(text: str)#

Bases: Message

bubble: ClassVar[bool] = True#

handler_name: ClassVar[str] = 'on_command_input_complete'#: Name of the default message handler.

no_dispatch: ClassVar[bool] = False#

time#

verbose: ClassVar[bool] = False#

action_complete() → None#: Callback for tab-complete attempt.

action_cursor_down() → None#

action_cursor_up() → None#

can_focus: bool = True#: Widget may receive focus.

can_focus_children: bool = True#: Widget’s children may receive focus.

on_input_submitted(message: Submitted) → None#

class redun.console.widgets.ExecutionList(executions: List[Execution] | None, **kwargs)#

Bases: DataTable

Table of redun Executions.

class Selected(execution: Execution)#

Bases: Message

bubble: ClassVar[bool] = True#

handler_name: ClassVar[str] = 'on_execution_list_selected'#: Name of the default message handler.

no_dispatch: ClassVar[bool] = False#

time#

verbose: ClassVar[bool] = False#

can_focus: bool = True#: Widget may receive focus.

can_focus_children: bool = True#: Widget’s children may receive focus.

executions#

Create a reactive attribute.

Parameters:

default – A default value or callable that returns a default.
layout – Perform a layout on change.
repaint – Perform a repaint on change.
init – Call watchers on initialize (post mount).
always_update – Call watchers even when the new value equals the old value.

key_enter(event: Key) → None#: Callback for pressing enter key on an Execution.

watch_executions() → None#: Set the list of Executions to display.

class redun.console.widgets.Interpreter(locals, write)#

Bases: InteractiveInterpreter

Interactive Python interpreter used by the ReplScreen.

run(code: str) → Any#: Execute a code object.

class redun.console.widgets.JobList(jobs: List[Job] | None, **kwargs)#

Bases: DataTable

Table of redun Jobs.

class Selected(job: Job)#

Bases: Message

bubble: ClassVar[bool] = True#

handler_name: ClassVar[str] = 'on_job_list_selected'#: Name of the default message handler.

no_dispatch: ClassVar[bool] = False#

time#

verbose: ClassVar[bool] = False#

can_focus: bool = True#: Widget may receive focus.

can_focus_children: bool = True#: Widget’s children may receive focus.

jobs#

Create a reactive attribute.

Parameters:

default – A default value or callable that returns a default.
layout – Perform a layout on change.
repaint – Perform a repaint on change.
init – Call watchers on initialize (post mount).
always_update – Call watchers even when the new value equals the old value.

key_enter(event: Key) → None#: Callback for pressing enter key on a Job.

watch_jobs() → None#

class redun.console.widgets.JobStatusTable(execution_id: str, **kwargs)#

Bases: Table

Table of redun Job statuses by task.

JOB_STATUSES = ['RUNNING', 'FAILED', 'CACHED', 'DONE', 'TOTAL']#

can_focus: bool = True#: Widget may receive focus.

can_focus_children: bool = True#: Widget’s children may receive focus.

async load_jobs() → None#

on_mount() → None#

class redun.console.widgets.RedunFooter(page: int)#

Bases: Footer

Footer used on Screens that use pagination.

can_focus: bool = False#: Widget may receive focus.

can_focus_children: bool = True#: Widget’s children may receive focus.

page#

Create a reactive attribute.

Parameters:

default – A default value or callable that returns a default.
layout – Perform a layout on change.
repaint – Perform a repaint on change.
init – Call watchers on initialize (post mount).
always_update – Call watchers even when the new value equals the old value.

render() → ConsoleRenderable | RichCast | str#

Get text or Rich renderable for this widget.

Implement this for custom widgets.

Example

```python from textual.app import RenderableType from textual.widget import Widget

class CustomWidget(Widget):

def render(self) -> RenderableType:: return “Welcome to [bold red]Textual[/]!”

```

Returns:: Any renderable.

class redun.console.widgets.RedunHeader(title: str, **kwargs)#

Bases: Static

Header used on all Screen to display the current CLI command (e.g. argv).

CSS_PATH = 'style.css'#

can_focus: bool = False#: Widget may receive focus.

can_focus_children: bool = True#: Widget’s children may receive focus.

update(title: str) → None#

Update the widget’s content area with new text or Rich renderable.

Parameters:: renderable – A new rich renderable. Defaults to empty renderable;

class redun.console.widgets.Table(*, show_header: bool = True, show_row_labels: bool = True, fixed_rows: int = 0, fixed_columns: int = 0, zebra_stripes: bool = False, header_height: int = 1, show_cursor: bool = True, cursor_foreground_priority: Literal['renderable', 'css'] = 'css', cursor_background_priority: Literal['renderable', 'css'] = 'renderable', cursor_type: Literal['cell', 'row', 'column', 'none'] = 'cell', cell_padding: int = 1, name: str | None = None, id: str | None = None, classes: str | None = None, disabled: bool = False)#

Bases: DataTable

Generic DataTable that uses Enter key to select cells.

class Selected(table: Table)#

Bases: Message

bubble: ClassVar[bool] = True#

handler_name: ClassVar[str] = 'on_table_selected'#: Name of the default message handler.

no_dispatch: ClassVar[bool] = False#

time#

verbose: ClassVar[bool] = False#

can_focus: bool = True#: Widget may receive focus.

can_focus_children: bool = True#: Widget’s children may receive focus.

key_enter(event: Key) → None#: Callback for pressing enter key.

class redun.console.widgets.TagLinks(link_patterns: List[str], tags: List[Tag])#

Bases: Container

Displays links derived from redun Tags.

can_focus: bool = False#: Widget may receive focus.

can_focus_children: bool = True#: Widget’s children may receive focus.

compose() → Iterable[Widget]#

Called by Textual to create child widgets.

Extend this to build a UI.

Example

```python def compose(self) -> ComposeResult:

```

class redun.console.widgets.ValueSpan(value: ~redun.backends.db.Value, format: str = '{value}', max_size: int = 100, on_click=<function ValueSpan.<lambda>>)#

Bases: Static

Rendering of a redun Value that is linkable and collapsible.

action_click() → None#

action_expand() → None#

can_focus: bool = False#: Widget may receive focus.

can_focus_children: bool = True#: Widget’s children may receive focus.

expanded#

Create a reactive attribute.

Parameters:

default – A default value or callable that returns a default.
layout – Perform a layout on change.
repaint – Perform a repaint on change.
init – Call watchers on initialize (post mount).
always_update – Call watchers even when the new value equals the old value.

render() → Any#

Get a rich renderable for the widget’s content.

Returns:: A rich renderable.

redun.console package#

Submodules#

redun.console.app module#

redun.console.parser module#

redun.console.screens module#

redun.console.upstream_screen module#

redun.console.utils module#

redun.console.widgets module#

Module contents#