Execution Control and Batch Processing

def prompt(questions: Union[Dict, Iterable[Mapping]], 
           answers: Optional[Mapping] = None, patch_stdout: bool = False, 
           true_color: bool = False, kbi_msg: str = DEFAULT_KBI_MESSAGE, 
           **kwargs) -> Dict[str, Any]:
    """
    Execute multiple prompts synchronously with error handling.
    
    Args:
        questions: Questions configuration as dict or iterable of mappings
        answers: Pre-filled answers to skip certain questions
        patch_stdout: Patch stdout to prevent interference
        true_color: Enable true color support
        kbi_msg: Message displayed on keyboard interrupt
        **kwargs: Additional arguments passed to individual prompts
        
    Returns:
        Dictionary mapping question names to user responses
        
    Raises:
        KeyboardInterrupt: User cancelled with appropriate message
        PromptParameterException: Missing required prompt parameters
    """

def prompt_async(questions: Union[Dict, Iterable[Mapping]], 
                 answers: Optional[Mapping] = None, patch_stdout: bool = False, 
                 true_color: bool = False, kbi_msg: str = DEFAULT_KBI_MESSAGE, 
                 **kwargs) -> Dict[str, Any]:
    """
    Execute multiple prompts asynchronously with error handling.
    
    Args:
        questions: Questions configuration as dict or iterable of mappings
        answers: Pre-filled answers to skip certain questions
        patch_stdout: Patch stdout to prevent interference  
        true_color: Enable true color support
        kbi_msg: Message displayed on keyboard interrupt
        **kwargs: Additional arguments passed to individual prompts
        
    Returns:
        Dictionary mapping question names to user responses
        
    Raises:
        KeyboardInterrupt: User cancelled with appropriate message
        PromptParameterException: Missing required prompt parameters
    """

def unsafe_prompt(questions: Union[Dict, Iterable[Mapping]], 
                  answers: Optional[Mapping] = None, patch_stdout: bool = False, 
                  true_color: bool = False, **kwargs) -> Dict[str, Any]:
    """
    Execute multiple prompts synchronously without error handling.
    
    Args:
        questions: Questions configuration as dict or iterable of mappings
        answers: Pre-filled answers to skip certain questions
        patch_stdout: Patch stdout to prevent interference
        true_color: Enable true color support
        **kwargs: Additional arguments passed to individual prompts
        
    Returns:
        Dictionary mapping question names to user responses
    """

def unsafe_prompt_async(questions: Union[Dict, Iterable[Mapping]], 
                        answers: Optional[Mapping] = None, patch_stdout: bool = False, 
                        true_color: bool = False, **kwargs) -> Dict[str, Any]:
    """
    Execute multiple prompts asynchronously without error handling.
    
    Args:
        questions: Questions configuration as dict or iterable of mappings
        answers: Pre-filled answers to skip certain questions
        patch_stdout: Patch stdout to prevent interference
        true_color: Enable true color support
        **kwargs: Additional arguments passed to individual prompts
        
    Returns:
        Dictionary mapping question names to user responses
    """

class Question:
    def ask(self, patch_stdout: bool = False, 
            kbi_msg: str = DEFAULT_KBI_MESSAGE) -> Any:
        """
        Execute question synchronously with error handling.
        
        Args:
            patch_stdout: Patch stdout to prevent interference
            kbi_msg: Message displayed on keyboard interrupt
            
        Returns:
            User response value
            
        Raises:
            KeyboardInterrupt: User cancelled with appropriate message
        """

    def unsafe_ask(self, patch_stdout: bool = False) -> Any:
        """
        Execute question synchronously without error handling.
        
        Args:
            patch_stdout: Patch stdout to prevent interference
            
        Returns:
            User response value
        """

    def ask_async(self, patch_stdout: bool = False, 
                  kbi_msg: str = DEFAULT_KBI_MESSAGE) -> Any:
        """
        Execute question asynchronously with error handling.
        
        Args:
            patch_stdout: Patch stdout to prevent interference
            kbi_msg: Message displayed on keyboard interrupt
            
        Returns:
            User response value
            
        Raises:
            KeyboardInterrupt: User cancelled with appropriate message
        """

    def unsafe_ask_async(self, patch_stdout: bool = False) -> Any:
        """
        Execute question asynchronously without error handling.
        
        Args:
            patch_stdout: Patch stdout to prevent interference
            
        Returns:
            User response value
        """

    def skip_if(self, condition: bool, default: Any = None) -> Question:
        """
        Conditionally skip question execution.
        
        Args:
            condition: If True, skip question and return default
            default: Value to return when question is skipped
            
        Returns:
            Self for method chaining
        """

class PromptParameterException(ValueError):
    """
    Exception raised when prompt parameters are missing or invalid.
    
    Inherits from ValueError for compatibility with standard error handling.
    """

import questionary

# Questions as dictionary
questions = {
    "name": {
        "type": "text",
        "message": "What's your name?"
    },
    "age": {
        "type": "text", 
        "message": "What's your age?",
        "validate": lambda x: "Must be a number" if not x.isdigit() else True
    },
    "confirmed": {
        "type": "confirm",
        "message": "Is this correct?"
    }
}

answers = questionary.prompt(questions)
print(f"Hello {answers['name']}, age {answers['age']}")

import questionary

# Questions as list for ordered execution
questions = [
    {
        "type": "select",
        "name": "deployment_type", 
        "message": "Deployment type:",
        "choices": ["development", "staging", "production"]
    },
    {
        "type": "text",
        "name": "app_name",
        "message": "Application name:",
        "default": "my-app"
    },
    {
        "type": "confirm",
        "name": "auto_deploy",
        "message": "Enable automatic deployment?"
    }
]

config = questionary.prompt(questions)
print(f"Deploying {config['app_name']} to {config['deployment_type']}")

import questionary

questions = {
    "database_host": {
        "type": "text",
        "message": "Database host:",
        "default": "localhost"
    },
    "database_port": {
        "type": "text", 
        "message": "Database port:",
        "default": "5432"
    },
    "database_name": {
        "type": "text",
        "message": "Database name:"
    }
}

# Pre-fill some answers
existing_config = {
    "database_host": "prod-db.example.com",
    "database_port": "3306"
}

# Only database_name will be prompted
final_config = questionary.prompt(questions, answers=existing_config)
print("Database config:", final_config)

import questionary
import asyncio

async def async_setup():
    questions = {
        "service_name": {
            "type": "text",
            "message": "Service name:"
        },
        "replicas": {
            "type": "select",
            "message": "Number of replicas:",
            "choices": ["1", "3", "5", "10"]
        },
        "monitoring": {
            "type": "confirm", 
            "message": "Enable monitoring?"
        }
    }
    
    config = await questionary.prompt_async(questions)
    print(f"Configuring {config['service_name']} with {config['replicas']} replicas")
    return config

# Run async
# config = asyncio.run(async_setup())

import questionary

# Create question instances
name_question = questionary.text("Enter your name:")
age_question = questionary.text("Enter your age:")

# Execute with different methods
try:
    name = name_question.ask()
    age = age_question.ask()
    print(f"Hello {name}, age {age}")
except KeyboardInterrupt:
    print("Setup cancelled")

# Using unsafe methods (no error handling)
backup_name = name_question.unsafe_ask()

import questionary

# Ask initial question
user_type = questionary.select(
    "User type:",
    choices=["admin", "regular"]
).ask()

# Conditional execution based on response
admin_password = questionary.password("Admin password:").skip_if(
    condition=user_type != "admin",
    default=None
)

password = admin_password.ask()
if password:
    print("Admin access granted")
else:
    print("Regular user mode")

import questionary
from questionary import PromptParameterException

def safe_prompt_execution():
    questions = {
        "critical_setting": {
            "type": "confirm",
            "message": "Enable critical feature?",
            "default": False
        }
    }
    
    try:
        answers = questionary.prompt(
            questions,
            kbi_msg="Configuration cancelled - using defaults"
        )
        return answers
        
    except KeyboardInterrupt:
        print("User cancelled - applying safe defaults")
        return {"critical_setting": False}
        
    except PromptParameterException as e:
        print(f"Configuration error: {e}")
        return None
        
    except Exception as e:
        print(f"Unexpected error: {e}")
        return None

result = safe_prompt_execution()

import questionary

def interactive_wizard():
    """Multi-step wizard with branching logic"""
    
    # Step 1: Choose workflow type
    workflow = questionary.select(
        "What would you like to configure?",
        choices=["database", "web_server", "cache"]
    ).ask()
    
    base_questions = {
        "environment": {
            "type": "select",
            "message": "Environment:",
            "choices": ["development", "staging", "production"]
        }
    }
    
    # Step 2: Environment-specific questions
    if workflow == "database":
        db_questions = {
            **base_questions,
            "db_type": {
                "type": "select",
                "message": "Database type:",
                "choices": ["postgresql", "mysql", "mongodb"]
            },
            "backup_enabled": {
                "type": "confirm",
                "message": "Enable automated backups?"
            }
        }
        config = questionary.prompt(db_questions)
        
    elif workflow == "web_server":
        web_questions = {
            **base_questions,
            "port": {
                "type": "text",
                "message": "Port number:",
                "default": "8080"
            },
            "ssl_enabled": {
                "type": "confirm",
                "message": "Enable SSL?"
            }
        }
        config = questionary.prompt(web_questions)
        
    else:  # cache
        cache_questions = {
            **base_questions,
            "cache_size": {
                "type": "select",
                "message": "Cache size:",
                "choices": ["128MB", "512MB", "1GB", "4GB"]
            }
        }
        config = questionary.prompt(cache_questions)
    
    # Step 3: Confirmation
    confirmed = questionary.confirm(
        f"Apply {workflow} configuration?",
        default=True
    ).ask()
    
    if confirmed:
        print(f"Applying {workflow} configuration:", config)
        return config
    else:
        print("Configuration cancelled")
        return None

# Run the wizard
# final_config = interactive_wizard()

import questionary

# Control stdout patching for integration with other tools
questions = {
    "setting": {
        "type": "text",
        "message": "Enter setting value:"
    }
}

# Patch stdout to prevent interference with other output
config = questionary.prompt(questions, patch_stdout=True)

# Enable true color support for better styling
styled_config = questionary.prompt(
    questions, 
    true_color=True,
    patch_stdout=True
)