\n\n\n\n
- Directory Structure<\/a><\/li>
- main.py<\/a><\/li>
- Configuration File<\/a><\/li>
- sftp_config.yaml<\/a><\/li>
- config<\/a><\/li>
- File: __init__.py<\/a><\/li>
- File: file_upload.py<\/a><\/li>
- __set_file_paths<\/a><\/li>
- __connect_client<\/a><\/li>
- known_hosts file<\/a><\/li>
- send_files<\/a><\/li><\/ol><\/div>\n\n\n\n
\n\n\n\nDirectory Structure<\/h2>\n\n\n\n
The directory structure will look like the following:<\/p>\n\n\n\n
![\"\"](\"https:\/\/eipsoftware.com\/musings\/wp-content\/uploads\/2024\/10\/sftp_file_structure-1.png\")
<\/figure>\n\n\n\n
\n\n\n\nmain.py<\/h2>\n\n\n\n
This is the file we will invoke to start the program. It is rather simple.<\/p>\n\n\n\n
# -- LIBRARY -----------------------------------------------------------------\nimport sys\n\n# -- LOCAL FILES -------------------------------------------------------------\nfrom src.file_upload import FileUpload\n\ndef main(argv) -> None:\n \n file_list = ['file_one.txt', 'file_two.txt']\n \n # upload files to sftp\n if len(file_list) > 0:\n fp = FileUpload()\n fp.send_files(file_list=file_list)\n\n\n# Start the program\nif __name__ == '__main__':\n main(sys.argv[1:])<\/pre><\/div>\n\n\n\n
<\/p>\n\n\n\n
# -- LIBRARY -----------------------------------------------------------------\nimport sys\n\n# -- LOCAL FILES -------------------------------------------------------------\nfrom src.file_upload import FileUpload\n\ndef main(argv) -> None:\n \n file_list = ['file_one.txt', 'file_two.txt']\n \n # upload files to sftp\n if len(file_list) > 0:\n fp = FileUpload()\n fp.send_files(file_list=file_list)\n\n\n# Start the program\nif __name__ == '__main__':\n main(sys.argv[1:])<\/pre><\/div>\n\n\n\n
The file is using the sys.argv[1:] to allow you to pass in arguments when you are running the program. In an effort to keep focus, we won’t being using arguments, and will just hard code the file names in the main function.<\/p>\n\n\n\n
Now let’s go line by line.<\/p>\n\n\n\n
# -- LIBRARY -----------------------------------------------------------------\nimport sys\n\n# -- LOCAL FILES -------------------------------------------------------------\nfrom src.file_upload import FileUpload\n\ndef main(argv) -> None:\n \n file_list = ['file_one.txt', 'file_two.txt']\n \n # upload files to sftp\n if len(file_list) > 0:\n fp = FileUpload()\n fp.send_files(file_list=file_list)\n\n\n# Start the program\nif __name__ == '__main__':\n main(sys.argv[1:])<\/pre><\/div>\n\n\n\n
This is the FileUpload class we will create later.<\/p>\n\n\n\n
# -- LIBRARY -----------------------------------------------------------------\nimport sys\n\n# -- LOCAL FILES -------------------------------------------------------------\nfrom src.file_upload import FileUpload\n\ndef main(argv) -> None:\n \n file_list = ['file_one.txt', 'file_two.txt']\n \n # upload files to sftp\n if len(file_list) > 0:\n fp = FileUpload()\n fp.send_files(file_list=file_list)\n\n\n# Start the program\nif __name__ == '__main__':\n main(sys.argv[1:])<\/pre><\/div>\n\n\n\n
The main function, pretty simple isn’t it?<\/p>\n\n\n\n
file_list is a list of files we will upload to the sFTP server. Conversely you could have a list of files you wish to download.<\/p>\n\n\n\n
Next we check if the file_list has any values in it; in case the list is being passed in via sys.argv. <\/p>\n\n\n\n
Next we assign fp to the FileUpload class<\/p>\n\n\n\n
Last step is we send the files, by calling the send_files method in the class.<\/p>\n\n\n\n
<\/p>\n\n\n\n
Configuration File<\/h2>\n\n\n\n
In the configuration directory we have three files. <\/p>\n\n\n\n
- \n
- __init__.py<\/li>\n\n\n\n
- known_hosts<\/a><\/li>\n\n\n\n
- sftp_config.yaml<\/li>\n<\/ol>\n\n\n\n
We are going to look at the sftp_config.yaml configuration file and the __init__.py file first. We will come back to the known_hosts<\/a> file later on.<\/p>\n\n\n\n
\n\n\n\nsftp_config.yaml<\/h2>\n\n\n\n
The configuration file will store the details on how to connect to the sFTP server and what files we want to upload.<\/p>\n\n\n\n
# ----------------------------------------------------------------------------\n# SFTP CONFIGURATION OPTIONS\n# ----------------------------------------------------------------------------\n# sftp connection values\n---\nSERVER:\n HOSTNAME: 'THESFTP-SERVER'\n USERNAME: 'myusername'\n KEY_FILENAME: 'C:\\\\Users\\\\my_name\\\\.ssh\\\\thekey.pem'\n KNOWN_HOSTS: '\/config\/known_hosts'\n\nFILE:\n LOCAL_DIRECTORY: 'main\/assets\/'\n REMOTE_DIRECTORY: '\/home\/username\/directory\/directory\/main\/assets\/'\n\nFILE_NAMES:\n ONE: 'image_one.png'\n TWO: 'image_two.png'\n THREE: 'image_three.png'\n\nWAIT_PERIOD: &waitperiod \"10\"<\/pre><\/div>\n\n\n\n
The configuration file is in plain text, the values are pretty self-explanatory. We will talk about the KEY_FILENAME and KNOWN_HOSTS<\/a> later on.<\/p>\n\n\n\n
Make sure to specify your details, including HOSTNAME, USERNAME, LOCAL_DIRECTORY, REMOTE_DIRECTORY, etc…<\/p>\n\n\n\n
\n\n\n\nconfig<\/h2>\n\n\n\n
The init file will be used to load the .yaml configuration file. It is handy way of being able to store the sFTP configuration parameters without having to change the main parts of the code. <\/p>\n\n\n\n
import pathlib, sys\nfrom box import Box<\/pre><\/div>\n\n\n\n
We need three libraries the standard pathlib and sys, Additionally I use python-box<\/a> to help referencing values in the configuration. Python-box has a dependency on yaml parser. I use ruamel, finding it easier and quicker.<\/p>\n\n\n\n
If not already installed, from your command line you can use pip to install them.<\/p>\n\n\n\n
pip install python-box\npip install ruamel.yaml<\/code><\/pre>\n\n\n\n
\n\n\n\nFile: __init__.py<\/h2>\n\n\n\n
# ----------------------------------------------------------------------------\n# load configuration\n# ----------------------------------------------------------------------------\ndef load_config(file_name: str) -> Box:\n path = pathlib.Path(__file__).parent \/ file_name\n try: \n with path.open(mode=\"r\") as config_file:\n return Box(frozen_box=True).from_yaml(config_file.read())\n\n except FileNotFoundError as e:\n print(f\"The directory {path.name} does not exist \\n\\n {e}\")\n sys.exit(1)\n except PermissionError as e:\n print(f\"Permission denied to access the directory {path.name}\\n\\n {e}\")\n sys.exit(1)\n except OSError as e:\n print(f\"An OS error occurred: {e}\")\n sys.exit(1)\n\n finally:\n path = None<\/pre><\/div>\n\n\n\nThe function load_config will take a file name as an argument and return the python-box object. <\/p>\n\n\n\n
Next we get the path of the file and store it in the path variable.<\/p>\n\n\n\n
try: \n with path.open(mode=\"r\") as config_file:\n return Box(frozen_box=True).from_yaml(config_file.read())<\/pre><\/div>\n\n\n\n
Now we will open the file, and convert the .yaml file into a Box object to be used by the FileUpload class later on.<\/p>\n\n\n\n
sftp_config:Box = load_config(\"sftp_config.yaml\")<\/pre><\/div>\n\n\n\n
And last step, we call the load_config function.<\/p>\n\n\n\n
\n\n\n\nFile: file_upload.py<\/h2>\n\n\n\n
The FileUpload class will be used to connect to the sFTP server and than upload files. Additional methods could be added to download files, I will leave that exercise to you.<\/p>\n\n\n\n
I am going to use the paramiko <\/a>package. Again if not installed, from the command line use pip to install.<\/p>\n\n\n\n
pip install paramiko\n<\/code><\/pre>\n\n\n\n<\/p>\n\n\n\n
# -- LIBRARY -----------------------------------------------------------------\nimport os\nfrom pathlib import Path\n\nimport paramiko as pa\n\n# -- LOCAL FILES -------------------------------------------------------------\nfrom config import sftp_config\n\nclass FileUpload():\n def __init__(self) -> None:\n self._file_list: list = None\n self._ssh_client: pa.SSHClient = None\n self._sftp_client: pa.SFTPClient = None\n self._known_hosts_file_path: Path = None\n\n # set the file paths\n self.__set_file_paths()\n\n def __set_file_paths(self) -> None:\n self._known_hosts_file_path = os.path.normpath(\"\".join([os.getcwd(),sftp_config.SERVER.KNOWN_HOSTS]))\n\n if os.path.isfile(self._known_hosts_file_path) == False:\n raise Exception(f\"Known_Hosts file can't be found: {self._known_hosts_file_path}\")\n\n\n def __connect_client(self) -> None:\n \n # get the ssh key\n sftp_key = pa.RSAKey.from_private_key_file(sftp_config.SERVER.KEY_FILENAME)\n\n # set the ssh client\n self._ssh_client = pa.SSHClient()\n\n self._ssh_client.load_host_keys(self._known_hosts_file_path)\n\n # connect to server\n self._ssh_client.connect(hostname=sftp_config.SERVER.HOSTNAME,\n username=sftp_config.SERVER.USERNAME,\n pkey=sftp_key)\n\n\n def send_files(self, file_list:list) -> None:\n\n # connect to ssh client\n self.__connect_client()\n \n # init the sftp client\n self._sftp_client = self._ssh_client.open_sftp()\n\n self._sftp_client.chdir(sftp_config.FILE.REMOTE_DIRECTORY)\n \n # cycle through each file and put them on server\n for file in file_list:\n local_file_path = os.path.join(file)\n remote_file_path = \"\".join([sftp_config.FILE.REMOTE_DIRECTORY, os.path.basename(file)])\n try:\n self._sftp_client.put(localpath=local_file_path,\n remotepath=remote_file_path)\n \n except FileNotFoundError as err:\n print(f\"File {local_file_path} not found locally\")\n \n self._sftp_client.close()\n self._ssh_client.close()\n<\/pre><\/div>\n\n\n\n<\/p>\n\n\n\n
# -- LIBRARY -----------------------------------------------------------------\nimport os\nfrom pathlib import Path\n\nimport paramiko as pa\n\n# -- LOCAL FILES -------------------------------------------------------------\nfrom config import sftp_config<\/pre><\/div>\n\n\n\n
Here we will import in standard libraries and the sftp_config <\/a>file we created earlier. <\/p>\n\n\n\n
class FileUpload():\n def __init__(self) -> None:\n self._file_list: list = None\n self._ssh_client: pa.SSHClient = None\n self._sftp_client: pa.SFTPClient = None\n self._known_hosts_file_path: Path = None\n\n # set the file paths\n self.__set_file_paths()<\/pre><\/div>\n\n\n\n
Here we are defining the class, and setting some variables to hold, the ssh client, sftp client, and known_hosts file_path.<\/p>\n\n\n\n
<\/p>\n\n\n\n
__set_file_paths<\/h2>\n\n\n\n
def __set_file_paths(self) -> None:\n self._known_hosts_file_path = os.path.normpath(\"\".join([os.getcwd(),sftp_config.SERVER.KNOWN_HOSTS]))\n\n if os.path.isfile(self._known_hosts_file_path) == False:\n raise Exception(f\"Known_Hosts file can't be found: {self._known_hosts_file_path}\")<\/pre><\/div>\n\n\n\n - sftp_config.yaml<\/li>\n<\/ol>\n\n\n\n
- main.py<\/a><\/li>