InquirerApp class for building custom multi-prompt flows and TUI applications with complex logic.
from inquirer_textual.InquirerApp import InquirerApp
from inquirer_textual.widgets.InquirerText import InquirerText
from inquirer_textual.widgets.InquirerSelect import InquirerSelect
from inquirer_textual.common.Shortcut import Shortcutfrom typing import TypeVar, Generic, Callable, Any
from asyncio import AbstractEventLoop
from textual.app import App, AutopilotCallbackType
T = TypeVar('T')
class InquirerApp(App[Result[T]], Generic[T]):
"""
Main application class for interactive TUI applications.
Inherits from Textual's App and extends with inquirer functionality.
Type parameter T specifies result value type.
"""
# Instance attributes
widget: InquirerWidget | None = None
shortcuts: list[Shortcut] | None = None
header: str | list[str] | None = None
show_footer: bool = True
result: Result[T] | None = None
inquiry_func: Callable[[InquirerApp[T]], None] | None = None
# Internal attributes (advanced usage only)
result_ready: Event | None = None # Threading event for prompt() synchronization
inquiry_func_stop: bool = False # Flag indicating app has been stoppeddef __init__(self) -> None:
"""Initialize InquirerApp with default settings."""def run(
*,
headless: bool = False,
inline: bool = False,
inline_no_clear: bool = False,
mouse: bool = True,
size: tuple[int, int] | None = None,
auto_pilot: AutopilotCallbackType | None = None,
loop: AbstractEventLoop | None = None,
inquiry_func: Callable[[InquirerApp[T]], None] | None = None
) -> Result[T]:
"""
Run the application.
Args:
inline: If True, runs inline in terminal (preserves history)
inquiry_func: Function for building custom multi-prompt flows
Returns:
Result[T]: Final result when app exits
"""
def prompt(
self,
widget: InquirerWidget,
shortcuts: list[Shortcut] | None = None
) -> Result[T]:
"""
Display widget and wait for input. Use inside inquiry_func.
Args:
widget: Widget to display
shortcuts: Optional shortcuts for this prompt
Returns:
Result[T]: Result from this prompt
Raises:
RuntimeError: If app has been stopped
"""
def stop(self, value: Any = None):
"""
Stop the application with return value.
Args:
value: Value to return
"""
def focus_widget(self):
"""Focus the current widget."""
def get_theme_variable_defaults(self) -> dict[str, str]:
"""
Get default theme colors.
Returns:
dict: Theme variable names to color values
- 'select-question-mark': '#e5c07b'
- 'select-list-item-highlight-foreground': '#61afef'
- 'input-color': '#98c379'
"""class InquirerApp:
# CSS styling for the application
CSS: str = """
App {
background: black;
}
Screen {
border-top: none;
border-bottom: none;
background: transparent;
height: auto;
}
"""
# Textual app configuration
ENABLE_COMMAND_PALETTE: bool = False
INLINE_PADDING: int = 0
# Key bindings
BINDINGS: list[Binding] = [
Binding("ctrl+d", "quit", "Quit", show=False, priority=True)
]Notes:
CSS: Default styling for the app interfaceENABLE_COMMAND_PALETTE: Disables Textual's command paletteINLINE_PADDING: Padding for inline modeBINDINGS: Ctrl+D is bound to quit actionapp = InquirerApp()
app.widget = InquirerText("Enter name:")
result = app.run(inline=True)
if result.command == 'select':
print(f"Name: {result.value}")app = InquirerApp()
app.header = "User Registration"
app.widget = InquirerText("Username:")
result = app.run(inline=True)app = InquirerApp()
app.header = [
"Configuration Wizard",
"Please answer the following questions"
]
app.widget = InquirerText("Project name:")
result = app.run(inline=True)app = InquirerApp()
app.widget = InquirerText("Input:")
app.shortcuts = [
Shortcut('escape', 'cancel', 'Cancel'),
Shortcut('ctrl+s', 'save', 'Save')
]
result = app.run(inline=True)The inquiry_func parameter enables complex multi-prompt flows with conditional logic.
def inquiry_func(app: InquirerApp[T]) -> None:
"""
Custom flow function.
Args:
app: InquirerApp instance
Use app.prompt() to display widgets and app.stop() to set final value.
"""
passdef registration_flow(app):
# Prompt 1
name_result = app.prompt(InquirerText("Name:"))
if name_result.command != 'select':
return # User quit
# Prompt 2
email_result = app.prompt(InquirerText("Email:"))
if email_result.command != 'select':
return
# Set final value
app.stop({
'name': name_result.value,
'email': email_result.value
})
app = InquirerApp()
result = app.run(inline=True, inquiry_func=registration_flow)
if result.value:
print(f"Registered: {result.value['name']}")def conditional_flow(app):
# Ask user role
role_result = app.prompt(InquirerSelect(
"Select role:",
["Developer", "Designer", "Manager"]
))
if role_result.command != 'select':
return
# Different questions based on role
if role_result.value == "Developer":
lang_result = app.prompt(InquirerSelect(
"Primary language:",
["Python", "JavaScript", "Go"]
))
if lang_result.command == 'select':
app.stop({'role': role_result.value, 'language': lang_result.value})
else:
# No additional questions
app.stop({'role': role_result.value, 'language': None})
app = InquirerApp()
result = app.run(inline=True, inquiry_func=conditional_flow)def collect_items_flow(app):
"""Collect items until user quits or enters empty."""
items = []
while True:
item_result = app.prompt(InquirerText("Add item (empty to finish):"))
if item_result.command != 'select':
break # User quit
if not item_result.value.strip():
break # Empty input
items.append(item_result.value)
more_result = app.prompt(InquirerConfirm("Add another?", default=True))
if not more_result.value:
break
app.stop(items)
app = InquirerApp()
result = app.run(inline=True, inquiry_func=collect_items_flow)
print(f"Collected: {result.value}")def validated_flow(app):
"""Flow with custom validation and retry logic."""
while True:
email_result = app.prompt(InquirerText("Email:"))
if email_result.command != 'select':
return # User quit
if '@' in email_result.value and '.' in email_result.value:
# Valid email
app.stop(email_result.value)
return
else:
# Invalid - show error and retry
app.prompt(InquirerText("Invalid email! Press Enter to retry"))
app = InquirerApp()
result = app.run(inline=True, inquiry_func=validated_flow)class WizardState:
def __init__(self):
self.data = {}
self.step = 0
def wizard_flow(app):
state = WizardState()
# Step 1: Basic info
name_result = app.prompt(InquirerText("Name:"))
if name_result.command != 'select':
return
state.data['name'] = name_result.value
state.step += 1
# Step 2: Configuration
env_result = app.prompt(InquirerSelect("Environment:", ["Dev", "Prod"]))
if env_result.command != 'select':
return
state.data['environment'] = env_result.value
state.step += 1
# Step 3: Confirmation
confirm_result = app.prompt(InquirerConfirm("Is this correct?"))
if confirm_result.command == 'select' and confirm_result.value:
app.stop(state.data)
app = InquirerApp()
result = app.run(inline=True, inquiry_func=wizard_flow)def survey_flow(app):
satisfied_result = app.prompt(InquirerConfirm("Satisfied with service?", default=True))
if not satisfied_result.value:
# Dissatisfied - ask follow-up
issue_result = app.prompt(InquirerSelect(
"Main issue?",
["Too slow", "Too expensive", "Missing features"]
))
details_result = app.prompt(InquirerText("Details:"))
app.stop({
'satisfied': False,
'issue': issue_result.value,
'details': details_result.value
})
else:
# Satisfied - just get rating
rating_result = app.prompt(InquirerSelect(
"Rating:",
["5 - Excellent", "4 - Good", "3 - Average"]
))
app.stop({
'satisfied': True,
'rating': rating_result.value
})
app = InquirerApp()
result = app.run(inline=True, inquiry_func=survey_flow)def flow_with_shortcuts(app):
common_shortcuts = [Shortcut('escape', 'cancel', 'Cancel')]
r1 = app.prompt(InquirerText("Field 1:"), shortcuts=common_shortcuts)
if r1.command == 'cancel':
app.stop(None)
return
r2 = app.prompt(InquirerText("Field 2:"), shortcuts=common_shortcuts)
if r2.command == 'cancel':
app.stop(None)
return
app.stop((r1.value, r2.value))
app = InquirerApp()
result = app.run(inline=True, inquiry_func=flow_with_shortcuts)def ask_text(app, message, shortcuts=None):
"""Helper to ask text and return value or None."""
result = app.prompt(InquirerText(message), shortcuts=shortcuts)
return result.value if result.command == 'select' else None
def ask_confirm(app, message, default=False):
"""Helper to ask confirmation and return bool."""
result = app.prompt(InquirerConfirm(message, default=default))
return result.value if result.command == 'select' else None
def flow(app):
username = ask_text(app, "Username:")
if not username:
return
if ask_confirm(app, "Continue?", default=True):
app.stop(username)
app = InquirerApp()
result = app.run(inline=True, inquiry_func=flow)# Inline mode - embeds in terminal at cursor position
# Preserves terminal history
app = InquirerApp()
app.widget = InquirerText("Name:")
result = app.run(inline=True)
# Full screen mode - takes over entire terminal
# Like vim or less
app = InquirerApp()
app.widget = InquirerText("Name:")
result = app.run() # inline=False (default)class CustomApp(InquirerApp):
def get_theme_variable_defaults(self) -> dict[str, str]:
return {
'select-question-mark': '#ff0000', # Red
'select-list-item-highlight-foreground': '#00ff00', # Green
'input-color': '#0000ff' # Blue
}
app = CustomApp()
app.widget = InquirerSelect("Choose:", ["A", "B", "C"])
result = app.run(inline=True)def safe_flow(app):
try:
name_result = app.prompt(InquirerText("Name:"))
if name_result.command != 'select':
app.stop(None)
return
# Custom validation
if len(name_result.value) < 2:
error_result = app.prompt(InquirerText("Name too short! Press Enter to exit"))
app.stop(None)
return
app.stop(name_result.value)
except Exception as e:
print(f"Error: {e}")
app.stop(None)
app = InquirerApp()
result = app.run(inline=True, inquiry_func=safe_flow)def menu_system():
"""Interactive menu with actions."""
from inquirer_textual.common.Choice import Choice
def menu_flow(app):
while True:
action_result = app.prompt(InquirerSelect(
"Main Menu:",
[
Choice("Create New", command="create"),
Choice("View Existing", command="view"),
Choice("Settings", command="settings"),
Choice("Exit", command="exit")
]
))
if action_result.command == 'create':
name = app.prompt(InquirerText("Item name:"))
if name.command == 'select':
print(f"Created: {name.value}")
elif action_result.command == 'view':
print("Viewing items...")
elif action_result.command == 'settings':
print("Opening settings...")
elif action_result.command == 'exit' or action_result.command == 'quit':
app.stop(None)
return
app = InquirerApp()
app.run(inline=True, inquiry_func=menu_flow)
menu_system()from dataclasses import dataclass, field
from typing import List, Dict, Any
@dataclass
class ApplicationState:
"""State container for complex flows."""
user_data: Dict[str, Any] = field(default_factory=dict)
history: List[str] = field(default_factory=list)
current_step: int = 0
total_steps: int = 0
def add_step(self, step_name: str, data: Any):
self.history.append(step_name)
self.user_data[step_name] = data
self.current_step += 1
def can_go_back(self) -> bool:
return self.current_step > 0
def go_back(self):
if self.can_go_back():
last_step = self.history.pop()
del self.user_data[last_step]
self.current_step -= 1
def stateful_wizard(app):
state = ApplicationState(total_steps=3)
# Step 1: User info
name_result = app.prompt(InquirerText("Name:"))
if name_result.command != 'select':
return
state.add_step('name', name_result.value)
# Step 2: Role selection
role_result = app.prompt(InquirerSelect("Role:", ["Admin", "User", "Guest"]))
if role_result.command != 'select':
return
state.add_step('role', role_result.value)
# Step 3: Permissions based on role
if state.user_data['role'] == "Admin":
perms = ["read", "write", "delete", "admin"]
else:
perms = ["read", "write"]
perm_result = app.prompt(InquirerCheckbox("Permissions:", perms))
if perm_result.command != 'select':
return
state.add_step('permissions', perm_result.value)
app.stop(state.user_data)
app = InquirerApp()
result = app.run(inline=True, inquiry_func=stateful_wizard)def dynamic_form_builder(field_specs):
"""Generate form from field specifications."""
def form_flow(app):
results = {}
for spec in field_specs:
widget = None
if spec['type'] == 'text':
widget = InquirerText(spec['label'], default=spec.get('default', ''))
elif spec['type'] == 'number':
widget = InquirerNumber(spec['label'])
elif spec['type'] == 'select':
widget = InquirerSelect(spec['label'], spec['choices'])
elif spec['type'] == 'confirm':
widget = InquirerConfirm(spec['label'], default=spec.get('default', False))
elif spec['type'] == 'checkbox':
widget = InquirerCheckbox(spec['label'], spec['choices'])
if widget:
result = app.prompt(widget)
if result.command != 'select':
return # User quit
results[spec['key']] = result.value
app.stop(results)
return form_flow
# Define form structure
form_fields = [
{'type': 'text', 'key': 'name', 'label': 'Full Name:', 'default': ''},
{'type': 'select', 'key': 'country', 'label': 'Country:', 'choices': ['US', 'UK', 'CA']},
{'type': 'number', 'key': 'age', 'label': 'Age:'},
{'type': 'confirm', 'key': 'subscribe', 'label': 'Subscribe?', 'default': True}
]
app = InquirerApp()
result = app.run(inline=True, inquiry_func=dynamic_form_builder(form_fields))def nested_configuration_flow(app):
"""Nested configuration with sub-flows."""
def database_config(app):
"""Sub-flow for database configuration."""
db_type = app.prompt(InquirerSelect("Database:", ["MySQL", "PostgreSQL", "MongoDB"]))
if db_type.command != 'select':
return None
host = app.prompt(InquirerText("Host:", default="localhost"))
if host.command != 'select':
return None
port = app.prompt(InquirerNumber("Port:"))
if port.command != 'select':
return None
return {
'type': db_type.value,
'host': host.value,
'port': int(port.value)
}
def cache_config(app):
"""Sub-flow for cache configuration."""
use_cache = app.prompt(InquirerConfirm("Enable caching?", default=True))
if not use_cache.value:
return None
cache_type = app.prompt(InquirerSelect("Cache type:", ["Redis", "Memcached", "In-Memory"]))
if cache_type.command != 'select':
return None
return {'type': cache_type.value}
# Main flow
config = {}
# Database configuration
config['database'] = database_config(app)
if config['database'] is None:
return
# Cache configuration
config['cache'] = cache_config(app)
# API configuration
api_key = app.prompt(InquirerSecret("API Key:"))
if api_key.command != 'select':
return
config['api_key'] = api_key.value
app.stop(config)
app = InquirerApp()
result = app.run(inline=True, inquiry_func=nested_configuration_flow)def paginated_selection_flow(app):
"""Handle large choice lists with pagination concept."""
all_items = [f"Item {i}" for i in range(1, 101)] # 100 items
page_size = 10
current_page = 0
total_pages = (len(all_items) + page_size - 1) // page_size
while True:
start_idx = current_page * page_size
end_idx = min(start_idx + page_size, len(all_items))
page_items = all_items[start_idx:end_idx]
# Add navigation choices
from inquirer_textual.common.Choice import Choice
choices = [Choice(item, data=item) for item in page_items]
if current_page > 0:
choices.insert(0, Choice("< Previous Page", command="prev"))
if current_page < total_pages - 1:
choices.append(Choice("Next Page >", command="next"))
choices.append(Choice("Cancel", command="cancel"))
result = app.prompt(InquirerSelect(
f"Select item (Page {current_page + 1}/{total_pages}):",
choices
))
if result.command == 'prev':
current_page -= 1
elif result.command == 'next':
current_page += 1
elif result.command == 'cancel' or result.command == 'quit':
app.stop(None)
return
elif result.command == 'select':
app.stop(result.value.data)
return
app = InquirerApp()
result = app.run(inline=True, inquiry_func=paginated_selection_flow)def progress_tracking_flow(app):
"""Flow with progress indicators."""
total_steps = 5
completed = []
for step in range(1, total_steps + 1):
progress = f"[Step {step}/{total_steps}] Progress: {len(completed)}/{total_steps - 1} completed"
result = app.prompt(InquirerText(f"{progress}\nEnter value for step {step}:"))
if result.command != 'select':
app.stop(completed)
return
completed.append(result.value)
app.stop(completed)
app = InquirerApp()
result = app.run(inline=True, inquiry_func=progress_tracking_flow)def limited_retry_flow(app):
"""Flow with retry limits for validation."""
max_attempts = 3
attempts = 0
while attempts < max_attempts:
password = app.prompt(InquirerSecret("Password (min 8 chars):"))
if password.command != 'select':
return # User quit
if len(password.value) >= 8:
app.stop(password.value)
return
else:
attempts += 1
remaining = max_attempts - attempts
if remaining > 0:
app.prompt(InquirerText(f"Password too short! {remaining} attempts remaining. Press Enter."))
else:
app.prompt(InquirerText("Maximum attempts reached. Press Enter to exit."))
app.stop(None)
return
app = InquirerApp()
result = app.run(inline=True, inquiry_func=limited_retry_flow)def preview_confirmation_flow(app):
"""Collect data, preview, and confirm before submission."""
data = {}
# Collect data
data['name'] = app.prompt(InquirerText("Name:")).value
data['email'] = app.prompt(InquirerText("Email:")).value
data['role'] = app.prompt(InquirerSelect("Role:", ["User", "Admin"])).value
# Show preview
preview = f"""
Review your information:
- Name: {data['name']}
- Email: {data['email']}
- Role: {data['role']}
"""
confirm = app.prompt(InquirerConfirm(f"{preview}\nIs this correct?", default=True))
if confirm.value:
app.stop(data)
else:
# User can restart or cancel
retry = app.prompt(InquirerConfirm("Start over?", default=False))
if retry.value:
preview_confirmation_flow(app) # Recursive restart
else:
app.stop(None)
app = InquirerApp()
result = app.run(inline=True, inquiry_func=preview_confirmation_flow)Controls how the app is displayed in the terminal.
# inline=True: Embedded in terminal
# - Appears at current cursor position
# - Preserves terminal history above
# - Clears prompt after completion
# - Best for CLI tools and scripts
app.run(inline=True)
# inline=False (default): Full-screen TUI
# - Uses alternative screen buffer
# - Entire terminal taken over
# - Returns to normal screen when done
# - Best for complex applications
app.run(inline=False)Variant of inline that doesn't clear the prompt.
# inline_no_clear=True: Inline but leaves output visible
# - Prompt remains in terminal after completion
# - Useful for debugging or keeping record
app.run(inline_no_clear=True)Run without display (for testing).
# headless=True: No visual output
# - Useful for automated testing
# - No terminal interaction
# - Must provide auto_pilot for input
app.run(headless=True, auto_pilot=test_pilot)Enable/disable mouse support.
# mouse=True (default): Mouse interactions enabled
app.run(mouse=True)
# mouse=False: Keyboard only
# - Useful for environments without mouse support
app.run(mouse=False)Override terminal size.
# size=(width, height): Custom terminal size
# - Useful for testing different screen sizes
# - Overrides actual terminal dimensions
app.run(size=(80, 24)) # 80 columns x 24 rowsProvide custom event loop.
import asyncio
custom_loop = asyncio.new_event_loop()
app.run(loop=custom_loop)InquirerApp.__init__() calledapp.run() calledapp.stop() called with final valueInquirerWidget.Submit messageapp.prompt() returns Resultapp.prompt() multiple timesapp.stop() or return to exitdef flow(app):
result = app.prompt(InquirerText("Input:"))
# ALWAYS check command before using value
if result.command != 'select':
app.stop(None)
return
# Now safe to use result.value
app.stop(result.value)def flow(app):
r1 = app.prompt(InquirerText("First:"))
if r1.command != 'select':
return # User quit - don't call app.stop()
r2 = app.prompt(InquirerText("Second:"))
if r2.command != 'select':
return
app.stop((r1.value, r2.value))def flow(app):
r1 = app.prompt(InquirerText("First:"))
app.stop(r1.value)
# WRONG - This raises RuntimeError
# r2 = app.prompt(InquirerText("Second:"))
# CORRECT - Return after stop
returndef ask_required_text(app, message):
"""Reusable helper for required text input."""
while True:
result = app.prompt(InquirerText(message))
if result.command != 'select':
return None
if result.value.strip():
return result.value
app.prompt(InquirerText("Required! Press Enter to retry."))
def flow(app):
name = ask_required_text(app, "Name:")
if name:
app.stop(name)class FlowContext:
"""Context for managing flow state."""
def __init__(self):
self.data = {}
self.metadata = {}
def set(self, key, value, metadata=None):
self.data[key] = value
if metadata:
self.metadata[key] = metadata
def get(self, key, default=None):
return self.data.get(key, default)
def flow(app):
ctx = FlowContext()
name = app.prompt(InquirerText("Name:"))
if name.command == 'select':
ctx.set('name', name.value, {'timestamp': time.time()})
app.stop(ctx.data)