From b41e270b8ec36ed10e564ef1fe867a0c6a27016b Mon Sep 17 00:00:00 2001 From: elazar Date: Sun, 2 Nov 2025 16:03:33 -0600 Subject: [PATCH] docs(connectors): add command wrapping best practices section Add comprehensive guidance for custom connector implementations on: - Using make_unix_command_for_host() for proper shell wrapping - Filtering connector control parameters with extract_control_arguments() - Why command wrapping matters for shell operators - Example implementation following PyInfra best practices - References to built-in connector implementations This addresses a documentation gap where connector developers may not be aware of the need to filter control parameters (_success_exit_codes, _timeout, _get_pty, _stdin) before calling make_unix_command_for_host(). Without proper filtering, connectors will encounter TypeError when these parameters are passed to make_unix_command(), which does not accept them. The new section recommends using the existing extract_control_arguments() utility from pyinfra.connectors.util, which is already used by built-in connectors like docker, dockerssh, and chroot. --- docs/api/connectors.md | 56 +++++++++++++++++++++++++++++++++++++++++- 1 file changed, 55 insertions(+), 1 deletion(-) diff --git a/docs/api/connectors.md b/docs/api/connectors.md index c32c9f69a..e2cd312cd 100644 --- a/docs/api/connectors.md +++ b/docs/api/connectors.md @@ -56,7 +56,7 @@ class MyConnector(BaseConnector): @staticmethod def make_names_data(_=None): ... # see above - + def run_shell_command( self, command: StringCommand, @@ -191,6 +191,60 @@ screen. `disconnect` can be used after all operations complete to clean up any connection/s remaining to the hosts being managed. +## Implementing `run_shell_command` + +When implementing `run_shell_command`, connectors should use pyinfra's command wrapping utilities rather than manually constructing commands. The `make_unix_command_for_host()` function from `pyinfra.connectors.util` handles shell wrapping, sudo elevation, environment variables, working directory changes, command retries and shell executable selection. + +Its worth being aware that when passing `arguments` to `make_unix_command_for_host()`, connector control parameters must be filtered out. These parameters (`_success_exit_codes`, `_timeout`, `_get_pty`, `_stdin`) are defined in `pyinfra.api.arguments.ConnectorArguments` and are meant for the connector's internal logic after command generation, not for command construction itself. + +The recommended approach is to use `extract_control_arguments()` from `pyinfra.connectors.util` which handles this filtering for you: + +```py +from pyinfra.connectors.util import extract_control_arguments, make_unix_command_for_host + +class MyConnector(BaseConnector): + handles_execution = True + + def run_shell_command( + self, + command: StringCommand, + print_output: bool = False, + print_input: bool = False, + **arguments: Unpack["ConnectorArguments"], + ) -> Tuple[bool, CommandOutput]: + """Execute a command with proper shell wrapping.""" + + # Extract and remove control parameters from arguments + # This modifies arguments dict in place and returns the extracted params + control_args = extract_control_arguments(arguments) + + # Generate properly wrapped command with sudo, environment, etc + # arguments now contains only command generation parameters + wrapped_command = make_unix_command_for_host( + self.state, + self.host, + command, + **arguments, + ) + + # Use control parameters for execution + timeout = control_args.get("_timeout") + success_exit_codes = control_args.get("_success_exit_codes", [0]) + + # Execute the wrapped command using your connector's method + exit_code, output = self._execute(wrapped_command, timeout=timeout) + + # Check success based on exit codes + success = exit_code in success_exit_codes + + return success, output +``` + +Without proper command wrapping, shell operators and complex commands will fail. For example `timeout 60 bash -c 'command' || true` executed without shell wrapping will result in `bash: ||: command not found`. PyInfra operations and fact gathering rely on shell operators (`&&`, `||`, pipes, redirects) so using `make_unix_command_for_host()` ensures your connector handles these correctly. + +For complete examples see pyinfra's built-in connectors in `pyinfra/connectors/docker.py`, `pyinfra/connectors/chroot.py`, `pyinfra/connectors/ssh.py` and `pyinfra/connectors/local.py`, as well as the command wrapping utilities in `pyinfra/connectors/util.py`. + + ## pyproject.toml In order for pyinfra to gain knowledge about your connector, you need to add the following snippet to your connector's `pyproject.toml`: