1. Llamada a comandos de la shell SINCRONA
Veamos como se llama a un proceso síncronamente ya sea por el ecomando entero o pasándole parámetros:
def run_text(cmd): ''' Executes a command and captures its output as text. Parameters: cmd: A list of command arguments to execute (e.g., ["certutil", "-L", "-d", "sql:/home/user/.pki/nssdb"]). Returns: The standard output of the command as a string if successful, or None if an error occurs. Example usage: # List certificates in the NSS database output = run_text(["certutil", "-L", "-d", "sql:/home/user/.pki/nssdb"]) ''' try: res = subprocess.run( cmd, capture_output=True, text=True, check=True ) return res.stdout except subprocess.CalledProcessError as e: print(f"Error ejecutando: {' '.join(cmd)}", file=sys.stderr) return None def run_bytes(cmd, input_data=None): ''' Executes a command and captures its output as bytes. Parameters: cmd: A list of command arguments to execute (e.g., ["openssl", "x509", "-in", "cert.pem", "-outform", "DER"]). input_data: Optional bytes to be sent to the command's standard input. Returns: The standard output of the command as bytes if successful, or None if an error occurs. Example usage: # Get the PEM-encoded certificate for a given nickname from the NSS database pem_bytes = run_bytes(["certutil", "-L", "-d", "sql:/home/user/.pki/nssdb", "-n", "My Certificate", "-a"]) ''' try: res = subprocess.run( cmd, input=input_data, capture_output=True, check=True ) return res.stdout except subprocess.CalledProcessError: return None
2. Llamada a comandos de la shell ASINCRONA
Veamos como se llama a un proceso asíncronamente ya sea por el ecomando entero o pasándole parámetros:
async def run_text_async(cmd: list[str]) -> str | None: ''' Executes a command asynchronously and captures its output as text. Parameters: cmd: A list of command arguments to execute (e.g., ["ls", "-l"]). Returns: The standard output of the command as a string if successful, or None if an error occurs. Example usage: # List files in the current directory asynchronously output = await run_text_async(["ls", "-l"]) ''' try: proc = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) stdout, stderr = await proc.communicate() if proc.returncode != 0: print(f"Error ejecutando: {' '.join(cmd)}", file=sys.stderr) if stderr: print(stderr.decode(errors="ignore").strip(), file=sys.stderr) return None return stdout.decode(errors="ignore") except Exception as e: print(f"Excepción ejecutando {' '.join(cmd)}: {e}", file=sys.stderr) return None async def run_bytes_async(cmd: list[str], input_data: bytes | None = None) -> bytes | None: ''' Executes a command asynchronously and captures its output as bytes. Parameters: cmd: A list of command arguments to execute (e.g., ["openssl", "x509", "-in", "cert.pem", "-outform", "DER"]). input_data: Optional bytes to be sent to the command's standard input. Returns: The standard output of the command as bytes if successful, or None if an error occurs. Example usage: # Get the PEM-encoded certificate for a given nickname from the NSS database asynchronously pem_bytes = await run_bytes_async(["certutil", "-L", "-d", " ''' try: proc = await asyncio.create_subprocess_exec( *cmd, stdin=asyncio.subprocess.PIPE if input_data is not None else None, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) stdout, stderr = await proc.communicate(input=input_data) if proc.returncode != 0: print(f"Error ejecutando: {' '.join(cmd)}", file=sys.stderr) if stderr: print(stderr.decode(errors="ignore").strip(), file=sys.stderr) return None return stdout except Exception as e: print(f"Excepción ejecutando {' '.join(cmd)}: {e}", file=sys.stderr) return None
