Python을 이용하여 SFPT에 연결, 파일을 나열, 업로드 및 다운로드하는 방법에 대해 설명합니다.
안전한 파일 또는 데이터의 전송을 목적으로 하는 SFTP는 많은 유저들로부터 가장 많이 애용되어 왔습니다. SFTP는 효율적이고 사용자 친화적인 프로토콜이지만, Python과 같은 프로그래밍 언어로 SFTP 서버에 연결하려면 조금 더 깊이 파고들어야 합니다. 아래의 기사를 참고하여, Python에서 SFTP를 이용해 접속하는 방법을 알아보세요.
요구사항
이 작업을 수행하기 전에 몇 가지 설정이 필요합니다. 여기서는 'pysftp'패키지를 사용하여 SFTP 서버에 연결하고 명령을 전달합니다. 설치 준비가 완료되면 다음명령을 수동으로 실행하십시오:
pip install pysftp또는 requirements.txt의 파일을 작성하고 그 안에서 종속성을 선언하십시오. 그런 다음 아래 내용을 저장합니다:
pysftp>=0.2.9다음과 같이 실행합니다:
pip install -r requirements.txtSFTP에 연결
이 부분에서는 sftp://user:password@host 라는 URI형식으로 SFTP 서버에 연결하는데 필요하는데 필요한 모든 정보가 들어있는 SFTPTOGO_URL이라는 환경 변수를 사용합니다. 이 변수는 `urlparse`를 사용해 URI 의 부분을 추출하며, 원격 서버의 호스트열쇠는 ~/.ssh/known_hosts를 사용해 디폴트로 검증됩니다.
연결이 설정되면 SFTP 클라이언트 개체가 변수 connection에 할당됩니다.
import pysftp
from urllib.parse import urlparse
import os
class Sftp:
def __init__(self, hostname, username, password, port=22):
"""Constructor Method"""
# Set connection object to None (initial value)
self.connection = None
self.hostname = hostname
self.username = username
self.password = password
self.port = port
def connect(self):
"""Connects to the sftp server and returns the sftp connection object"""
try:
# Get the sftp connection object
self.connection = pysftp.Connection(
host=self.hostname,
username=self.username,
password=self.password,
port=self.port,
)
except Exception as err:
raise Exception(err)
finally:
print(f"Connected to {self.hostname} as {self.username}.")
def disconnect(self):
"""Closes the sftp connection"""
self.connection.close()
print(f"Disconnected from host {self.hostname}")파일 목록
연결이 완료되었으므로 원격SFTP 서버의 파일 목록을 검색할 수 있습니다. 이것은 연결 객체의 listdir 함수 또는 listdir_attr 함수를 호출하는 방식으로 수행됩니다. 이러한 함수의 인수에 원격 패스 또는 인수의 지정 없음에 의해, 현재의 원격 패스의 파일과 디렉토리의 일람을 표시합니다. listdir 함수는 파일명의 리스트(문자열)를 돌려주어, listdir_attr 함수는 파일명을 포함하여, 파일 사이즈, 작성, 갱신의 타임 스탬프, 퍼미션을 포함한 SFTPAttributes 오브젝트의 리스트를 돌려줍니다. 이 예제에서는 래퍼 함수가 이터레이터를 반환하고 함수 호출자가 반환된 파일목록을 반복처리로 표시 합니다.
def listdir(self, remote_path):
"""lists all the files and directories in the specified path and returns them"""
for obj in self.connection.listdir(remote_path):
yield obj
def listdir_attr(self, remote_path):
"""lists all the files and directories (with their attributes) in the specified path and returns them"""
for attr in self.connection.listdir_attr(remote_path):
yield attr파일 업로드
다음 단계는 파일 업로드입니다. 연결 객체의 put 함수를 사용하여 로컬 파일에 대한 경로와 업로드가 끝날 때 파일이 위치해야 하는 원격 경로를 전달합니다. 함수 호출은 다음과 같습니다: connection.put("./local.txt", "./remote.txt")
def upload(self, source_local_path, remote_path):
"""
Uploads the source files from local to the sftp server.
"""
try:
print(
f"uploading to {self.hostname} as {self.username} [(remote path: {remote_path});(source local path: {source_local_path})]"
)
# Download file from SFTP
self.connection.put(source_local_path, remote_path)
print("upload completed")
except Exception as err:
raise Exception(err)파일 다운로드
마지막으로, 파일을 다운로드하는 방법입니다. 연결 객체의 get 함수를 사용하여 원격 파일에 대한 경로와 다운로드한 파일을 저장하는 로컬 경로를 전달합니다. 이 함수는 다음과 같이 호출합니다: connection.get("./remote.txt", "./download.txt")
def download(self, remote_path, target_local_path):
"""
Downloads the file from remote sftp server to local.
Also, by default extracts the file to the specified target_local_path
"""
try:
print(
f"downloading from {self.hostname} as {self.username} [(remote path : {remote_path});(local path: {target_local_path})]"
)
# Create the target directory if it does not exist
path, _ = os.path.split(target_local_path)
if not os.path.isdir(path):
try:
os.makedirs(path)
except Exception as err:
raise Exception(err)
# Download from remote sftp server to local
self.connection.get(remote_path, target_local_path)
print("download completed")
except Exception as err:
raise Exception(err)정리하면,
이상으로 모든 절차를 마쳤습니다! 만약, 이 프로그램을 처음부터 끝까지 실행해 보고 싶다면, 다음 코드를 복사하여 main.py로 저장하세요:
import pysftp
from urllib.parse import urlparse
import os
class Sftp:
def __init__(self, hostname, username, password, port=22):
"""Constructor Method"""
# Set connection object to None (initial value)
self.connection = None
self.hostname = hostname
self.username = username
self.password = password
self.port = port
def connect(self):
"""Connects to the sftp server and returns the sftp connection object"""
try:
# Get the sftp connection object
self.connection = pysftp.Connection(
host=self.hostname,
username=self.username,
password=self.password,
port=self.port,
)
except Exception as err:
raise Exception(err)
finally:
print(f"Connected to {self.hostname} as {self.username}.")
def disconnect(self):
"""Closes the sftp connection"""
self.connection.close()
print(f"Disconnected from host {self.hostname}")
def listdir(self, remote_path):
"""lists all the files and directories in the specified path and returns them"""
for obj in self.connection.listdir(remote_path):
yield obj
def listdir_attr(self, remote_path):
"""lists all the files and directories (with their attributes) in the specified path and returns them"""
for attr in self.connection.listdir_attr(remote_path):
yield attr
def download(self, remote_path, target_local_path):
"""
Downloads the file from remote sftp server to local.
Also, by default extracts the file to the specified target_local_path
"""
try:
print(
f"downloading from {self.hostname} as {self.username} [(remote path : {remote_path});(local path: {target_local_path})]"
)
# Create the target directory if it does not exist
path, _ = os.path.split(target_local_path)
if not os.path.isdir(path):
try:
os.makedirs(path)
except Exception as err:
raise Exception(err)
# Download from remote sftp server to local
self.connection.get(remote_path, target_local_path)
print("download completed")
except Exception as err:
raise Exception(err)
def upload(self, source_local_path, remote_path):
"""
Uploads the source files from local to the sftp server.
"""
try:
print(
f"uploading to {self.hostname} as {self.username} [(remote path: {remote_path});(source local path: {source_local_path})]"
)
# Download file from SFTP
self.connection.put(source_local_path, remote_path)
print("upload completed")
except Exception as err:
raise Exception(err)
if __name__ == "__main__":
sftp_url = os.environ.get("SFTPTOGO_URL")
if not sftp_url:
print("First, please set environment variable SFTPTOGO_URL and try again.")
exit(0)
parsed_url = urlparse(sftp_url)
sftp = Sftp(
hostname=parsed_url.hostname,
username=parsed_url.username,
password=parsed_url.password,
)
# Connect to SFTP
sftp.connect()
# Lists files with attributes of SFTP
path = "/"
print(f"List of files with attributes at location {path}:")
for file in sftp.listdir_attr(path):
print(file.filename, file.st_mode, file.st_size, file.st_atime, file.st_mtime)
# Upload files to SFTP location from local
local_path = "/Users/saggi/Downloads/tls2.png"
remote_path = "/tls2.png"
sftp.upload(local_path, remote_path)
# Lists files of SFTP location after upload
print(f"List of files at location {path}:")
print([f for f in sftp.listdir(path)])
# Download files from SFTP
sftp.download(
remote_path, os.path.join(remote_path, local_path + '.backup')
)
# Disconnect from SFTP
sftp.disconnect()마지막으로 다음 명령으로 실행합니다:
python main.py
Python을 사용하여 SFTP에 연결할 수 있었습니다. 축하합니다!
또한, Github에 있는 더 많은 코드 샘플들을 체크해 보세요.