Properly renamed single_shot_entry to single_entry

sia/single_entry.py

@@ -0,0 +1,121 @@
+from datetime import datetime
+import subprocess
+import xml.etree.ElementTree as ET
+from typing import Optional
+
+from .entry import Entry
+from .util import escape_text_for_xml
+
+class SingleEntry(Entry):
+    """
+    Entry type for one-time script executions.
+    
+    Attributes:
+        script: The script/command to execute
+        stdout: Captured standard output from script execution
+        stderr: Captured standard error from script execution 
+        exit_code: Process exit code after completion
+        _executed: Whether the script has been executed
+    """
+    
+    def __init__(self, script: str, id: str, timestamp: datetime):
+        """
+        Initialize a new single shot entry.
+        
+        Args:
+            script: The script/command to execute
+            id: Unique identifier for this entry
+            timestamp: Creation timestamp for this entry
+        """
+        super().__init__(id, timestamp)
+        self._script = script
+        self._stdout = ""
+        self._stderr = ""
+        self._exit_code: Optional[int] = None
+        self._executed = False
+
+    @property
+    def script(self) -> str:
+        """Get the script/command being executed."""
+        return self._script
+
+    @property
+    def stdout(self) -> str:
+        """Get the captured standard output."""
+        return self._stdout
+
+    @property
+    def stderr(self) -> str:
+        """Get the captured standard error."""
+        return self._stderr
+
+    @property
+    def exit_code(self) -> Optional[int]:
+        """Get the exit code of the process (None if still running)."""
+        return self._exit_code
+        
+    def update(self) -> None:
+        """
+        Execute the script if not already executed.
+        Captures stdout, stderr and exit code.
+        """
+        if self._executed:
+            return
+            
+        try:
+            # Run script and capture output
+            process = subprocess.run(
+                self.script,
+                shell=True,
+                capture_output=True,
+                text=True
+            )
+            
+            # Store results
+            self._stdout = process.stdout
+            self._stderr = process.stderr
+            self._exit_code = process.returncode
+            
+        except subprocess.SubprocessError as e:
+            # Handle subprocess errors by capturing the error message
+            self._stdout = ""
+            self._stderr = str(e)
+            self._exit_code = -1
+            
+        except Exception as e:
+            # Handle any other errors
+            self._stdout = ""
+            self._stderr = f"Error executing script: {str(e)}"
+            self._exit_code = -1
+            
+        self._executed = True
+
+    def generate_context(self) -> ET.Element:
+        """
+        Generate an XML Element representing this single shot entry.
+        
+        Returns:
+            ET.Element: XML element containing the entry's data
+        """
+        # Create root element
+        element = ET.Element("single", {
+            "id": self.id,
+        })
+        
+        # Add script as CDATA or escaped text
+        element.text = escape_text_for_xml(self.script)
+        
+        # Only add output elements if script has executed
+        if self._executed:
+            # Add stdout element if there is output
+            stdout_elem = ET.SubElement(element, "stdout")
+            stdout_elem.text = escape_text_for_xml(self._stdout)
+            
+            # Add stderr element if there are errors
+            stderr_elem = ET.SubElement(element, "stderr")
+            stderr_elem.text = escape_text_for_xml(self._stderr)
+            
+            # Add exit code attribute after execution
+            element.set("exit_code", str(self._exit_code))
+            
+        return element