Python-powered shell providing superset of Python with shell primitives for cross-platform command execution and automation.
Xonsh provides a comprehensive event system that enables customization and extension of shell behavior through event handlers. Events are fired at key points during command execution, shell lifecycle, and environment changes, allowing for powerful customization and automation.
from xonsh.events import AbstractEvent, Event, LoadEvent, EventManager
class AbstractEvent(collections.abc.MutableSet):
"""Abstract base class for xonsh events."""
def fire(self, **kwargs) -> None:
"""Fire event with keyword arguments.
Parameters
----------
**kwargs
Event-specific arguments passed to handlers
"""
def add(self, func: callable) -> None:
"""Add event handler function.
Parameters
----------
func : callable
Handler function to add
"""
def discard(self, func: callable) -> None:
"""Remove event handler function.
Parameters
----------
func : callable
Handler function to remove
"""
class Event(AbstractEvent):
"""Standard event implementation."""
def __init__(self, name: str = None, doc: str = None,
prio: int = 0, traceback: bool = True):
"""Create event.
Parameters
----------
name : str, optional
Event name
doc : str, optional
Event documentation
prio : int, default 0
Event priority for handler ordering
traceback : bool, default True
Whether to show tracebacks on handler errors
"""
class LoadEvent(AbstractEvent):
"""Event that loads handlers from configuration."""
def __init__(self, name: str = None, doc: str = None,
prio: int = 0, traceback: bool = True):
"""Create load event with configuration loading."""
class EventManager:
"""Central event manager for xonsh."""
def __init__(self):
"""Initialize event manager."""
def __getattr__(self, name: str) -> Event:
"""Get or create event by name.
Parameters
----------
name : str
Event name
Returns
-------
Event
Event instance
"""from xonsh.events import events
# Global event manager instance
events: EventManager # Main event manager
# Event registration patterns
@events.on_precommand
def my_precommand_handler(cmd: str) -> None:
"""Handler for precommand event."""
pass
# Alternative registration
def my_handler(**kwargs):
"""Generic event handler."""
pass
events.on_postcommand.add(my_handler)from xonsh.events import events
@events.on_transform_command
def transform_command_handler(cmd: str, **kwargs) -> str:
"""Transform command before parsing.
Parameters
----------
cmd : str
Original command string
**kwargs
Additional context
Returns
-------
str
Transformed command string
"""
# Example: expand abbreviations
if cmd.startswith('g '):
return 'git ' + cmd[2:]
return cmd
@events.on_precommand
def precommand_handler(cmd: str, **kwargs) -> None:
"""Called before command execution.
Parameters
----------
cmd : str
Command to be executed
**kwargs
Execution context
"""
print(f"About to execute: {cmd}")
@events.on_postcommand
def postcommand_handler(cmd: str, rtn: int, out: str = None,
ts: list = None, **kwargs) -> None:
"""Called after command execution.
Parameters
----------
cmd : str
Executed command
rtn : int
Return code (0 for success)
out : str, optional
Command output if captured
ts : list, optional
Timestamps [start_time, end_time]
**kwargs
Additional context
"""
if rtn != 0:
print(f"Command failed with return code {rtn}")
@events.on_command_not_found
def command_not_found_handler(cmd: list[str], **kwargs) -> None:
"""Called when command is not found.
Parameters
----------
cmd : list[str]
Command arguments that were not found
**kwargs
Context information
"""
cmd_name = cmd[0] if cmd else "unknown"
print(f"Command '{cmd_name}' not found. Did you mean something else?")@events.on_pre_prompt_format
def pre_prompt_format_handler(**kwargs) -> None:
"""Called before prompt formatting.
Parameters
----------
**kwargs
Prompt context
"""
# Prepare prompt context
pass
@events.on_pre_prompt
def pre_prompt_handler(**kwargs) -> None:
"""Called just before showing prompt.
Parameters
----------
**kwargs
Prompt state
"""
# Last-minute prompt customization
pass
@events.on_post_prompt
def post_prompt_handler(**kwargs) -> None:
"""Called after prompt input is received.
Parameters
----------
**kwargs
Input context
"""
# Process prompt input
pass@events.on_chdir
def chdir_handler(olddir: str, newdir: str, **kwargs) -> None:
"""Called when current directory changes.
Parameters
----------
olddir : str
Previous directory path
newdir : str
New directory path
**kwargs
Change context
"""
print(f"Directory changed: {olddir} -> {newdir}")
# Auto-activate virtual environments
import os
venv_activate = os.path.join(newdir, 'venv', 'bin', 'activate')
if os.path.exists(venv_activate):
print("Virtual environment detected")@events.on_envvar_new
def envvar_new_handler(name: str, value: str, **kwargs) -> None:
"""Called when new environment variable is set.
Parameters
----------
name : str
Variable name
value : str
Variable value
**kwargs
Context information
"""
if name.startswith('PROJECT_'):
print(f"Project variable set: {name}={value}")
@events.on_envvar_change
def envvar_change_handler(name: str, oldvalue: str, newvalue: str,
**kwargs) -> None:
"""Called when environment variable changes.
Parameters
----------
name : str
Variable name
oldvalue : str
Previous value
newvalue : str
New value
**kwargs
Context information
"""
if name == 'PATH':
print("PATH was modified")def filtered_event_handler(cmd: str, **kwargs) -> None:
"""Event handler with filtering logic."""
# Only handle git commands
if not cmd.startswith('git '):
return
# Extract git subcommand
parts = cmd.split()
if len(parts) > 1:
subcmd = parts[1]
print(f"Git subcommand: {subcmd}")
@events.on_precommand
def validate_command_handler(cmd: str, **kwargs) -> None:
"""Validate commands before execution."""
dangerous_commands = ['rm -rf /', 'dd if=/dev/zero']
for dangerous in dangerous_commands:
if dangerous in cmd:
print(f"WARNING: Potentially dangerous command: {cmd}")
response = input("Are you sure? (y/N): ")
if response.lower() != 'y':
raise KeyboardInterrupt("Command cancelled by user")from xonsh.built_ins import XSH
def development_mode_handler(cmd: str, **kwargs) -> None:
"""Handler that only runs in development mode."""
env = XSH.env
if not env.get('DEVELOPMENT_MODE'):
return
# Development-specific logging
with open('/tmp/dev_commands.log', 'a') as f:
import datetime
timestamp = datetime.datetime.now().isoformat()
f.write(f"{timestamp}: {cmd}\n")
events.on_precommand.add(development_mode_handler)def create_logging_handler(logfile: str):
"""Factory function for logging handlers."""
def logging_handler(cmd: str, **kwargs):
with open(logfile, 'a') as f:
import datetime
timestamp = datetime.datetime.now().isoformat()
f.write(f"{timestamp}: {cmd}\n")
return logging_handler
# Create specialized loggers
git_logger = create_logging_handler('/tmp/git_commands.log')
system_logger = create_logging_handler('/tmp/system_commands.log')
@events.on_precommand
def dispatch_logging(cmd: str, **kwargs):
"""Dispatch to appropriate logger based on command."""
if cmd.startswith('git '):
git_logger(cmd, **kwargs)
elif cmd.startswith(('ls', 'cd', 'pwd')):
system_logger(cmd, **kwargs)def register_project_handlers():
"""Register project-specific event handlers."""
@events.on_chdir
def project_chdir_handler(olddir, newdir, **kwargs):
# Check for project files
import os
if os.path.exists(os.path.join(newdir, 'pyproject.toml')):
print("Entered Python project directory")
elif os.path.exists(os.path.join(newdir, 'package.json')):
print("Entered Node.js project directory")
return project_chdir_handler
# Register conditionally
if XSH.env.get('ENABLE_PROJECT_DETECTION'):
handler = register_project_handlers()# Store handler references for later removal
active_handlers = []
def temporary_debug_handler(cmd: str, **kwargs):
"""Temporary debugging handler."""
print(f"DEBUG: {cmd}")
# Add handler and store reference
events.on_precommand.add(temporary_debug_handler)
active_handlers.append(temporary_debug_handler)
# Remove handler later
def cleanup_handlers():
"""Remove temporary handlers."""
for handler in active_handlers:
events.on_precommand.discard(handler)
active_handlers.clear()def load_xontrib_with_events():
"""Load xontrib and register its event handlers."""
from xonsh.xontribs import xontribs_load
# Load xontrib
xontribs_load(['my_extension'])
# Register extension-specific handlers
@events.on_postcommand
def extension_postcommand(cmd, rtn, **kwargs):
# Extension-specific post-command processing
passdef configure_event_system():
"""Configure event system based on user preferences."""
env = XSH.env
# Command timing
if env.get('ENABLE_COMMAND_TIMING'):
@events.on_precommand
def start_timer(**kwargs):
import time
XSH._command_start_time = time.time()
@events.on_postcommand
def end_timer(cmd, **kwargs):
if hasattr(XSH, '_command_start_time'):
duration = time.time() - XSH._command_start_time
if duration > 1.0: # Only show for slow commands
print(f"Command took {duration:.2f} seconds")
# Command history enhancement
if env.get('ENHANCED_HISTORY'):
@events.on_postcommand
def enhanced_history(cmd, rtn, **kwargs):
if rtn == 0: # Only successful commands
# Add to enhanced history with metadata
passdef robust_event_handler(cmd: str, **kwargs) -> None:
"""Event handler with proper error handling."""
try:
# Handler logic - example processing
print(f"Processing command: {cmd}")
except Exception as e:
# Log error but don't break event chain
import logging
logging.error(f"Event handler error: {e}")
# Optionally re-raise for critical errors
# raise
# Event system handles exceptions gracefully
events.on_precommand.add(robust_event_handler)The event system provides powerful hooks into xonsh's execution flow, enabling sophisticated customization, automation, and extension of shell behavior through a clean, composable interface.
Install with Tessl CLI
npx tessl i tessl/pypi-xonsh