ts_executor.py•9.75 kB
#!/usr/bin/env python3
"""
TypeScript Executor - Execute TypeScript methods from Python
Supports calling TypeScript functions with complex parameters and async/await
"""
import json
import os
import subprocess
import tempfile
from pathlib import Path
from typing import Any, Dict, List, Optional
class TypeScriptExecutor:
"""TypeScript method executor"""
def __init__(self, working_dir: Optional[str] = None):
"""Initialize TypeScript executor
Args:
working_dir: Working directory, defaults to current directory
"""
self.working_dir = working_dir or os.getcwd()
def call_method(self, ts_file_path: str, method_name: str, *args, **kwargs) -> Any:
"""Call TypeScript method
Args:
ts_file_path: TypeScript file path
method_name: Method name
*args: Positional arguments
**kwargs: Keyword arguments
Returns:
Execution result
"""
# Convert relative path to absolute path
if not os.path.isabs(ts_file_path):
ts_file_path = os.path.join(self.working_dir, ts_file_path)
# Ensure the target file exists
if not os.path.exists(ts_file_path):
raise FileNotFoundError(f"TypeScript file not found: {ts_file_path}")
# Get the directory of the target file
target_dir = os.path.dirname(ts_file_path)
# Create wrapper script
wrapper_code = self._create_wrapper_script(
ts_file_path, method_name, list(args), kwargs
)
# Create temporary file in the same directory as the target file
temp_fd, temp_file = tempfile.mkstemp(suffix=".ts", dir=target_dir)
try:
# Write wrapper script
with os.fdopen(temp_fd, "w", encoding="utf-8") as f:
f.write(wrapper_code)
# Execute TypeScript code using ts-node
# Use subprocess.Popen to capture output in real-time
process = subprocess.Popen(
["npx", "ts-node", temp_file],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
cwd=self.working_dir,
bufsize=1, # Line buffering
universal_newlines=True,
)
stdout_lines = []
stderr_lines = []
# Read output line by line and display console.log in real-time
while True:
output = process.stdout.readline()
if output == "" and process.poll() is not None:
break
if output:
line = output.strip()
stdout_lines.append(line)
# Try to parse as JSON to see if it's the final result
try:
json.loads(line)
# If it parses as JSON, it might be the final result, don't print it yet
except json.JSONDecodeError:
# If it's not JSON, it's likely a console.log, so print it
print(line)
# Get any remaining stderr
stderr_output = process.stderr.read()
if stderr_output:
stderr_lines.append(stderr_output.strip())
return_code = process.poll()
if return_code != 0:
error_msg = "\n".join(stderr_lines) if stderr_lines else "Unknown error"
raise RuntimeError(f"TypeScript execution failed: {error_msg}")
# Parse results from the last line that looks like JSON
for line in reversed(stdout_lines):
if line.strip():
try:
# Try to parse as JSON
return json.loads(line)
except json.JSONDecodeError:
continue
# If no JSON found, return the last non-empty line
for line in reversed(stdout_lines):
if line.strip():
return line
return None
except Exception as e:
raise RuntimeError(f"Execution error: {str(e)}")
finally:
# Clean up temporary file
os.unlink(temp_file)
def _create_wrapper_script(
self,
ts_file_path: str,
method_name: str,
args: List[Any],
kwargs: Dict[str, Any],
) -> str:
"""Create wrapper script
Args:
ts_file_path: TypeScript file path
method_name: Method name
args: Positional arguments
kwargs: Keyword arguments
Returns:
Wrapper script code
"""
# Use relative path for import, since temp file is in the same directory
ts_filename = os.path.basename(ts_file_path)
# Remove .ts extension, since import doesn't need it
if ts_filename.endswith(".ts"):
import_path = "./" + ts_filename[:-3]
else:
import_path = "./" + ts_filename
args_json = json.dumps(args)
kwargs_json = json.dumps(kwargs)
wrapper_code = f"""
import * as targetModule from '{import_path}';
async function executeMethod() {{
try {{
// Prepare arguments
const args: any[] = {args_json};
const kwargs: any = {kwargs_json};
// Get method
const method = (targetModule as any).{method_name};
if (typeof method !== 'function') {{
throw new Error(`Method '{method_name}' does not exist or is not a function`);
}}
// Call method
let result: any;
if (Object.keys(kwargs).length > 0) {{
// If there are keyword arguments, pass them as the last parameter
result = await method(...args, kwargs);
}} else {{
// Only positional arguments
result = await method(...args);
}}
// Output result
console.log(JSON.stringify(result));
}} catch (error: any) {{
console.error(JSON.stringify({{
error: (error as Error).message,
stack: (error as Error).stack
}}));
process.exit(1);
}}
}}
executeMethod();
"""
return wrapper_code
# Convenience function
def call_ts_method(
ts_file: str, method_name: str, *args, working_dir: Optional[str] = None, **kwargs
) -> Any:
"""Convenience function: Call TypeScript method
Args:
ts_file: TypeScript file path
method_name: Method name
*args: Positional arguments
working_dir: Working directory
**kwargs: Keyword arguments
Returns:
Execution result
"""
executor = TypeScriptExecutor(working_dir)
return executor.call_method(ts_file, method_name, *args, **kwargs)
# Usage example
if __name__ == "__main__":
# Create test TypeScript file
test_ts_content = """
export function add(a: number, b: number): number {
return a + b;
}
export function greet(name: string, options?: { formal?: boolean }): string {
const greeting = options?.formal ? "Hello" : "Hi";
return `${greeting}, ${name}!`;
}
export async function processData(data: any[]): Promise<{ count: number; items: any[] }> {
// Simulate async processing
await new Promise(resolve => setTimeout(resolve, 100));
return {
count: data.length,
items: data.map(item => ({ processed: true, original: item }))
};
}
export function complexFunction(
numbers: number[],
config: { multiplier: number; offset: number }
): { result: number[]; sum: number } {
const result = numbers.map(n => n * config.multiplier + config.offset);
const sum = result.reduce((a, b) => a + b, 0);
return { result, sum };
}
"""
# Write test file
with open("test_methods.ts", "w") as f:
f.write(test_ts_content)
try:
# Create executor
executor = TypeScriptExecutor()
print("=== TypeScript Method Execution Test ===")
# Test 1: Simple function
print("\n1. Testing simple addition function:")
result = executor.call_method("test_methods.ts", "add", 10, 20)
print(f" add(10, 20) = {result}")
# Test 2: Function with optional parameters
print("\n2. Testing greeting function:")
result1 = executor.call_method("test_methods.ts", "greet", "Alice")
print(f" greet('Alice') = {result1}")
result2 = executor.call_method(
"test_methods.ts", "greet", "Bob", {"formal": True}
)
print(f" greet('Bob', {{formal: true}}) = {result2}")
# Test 3: Async function
print("\n3. Testing async function:")
result = executor.call_method(
"test_methods.ts", "processData", [1, 2, 3, "hello"]
)
print(f" processData([1, 2, 3, 'hello']) = {result}")
# Test 4: Complex function
print("\n4. Testing complex function:")
result = executor.call_method(
"test_methods.ts",
"complexFunction",
[1, 2, 3, 4, 5],
{"multiplier": 2, "offset": 1},
)
print(f" complexFunction([1,2,3,4,5], {{multiplier:2, offset:1}}) = {result}")
# Test 5: Using convenience function
print("\n5. Testing convenience function:")
result = call_ts_method("test_methods.ts", "add", 100, 200)
print(f" call_ts_method('test_methods.ts', 'add', 100, 200) = {result}")
except Exception as e:
print(f"Error: {e}")
finally:
# Clean up test file
if os.path.exists("test_methods.ts"):
os.remove("test_methods.ts")