Python functions to fetch spreadsheets through SFTP and from Dropbox [closed]

Question

I have written two functions- first one is to read files from SFTP and second one is to read files from Dropbox.

Both functions have some similar line of code like validating extension and reading/saving file that can be extracted into a separated function and thus need your suggestion how to do it.

PFB my both functions-

class SftpHelper:
    def fetch_file_from_sftp(self, file_name, sheet_name=0):
      valid_extensions = ['csv', 'xls', 'xlsx']
      extension = file_name.split('.')[-1]
      sftp, transport = self.connect_to_sftp()
      remote_path = self.remote_dir + file_name
      data = io.BytesIO()
      sftp.getfo(remote_path, data, callback=None)
      if extension == 'csv':
          file_df = pd.read_csv(io.BytesIO(data.getvalue()))
      else:
          file_df = pd.read_excel(io.BytesIO(data.getvalue()), sheet_name=sheet_name)
      self.close_sftp_connection(sftp, transport)
      return file_df

class DropBoxHelper:
    def read_file_from_dropbox(self, file_name, sheet_name=0):
      valid_extensions = ['csv', 'xls', 'xlsx']
      extension = file_name.split('.')[-1]
      dbx = self.connect_to_dropbox()
      metadata,data=dbx.files_download(file_name)
      if extension == 'csv':
          file_df = pd.read_csv(io.BytesIO(data.content))
      else:
          file_df = pd.read_excel((io.BytesIO(data.content)), sheet_name=sheet_name)
      return file_df

Can anyone please help me to extract the common logix to a seperate function and then use that one in my two functions?

Answer 1

Since you are already using classes here, you could derive from a base class that have the shared behaviour and delegate to derived classes the specific behaviour of the connection:

from os.path import splitext


class _RemoteHelper
    def file_reader(self, file_name, sheet_name=0):
        _, extension = splitext(file_name)
        data = self._internal_file_reader(file_name)

        if extension == 'csv':
            return pd.read_csv(io.BytesIO(data))
        else:
            return pd.read_excel((io.BytesIO(data)), sheet_name=sheet_name)


class SftpHelper(_RemoteHelper):
    def _internal_file_reader(self, file_name):
        data = io.BytesIO()
        sftp, transport = self.connect_to_sftp()
        sftp.getfo(self.remote_dir + file_name, data, callback=None)
        self.close_sftp_connection(sftp, transport)
        return data.getvalue()


class DropBoxHelper(_RemoteHelper):
    def _internal_file_reader(self, file_name):
        dbx = self.connect_to_dropbox()
        _, data = dbx.files_download(file_name)
        return data.content

This have the neat advantage of harmonizing the interfaces accros both classes.

Answer 2

From looking at the code, some things seem off:

1. valid_extensions is defined, but not used
2. connect_to_sftp(), self.remote_dir, io.BytesIO(), sftp.getfo(), pd, self.close_sftp_connection() and a bunch of other functions/fields are not defined

That being said, the core problem is addressed by creating a parent class which both your classes can inherit from. It'd look something like this:

class FileHelper:
    def parse_fetched_file(self, file_name, data, sheet_name):
        valid_extensions = ['csv', 'xls', 'xlsx']
        extension = file_name.split('.')[-1]
        if extension == 'csv':
            return pd.read_csv(io.BytesIO(data.content))
        return pd.read_excel((io.BytesIO(data.content)), sheet_name=sheet_name)


class SftpHelper(FileHelper):
    def fetch_file_from_sftp(self, file_name, sheet_name = 0):
        sftp, transport = self.connect_to_sftp()
        remote_path = self.remote_dir + file_name
        data = io.BytesIO()
        sftp.getfo(remote_path, data, callback=None)
        file_df super(SftpHelper, self).parse_fetched_file(file_name, data, sheet_name)
        self.close_sftp_connection(sftp, transport)
        return file_df

class DropBoxHelper(FileHelper):
    def read_file_from_dropbox(self, file_name, sheet_name = 0):
        dbx = self.connect_to_dropbox()
        metadata, data = dbx.files_download(file_name)
        return super(DropBoxHelper, self).parse_fetched_file(file_name, data, sheet_name)

I'm not 100% sure that it's the most efficient syntax, but it gets the job done.