Super simple client to get files from/write files to a NAS.

nas_client.py

# License: Apache-2.0
# Copyright: Florian Kromer 2020

from smb.SMBConnection import SMBConnection
from typing import List, Union
from smb.base import SharedDevice, SharedFile
from tempfile import NamedTemporaryFile


class NasServerClient:
    """
    Depends on pysmb v1.2.1: https://pysmb.readthedocs.io/en/latest/#
    """
    def __init__(self, user, password, client_machine_name, server_name, server_ip):
        self.user = user
        self.password = password
        self.client_machine_name = client_machine_name
        self.server_name = server_name
        self.server_ip = server_ip
        self.conn = self._create_connection(user, password, client_machine_name, server_name)

    def get_disk_tree_shares(self) -> Union[List[SharedDevice], SharedDevice, None]:
        shares = self.conn.listShares()
        disk_tree_shares = []
        for s in shares:
            if s.type == 0 and not s.isTemporary:
                disk_tree_shares.append(s)
        dts_count = len(disk_tree_shares)
        if dts_count == 0:
            return None
        elif dts_count == 1:
            return disk_tree_shares[0]
        else:
            return disk_tree_shares

    def get_shared_dirs(self, share_name: str) -> List[SharedFile]:
        shared_dirs = []
        try:
            shared_dirs_or_files = self.conn.listPath(share_name, '/')
        except Exception as e:
            raise Exception(f"Could not get shared dirs of the share. The initial exception is: {e}")
        for sdf in shared_dirs_or_files:
            if sdf.isDirectory:
                shared_dirs.append(sdf)
        return shared_dirs

    def get_shared_files(self, share_name: str, dir_name: str = '/') -> List[SharedFile]:
        shared_files = []
        try:
            shared_dirs_or_files = self.conn.listPath(share_name, dir_name)
        except Exception as e:
            raise Exception(f"Could not get shared files of the share. The initial exception is: {e}")
        for sdf in shared_dirs_or_files:
            if not sdf.isDirectory:
                shared_files.append(sdf)
        return shared_files

    def get_file(self, share_name: str, file_path: str) -> NamedTemporaryFile:
        file_obj = NamedTemporaryFile()
        try:
            file_attributes, filesize = self.conn.retrieveFile(share_name, file_path, file_obj)
        except Exception as e:
            raise Exception(f"Could not read file. The initial exception is:\n{type(e).__name__}: {e.args}")
        return file_obj

    def write_file(self, share_name: str, file_path: str, file: NamedTemporaryFile) -> None:
        with file as f:
            nsc.conn.storeFile(share_name, file_path, f)

    def close_connection(self):
        self.conn.close()

    def _create_connection(self, user, password, client_machine_name, server_name):
        conn = SMBConnection(user, password, client_machine_name, server_name, use_ntlm_v2 = True)
        try:
            conn.connect(self.server_ip, 139)
        except Exception as e:
            raise Exception(f"Could not create a connection the NAS server. The initial exception is: {e}")
        return conn

    def __delete__(self, instance):
        self.conn.close()

    def __unicode__(self):
        return f"NasServerClient (user: {user}, password: {password}, client_machine_name: {client_machine_name}, server_name: {server_name}, server_ip: {server_ip})"

    def __str__(self):
        return self.__unicode__()