Marico's space

# Flawed initial implementation
cmd = [ sys.executable, "-c", "import sys; from __main__ import run_external_api_worker; run_external_api_worker(...)"
]

status: FAILED (Process exited with code 1: ImportError: cannot import name 'run_external_api_worker' from '__main__')

import sys
import time
import signal
import subprocess
from typing import Optional # Graceful Degradation: High-performance orjson fallback architecture
try: import orjson HAS_ORJSON = True
except ImportError: import json HAS_ORJSON = False from pydantic import BaseModel, Field class TaskPayload(BaseModel): task_id: str = Field(...,description="Unique task identifier") api_endpoint: str = Field(...,description="Target external API endpoint URI") timeout_seconds: float = Field(...,gt=0.0,description="Timeout limit in seconds") class TaskResult(BaseModel): task_id: str status: str message: str elapsed_time: float def serialize_data(data: dict, indent: bool = False) -> bytes: """Safe JSON serialization handling with automatic orjson fallback.""" if HAS_ORJSON: option = orjson.OPT_INDENT_2 if indent else 0 return orjson.dumps(data, option=option) else: if indent: return json.dumps(data, ensure_ascii=False, indent=2).encode("utf-8") return json.dumps(data, ensure_ascii=False).encode("utf-8") def deserialize_data(data_bytes: bytes) -> dict: """Safe JSON deserialization handling with automatic orjson fallback.""" if HAS_ORJSON: return orjson.loads(data_bytes) else: return json.loads(data_bytes.decode("utf-8")) def run_external_api_worker(payload_json_bytes: bytes) -> None: """ Subprocess entry point simulating an external workload. Handles SIGTERM signal propagation to ensure graceful resource cleanup. """ try: data = deserialize_data(payload_json_bytes) payload = TaskPayload(**data) except Exception as e: sys.stderr.write(f"Worker validation error: {e}\n") sys.exit(2) def handle_sigterm(signum, frame): # Graceful cleanup routine: Close network pools, flush buffers, release file descriptors
 sys.stderr.write(f"[{payload.task_id}] SIGTERM received. Cleaning up resources...\n") sys.stderr.write(f"[{payload.task_id}] Cleanup completed. Exiting.\n") sys.exit(143) # Register POSIX SIGTERM handler
 signal.signal(signal.SIGTERM, handle_sigterm) print(f"[{payload.task_id}] Starting request to {payload.api_endpoint}...", flush=True) try: elapsed = 0.0 interval = 0.1 while elapsed < payload.timeout_seconds: time.sleep(interval) elapsed += interval print(f"[{payload.task_id}] API request completed successfully.", flush=True) sys.exit(0) except Exception as e: sys.stderr.write(f"[{payload.task_id}] Exception occurred: {e}\n") sys.exit(1) finally: sys.stderr.write(f"[{payload.task_id}] Finally block executed.\n") def execute_task_with_timeout_propagation(payload: TaskPayload) -> TaskResult: """ Orchestrator boundary: Spawns the worker subprocess and manages timeout/signal propagation. Guarantees mitigation of orphaned processes and hanging threads. """ start_time = time.time() payload_bytes = serialize_data(payload.model_dump()) # Self-Execution pattern avoids module namespace and pickling pitfalls
 cmd = [ sys.executable, __file__, "--worker" ] process = subprocess.Popen( cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=False ) try: _, stderr_data = process.communicate(input=payload_bytes, timeout=payload.timeout_seconds) return_code = process.returncode elapsed = time.time() - start_time if return_code == 0: return TaskResult( task_id=payload.task_id, status="SUCCESS", message="Task completed normally.", elapsed_time=elapsed ) else: error_msg = stderr_data.decode("utf-8", errors="ignore").strip() return TaskResult( task_id=payload.task_id, status="FAILED