Last active
April 6, 2022 09:57
-
-
Save overengineer/078853fb84773b8e23887fbb1c2d43be to your computer and use it in GitHub Desktop.
Python Fabric Connection Class for Windows
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from fabric import Connection | |
from scp import SCPClient | |
import paramiko | |
# Wraps commands in bash.exe calls | |
# Also fixes scp problem | |
class WindowsConnection(Connection): | |
_default_bash_path = "C:/Program Files/Git/bin/bash.exe" | |
def resolve_connect_kwargs(self, connect_kwargs): | |
self._bash_path = connect_kwargs.pop( | |
"bash_path", WindowsConnection._default_bash_path | |
) | |
return super(WindowsConnection, self).resolve_connect_kwargs(connect_kwargs) | |
def _run(self, runner, cmd, **kwargs): | |
logger.debug(f"$ {cmd}") | |
cmd = self._prefix_commands(cmd) | |
cmd = cmd.replace("\\", "\\\\") | |
cmd = cmd.replace(r'"', r"\"") | |
bash_unofficial_strict_mode = ( | |
r"set -euo pipefail && IFS=$'\n\t' && LANG='' && MSYS_NO_PATHCONV=1" | |
) | |
inline_cmd = f'"{self._bash_path}" -c "{bash_unofficial_strict_mode} && {cmd}"' | |
return runner.run(inline_cmd, **kwargs) | |
def put(self, src, dst, recursive=True): | |
self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) | |
scp = SCPClient(self.client.get_transport(), progress=progress) | |
scp.put(src, dst, recursive=recursive) | |
scp.close() | |
print() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment