SFTP Download file from remote sftp server to local

Viewed 31

I got a code to download files from sever to local using python , I am getting the following error when I run the code First, please set the environment variable SFTPTOGO_URL and try again.

import pysftp from urllib.parse import urlparse import os, zipfile from datetime import date

class Sftp: def init(self, hostname, username, password, port=22): """Constructor Method""" self.connection = None self.hostname = '10.61.1.23' self.username = 'lynxx' self.password = 'lynxx' 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(r"sftp://lynxx:lynxx@10.61.1.23/")

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 = "D:\Lynxx\aco_mor_308_cvpsa_001"
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 = r'C:\Users\Khanyizo Yezo\Pictures'
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')
)

def zip_folder(filename): zip_file = os.path.join(os.path.dirname(filename),os.path.split(filename)[1]+".zip") zipf = zipfile.ZipFile(zip_file, 'w', zipfile.ZIP_DEFLATED) for root, dirs, files in os.walk(filename): for file in files: zipf.write(os.path.join(root, file), os.path.relpath(os.path.join(root, file), os.path.join(filename, '..'))) zipf.close()

# Disconnect from SFTP

sftp.disconnect()

0 Answers
Related