我想用Python在远程服务器上上传文件。我想事先检查远程路径是否确实存在,如果不存在,则创建它。伪代码:
if(remote_path not exist):
create_path(remote_path)
upload_file(local_file, remote_path)
我正在考虑在 Paramiko 中执行命令来创建路径(例如
mkdir -p remote_path
)。我想出了这个:
# I didn't test this code
import paramiko, sys
ssh = paramiko.SSHClient()
ssh.connect(myhost, 22, myusername, mypassword)
ssh.exec_command('mkdir -p ' + remote_path)
ssh.close
transport = paramiko.Transport((myhost, 22))
transport.connect(username = myusername, password = mypassword)
sftp = paramiko.SFTPClient.from_transport(transport)
sftp.put(local_path, remote_path)
sftp.close()
transport.close()
但是这个解决方案对我来说听起来不太好,因为我关闭了连接,然后再次重新打开它。有更好的方法吗?
SFTP 支持常用的 FTP 命令(chdir、mkdir 等...),因此请使用这些命令:
sftp = paramiko.SFTPClient.from_transport(transport)
try:
sftp.chdir(remote_path) # Test if remote_path exists
except IOError:
sftp.mkdir(remote_path) # Create remote_path
sftp.chdir(remote_path)
sftp.put(local_path, '.') # At this point, you are in remote_path in either case
sftp.close()
要完全模拟
mkdir -p
,您可以递归地处理remote_path:
import os.path
def mkdir_p(sftp, remote_directory):
"""Change to this directory, recursively making new folders if needed.
Returns True if any folders were created."""
if remote_directory == '/':
# absolute path so change directory to root
sftp.chdir('/')
return
if remote_directory == '':
# top-level relative directory must exist
return
try:
sftp.chdir(remote_directory) # sub-directory exists
except IOError:
dirname, basename = os.path.split(remote_directory.rstrip('/'))
mkdir_p(sftp, dirname) # make parent directories
sftp.mkdir(basename) # sub-directory missing, so created it
sftp.chdir(basename)
return True
sftp = paramiko.SFTPClient.from_transport(transport)
mkdir_p(sftp, remote_path)
sftp.put(local_path, '.') # At this point, you are in remote_path
sftp.close()
当然,如果remote_path还包含远程文件名,那么需要将其拆分,将目录传递给mkdir_p,并使用文件名代替“.”在 sftp.put 中。
一些更简单且更具可读性的东西
def mkdir_p(sftp, remote, is_dir=False):
"""
emulates mkdir_p if required.
sftp - is a valid sftp object
remote - remote path to create.
"""
dirs_ = []
if is_dir:
dir_ = remote
else:
dir_, basename = os.path.split(remote)
while len(dir_) > 1:
dirs_.append(dir_)
dir_, _ = os.path.split(dir_)
if len(dir_) == 1 and not dir_.startswith("/"):
dirs_.append(dir_) # For a remote path like y/x.txt
while len(dirs_):
dir_ = dirs_.pop()
try:
sftp.stat(dir_)
except:
print "making ... dir", dir_
sftp.mkdir(dir_)
今天必须这样做。这是我的做法。
def mkdir_p(sftp, remote_directory):
dir_path = str()
for dir_folder in remote_directory.split("/"):
if dir_folder == "":
continue
dir_path += r"/{0}".format(dir_folder)
try:
sftp.listdir(dir_path)
except IOError:
sftp.mkdir(dir_path)
您可以使用 pysftp 包:
import pysftp as sftp
#used to pypass key login
cnopts = sftp.CnOpts()
cnopts.hostkeys = None
srv = sftp.Connection(host="10.2.2.2",username="ritesh",password="ritesh",cnopts=cnopts)
srv.makedirs("a3/a2/a1", mode=777) # will happily make all non-existing directories
您可以查看此链接了解更多详情: https://pysftp.readthedocs.io/en/release_0.2.9/cookbook.html#pysftp-connection-mkdir
我的版本:
def is_sftp_dir_exists(sftp, path):
try:
sftp.stat(path)
return True
except Exception:
return False
def create_sftp_dir(sftp, path):
try:
sftp.mkdir(path)
except IOError as exc:
if not is_sftp_dir_exists(sftp, path):
raise exc
def create_sftp_dir_recursive(sftp, path):
parts = deque(Path(path).parts)
to_create = Path()
while parts:
to_create /= parts.popleft()
create_sftp_dir(sftp, str(to_create))
由于
EAFP 原则,我们首先尝试
mkdir
,而不尝试 listdir
/stat
(发出一个网络请求比发出多个网络请求性能更高)。
假设 sftp 操作成本高昂, 我会选择:
def sftp_mkdir_p(sftp, remote_directory):
dirs_exist = remote_directory.split('/')
dirs_make = []
# find level where dir doesn't exist
while len(dirs_exist) > 0:
try:
sftp.listdir('/'.join(dirs_exist))
break
except IOError:
value = dirs_exist.pop()
if value == '':
continue
dirs_make.append(value)
else:
return False
# ...and create dirs starting from that level
for mdir in dirs_make[::-1]:
dirs_exist.append(mdir)
sftp.mkdir('/'.join(dirs_exist))```
如果您想在没有异常流的情况下执行此操作,则仅针对异常流保留异常。您可以执行以下操作:
def ensure_sftp_directories_exist(sftp: SFTPClient, remote_dir: str) -> None:
cleaned_remote_dir = remote_dir.strip('/') # Remove leading and trailing slashes
sub_directories = cleaned_remote_dir.split("/") # Get all directories
current_directory = "/" # Root is start
for directory in sub_directories:
# Skip empty directory names for accidental double '/'
if not directory:
continue
current_directory += f"{directory}/"
if directory not in sftp.listdir(current_directory):
sftp.mkdir(current_directory)
备注: