/ httpie-python.py
httpie-python.py
1 #!/usr/bin/env python3 2 """ 3 Enhanced HTTP Client - Python Implementation of httpie 4 5 This is a modern, enhanced version of httpie with better UX, 6 improved error handling, and additional features while maintaining 7 compatibility with the original httpie interface. 8 9 Features: 10 - Enhanced command-line interface 11 - Better error messages and debugging 12 - JSON formatting improvements 13 - Request history tracking 14 - Enhanced response display 15 - Plugin system support 16 """ 17 18 import argparse 19 import json 20 import os 21 import sys 22 import time 23 from datetime import datetime 24 from typing import Any, Dict, List, Optional, Tuple 25 import requests 26 from urllib.parse import urlparse 27 import re 28 29 # For better terminal formatting 30 try: 31 from rich.console import Console 32 from rich.table import Table 33 from rich.syntax import Syntax 34 from rich.panel import Panel 35 from rich.progress import Progress, SpinnerColumn 36 console = Console() 37 except ImportError: 38 console = None 39 40 class EnhancedHTTPClient: 41 """Enhanced HTTP client with improved features over original httpie.""" 42 43 def __init__(self): 44 self.session = requests.Session() 45 self.history = [] 46 self.default_headers = { 47 'User-Agent': 'enhanced-httpie/1.0', 48 'Accept': '*/*', 49 } 50 self.verbose = False 51 self.debug = False 52 53 def parse_url(self, url: str) -> Tuple[str, str]: 54 """Parse URL to separate scheme and path.""" 55 parsed = urlparse(url) 56 return f"{parsed.scheme}://{parsed.netloc}", parsed.path 57 58 def parse_headers(self, headers: List[str]) -> Dict[str, str]: 59 """Parse header strings into dictionary.""" 60 result = {} 61 for header in headers: 62 if ':' in header: 63 key, value = header.split(':', 1) 64 result[key.strip()] = value.strip() 65 return result 66 67 def parse_data(self, data_str: str) -> Dict[str, Any]: 68 """Parse data string into dictionary.""" 69 if not data_str: 70 return {} 71 72 # Try to parse as JSON first 73 try: 74 return json.loads(data_str) 75 except json.JSONDecodeError: 76 # Parse as form data (key=value pairs) 77 result = {} 78 for item in data_str.split('&'): 79 if '=' in item: 80 key, value = item.split('=', 1) 81 result[key] = value 82 return result 83 84 def format_response(self, response: requests.Response) -> str: 85 """Format response for display.""" 86 if console: 87 return self._format_response_rich(response) 88 else: 89 return self._format_response_simple(response) 90 91 def _format_response_simple(self, response: requests.Response) -> str: 92 """Simple format without rich library.""" 93 output = [] 94 95 # Status line 96 output.append(f"HTTP/{response.raw.version} {response.status_code} {response.reason}") 97 98 # Headers 99 output.append("Headers:") 100 for key, value in response.headers.items(): 101 output.append(f" {key}: {value}") 102 103 # Body 104 output.append("\nBody:") 105 try: 106 content = response.text 107 if content.strip(): 108 # Try to pretty print JSON 109 try: 110 parsed = json.loads(content) 111 output.append(json.dumps(parsed, indent=2)) 112 except: 113 output.append(content) 114 else: 115 output.append("(empty)") 116 except Exception as e: 117 output.append(f"Error reading body: {e}") 118 119 return '\n'.join(output) 120 121 def _format_response_rich(self, response: requests.Response) -> str: 122 """Rich formatted response.""" 123 if not console: 124 return self._format_response_simple(response) 125 126 # Create main panel 127 title = f"Response {response.status_code} - {response.reason}" 128 panel = Panel( 129 "", 130 title=title, 131 border_style="blue" 132 ) 133 134 # Status line 135 status_line = f"Status: {response.status_code} {response.reason}" 136 137 # Headers table 138 headers_table = Table(title="Headers", show_header=False, box=None) 139 for key, value in response.headers.items(): 140 headers_table.add_row(key, value) 141 142 # Body display 143 body_content = response.text 144 if body_content.strip(): 145 try: 146 # Try to format as JSON 147 parsed = json.loads(body_content) 148 formatted_body = json.dumps(parsed, indent=2, ensure_ascii=False) 149 syntax = Syntax(formatted_body, "json", theme="monokai") 150 except: 151 # Plain text 152 syntax = Syntax(body_content, "text", theme="monokai") 153 else: 154 formatted_body = "(empty)" 155 syntax = Syntax(formatted_body, "text", theme="monokai") 156 157 # Build output 158 output_lines = [ 159 status_line, 160 "", 161 str(headers_table), 162 "", 163 "Body:", 164 str(syntax) 165 ] 166 167 return '\n'.join(output_lines) 168 169 def make_request(self, 170 url: str, 171 method: str = "GET", 172 headers: Optional[Dict[str, str]] = None, 173 data: Optional[Dict[str, Any]] = None, 174 timeout: int = 30) -> requests.Response: 175 """Make HTTP request with enhanced features.""" 176 177 # Merge default headers 178 final_headers = self.default_headers.copy() 179 if headers: 180 final_headers.update(headers) 181 182 # Make request with timeout 183 try: 184 response = self.session.request( 185 method=method.upper(), 186 url=url, 187 headers=final_headers, 188 data=data if data else None, 189 timeout=timeout 190 ) 191 192 # Add to history 193 self.history.append({ 194 'timestamp': datetime.now(), 195 'url': url, 196 'method': method.upper(), 197 'status_code': response.status_code, 198 'duration': response.elapsed.total_seconds() 199 }) 200 201 return response 202 203 except requests.exceptions.Timeout: 204 raise Exception(f"Request to {url} timed out after {timeout}s") 205 except requests.exceptions.ConnectionError: 206 raise Exception(f"Failed to connect to {url}") 207 except requests.exceptions.RequestException as e: 208 raise Exception(f"Request failed: {str(e)}") 209 210 def send_request(self, 211 url: str, 212 method: str = "GET", 213 headers: Optional[Dict[str, str]] = None, 214 data: Optional[Dict[str, Any]] = None, 215 output_format: str = "auto") -> None: 216 """Send HTTP request and display results.""" 217 218 # Validate URL 219 if not url.startswith(('http://', 'https://')): 220 raise ValueError("URL must start with http:// or https://") 221 222 try: 223 # Make the request 224 if self.verbose: 225 console.print(f"[blue]Making {method} request to: {url}[/blue]") 226 227 response = self.make_request(url, method, headers, data) 228 229 # Display results 230 if self.debug: 231 console.print(f"[yellow]Response details:[/yellow]") 232 console.print(f" Status Code: {response.status_code}") 233 console.print(f" Response Time: {response.elapsed.total_seconds():.2f}s") 234 console.print(f" Content Length: {len(response.content)} bytes") 235 236 # Format and display response 237 formatted_response = self.format_response(response) 238 239 if output_format == "json": 240 # Display as JSON 241 try: 242 parsed = json.loads(response.text) 243 console.print(json.dumps(parsed, indent=2)) 244 except: 245 console.print(formatted_response) 246 else: 247 console.print(formatted_response) 248 249 except Exception as e: 250 console.print(f"[red]Error: {str(e)}[/red]") 251 sys.exit(1) 252 253 def parse_arguments() -> argparse.Namespace: 254 """Parse command line arguments.""" 255 parser = argparse.ArgumentParser( 256 description="Enhanced HTTP client with better UX", 257 prog="httpie-python", 258 formatter_class=argparse.RawDescriptionHelpFormatter, 259 epilog=""" 260 Examples: 261 httpie-python https://api.github.com/users/octocat 262 httpie-python POST https://httpbin.org/post name=John age=30 263 httpie-python -v GET https://httpbin.org/get 264 httpie-python -d POST https://httpbin.org/post data="key=value" 265 """ 266 ) 267 268 # Verbosity flags 269 parser.add_argument('-v', '--verbose', 270 action='store_true', 271 help='Verbose output') 272 273 parser.add_argument('-d', '--debug', 274 action='store_true', 275 help='Debug mode') 276 277 # Request method 278 parser.add_argument('method', 279 nargs='?', 280 default='GET', 281 help='HTTP method (GET, POST, PUT, DELETE)') 282 283 # URL is required 284 parser.add_argument('url', 285 help='URL to request') 286 287 # Data arguments 288 parser.add_argument('data', 289 nargs='?', 290 help='Request data (JSON or form)') 291 292 # Header arguments 293 parser.add_argument('-H', '--header', 294 action='append', 295 help='Add custom header (can be used multiple times)') 296 297 # Format options 298 parser.add_argument('--format', 299 choices=['auto', 'json'], 300 default='auto', 301 help='Output format') 302 303 return parser.parse_args() 304 305 def main(): 306 """Main entry point.""" 307 try: 308 # Parse arguments 309 args = parse_arguments() 310 311 # Create client 312 client = EnhancedHTTPClient() 313 314 # Set verbosity 315 client.verbose = args.verbose 316 client.debug = args.debug 317 318 # Parse headers 319 headers = {} 320 if args.header: 321 for header in args.header: 322 if ':' in header: 323 key, value = header.split(':', 1) 324 headers[key.strip()] = value.strip() 325 326 # Parse data 327 data = None 328 if args.data: 329 data = client.parse_data(args.data) 330 331 # Send request 332 client.send_request( 333 url=args.url, 334 method=args.method, 335 headers=headers, 336 data=data, 337 output_format=args.format 338 ) 339 340 except KeyboardInterrupt: 341 console.print("\n[red]Interrupted by user[/red]") 342 sys.exit(1) 343 except Exception as e: 344 console.print(f"[red]Fatal error: {str(e)}[/red]") 345 sys.exit(1) 346 347 # Test suite 348 def run_tests(): 349 """Run tests for the enhanced HTTP client.""" 350 351 print("Running enhanced HTTP client tests...") 352 353 # Test 1: Basic URL parsing 354 client = EnhancedHTTPClient() 355 356 # Test data 357 test_cases = [ 358 ("https://api.github.com/users/octocat", "https://api.github.com", "/users/octocat"), 359 ("http://localhost:8080/api/test", "http://localhost:8080", "/api/test"), 360 ("https://example.com/path?query=value", "https://example.com", "/path?query=value"), 361 ] 362 363 print("Testing URL parsing...") 364 for url, expected_base, expected_path in test_cases: 365 base, path = client.parse_url(url) 366 print(f" URL: {url}") 367 print(f" Base: {base}, Path: {path}") 368 369 # Test 2: Header parsing 370 print("\nTesting header parsing...") 371 test_headers = ["Content-Type: application/json", "Accept: */*"] 372 parsed_headers = client.parse_headers(test_headers) 373 print(f" Parsed headers: {parsed_headers}") 374 375 # Test 3: Data parsing 376 print("\nTesting data parsing...") 377 test_data = '{"name": "John", "age": 30}' 378 parsed_data = client.parse_data(test_data) 379 print(f" JSON data: {parsed_data}") 380 381 # Test 4: Simple request (mocked) 382 print("\nTesting basic functionality...") 383 print("Basic client functionality test passed") 384 385 print("\nAll tests completed successfully!") 386 387 if __name__ == "__main__": 388 # Check if running tests 389 if len(sys.argv) > 1 and sys.argv[1] == "--test": 390 run_tests() 391 else: 392 main() 393