Upgrade to Python 3.7+ syntax

README.md

@@ -59,7 +59,7 @@ try:
     from gssapi.raw import inquire_sec_context_by_oid
     print("python-gssapi extension is available")
 except ImportError as exc:
-    print("python-gssapi extension is not available: %s" % str(exc))
+    print(f"python-gssapi extension is not available: {exc}")
 ```
 
 If it isn't available, then either a newer version of the system's gssapi implementation needs to be setup and

build_helpers/check-smb.py

@@ -1,36 +1,40 @@
+import logging
 import os
 import time
 import uuid
 
 from smbprotocol.connection import Connection
 
+log = logging.getLogger(__name__)
+logging.basicConfig(level=logging.INFO)
+
 
 def test_connection(server, port):
     conn = Connection(uuid.uuid4(), server, port=port)
-    print("Opening connection to %s:%d" % (server, port))
+    log.info("Opening connection to %s:%d", server, port)
     conn.connect(timeout=5)
     conn.disconnect(True)
 
 
 if __name__ == "__main__":
     server = os.environ.get("SMB_SERVER", "127.0.0.1")
     port = int(os.environ.get("SMB_PORT", 445))
-    print("Waiting for SMB server to be online")
+    log.info("Waiting for SMB server to be online")
 
     attempt = 1
     total_attempts = 20
     while attempt < total_attempts:
-        print("Starting attempt %d" % attempt)
+        log.info("Starting attempt %d", attempt)
         try:
             test_connection(server, port)
             break
         except Exception as e:
-            print("Connection attempt %d failed: %s" % (attempt, str(e)))
+            log.info("Connection attempt %d failed: %s", attempt, e)
             attempt += 1
             if attempt == total_attempts:
-                raise Exception("Timeout while waiting for SMB server to come online")
+                raise Exception("Timeout while waiting for SMB server to come online") from e
 
-            print("Sleeping for 5 seconds before next attempt")
+            log.info("Sleeping for 5 seconds before next attempt")
             time.sleep(5)
 
-    print("Connection successful")
+    log.info("Connection successful")

examples/high-level/directory-management.py

@@ -1,6 +1,11 @@
+import logging
+
 from smbclient import listdir, mkdir, register_session, rmdir, scandir
 from smbclient.path import isdir
 
+log = logging.getLogger(__name__)
+logging.basicConfig(level=logging.INFO)
+
 # Optional - register the server with explicit credentials
 register_session("server", username="admin", password="pass")
 
@@ -12,18 +17,18 @@
 
 # Checking whether a file is a directory
 d_filename = r"\\server\share\directory"
-print("Is file {} dir?: {}".format(d_filename, isdir(d_filename)))
+log.info("Is file %s dir?: %b", d_filename, isdir(d_filename))
 
 # List the files/directories inside a dir
 for filename in listdir(r"\\server\share\directory"):
-    print(filename)
+    log.info(filename)
 
 # Use scandir as a more efficient directory listing as it already contains info like stat and attributes.
 for file_info in scandir(r"\\server\share\directory"):
     file_inode = file_info.inode()
     if file_info.is_file():
-        print("File: %s %d" % (file_info.name, file_inode))
+        log.info("File: %s %d", file_info.name, file_inode)
     elif file_info.is_dir():
-        print("Dir: %s %d" % (file_info.name, file_inode))
+        log.info("Dir: %s %d", file_info.name, file_inode)
     else:
-        print("Symlink: %s %d" % (file_info.name, file_inode))
+        log.info("Symlink: %s %d", file_info.name, file_inode)

examples/low-level/directory-management.py

@@ -1,3 +1,4 @@
+import logging
 import uuid
 
 from smbprotocol.connection import Connection
@@ -15,11 +16,14 @@
 from smbprotocol.session import Session
 from smbprotocol.tree import TreeConnect
 
+log = logging.getLogger(__name__)
+logging.basicConfig(level=logging.INFO)
+
 server = "127.0.0.1"
 port = 445
 username = "smbuser"
 password = "smbpassword"
-share = r"\\%s\share" % server
+share = rf"\\{server}\share"
 dir_name = "directory"
 
 connection = Connection(uuid.uuid4(), server, port)
@@ -55,7 +59,7 @@
     )
 
     compound_messages = [
-        directory_file.write("Hello World".encode("utf-8"), 0, send=False),
+        directory_file.write(b"Hello World", 0, send=False),
         dir_open.query_directory("*", FileInformationClass.FILE_NAMES_INFORMATION, send=False),
         directory_file.close(False, send=False),
         dir_open.close(False, send=False),
@@ -70,7 +74,7 @@
     for dir_file in responses[1]:
         dir_files.append(dir_file["file_name"].get_value().decode("utf-16-le"))
 
-    print("Directory '%s\\%s' contains the files: '%s'" % (share, dir_name, "', '".join(dir_files)))
+    log.info("Directory '%s\\%s' contains the files: %s", share, dir_name, ", ".join(repr(file) for file in dir_files))
 
     # delete a directory (note the dir needs to be empty to delete on close)
     dir_open = Open(tree, dir_name)

examples/low-level/file-management.py

@@ -1,3 +1,4 @@
+import logging
 import uuid
 
 from smbprotocol.connection import Connection
@@ -27,11 +28,14 @@
 from smbprotocol.structure import FlagField
 from smbprotocol.tree import TreeConnect
 
+log = logging.getLogger(__name__)
+logging.basicConfig(level=logging.INFO)
+
 server = "127.0.0.1"
 port = 445
 username = "smbuser"
 password = "smbpassword"
-share = r"\\%s\share" % server
+share = rf"\\{server}\share"
 file_name = "file-test.txt"
 
 connection = Connection(uuid.uuid4(), server, port)
@@ -83,15 +87,15 @@
     # flag field, set the value and get the human readble string
     max_access = FlagField(size=4, flag_type=FilePipePrinterAccessMask, flag_strict=False)
     max_access.set_value(open_info[0]["maximal_access"].get_value())
-    print("Maximum access mask for file %s\\%s: %s" % (share, file_name, str(max_access)))
+    log.info("Maximum access mask for file %s\\%s: %s", share, file_name, max_access)
 
     # write to a file
     text = "Hello World, what a nice day to play with SMB"
     file_open.write(text.encode("utf-8"), 0)
 
     # read from a file
     file_text = file_open.read(0, 1024)
-    print("Text of file %s\\%s: %s" % (share, file_name, file_text.decode("utf-8")))
+    log.info("Text of file %s\\%s: %s", share, file_name, file_text.decode("utf-8"))
     file_open.close(False)
 
     # read and delete a file in a single SMB packet instead of 3
@@ -116,6 +120,6 @@
     for i, request in enumerate(requests):
         response = delete_msgs[i][1](request)
         responses.append(response)
-    print("Text of file when reading/deleting in 1 request: %s" % responses[1].decode("utf-8"))
+    log.info("Text of file when reading/deleting in 1 request: %s", responses[1].decode("utf-8"))
 finally:
     connection.disconnect(True)

setup.py

@@ -1,5 +1,4 @@
 #!/usr/bin/env python
-# -*- coding: utf-8 -*-
 # Copyright: (c) 2019, Jordan Borean (@jborean93) <jborean93@gmail.com>
 # MIT License (see LICENSE or https://opensource.org/licenses/MIT)
 

src/smbclient/__init__.py

@@ -1,4 +1,3 @@
-# -*- coding: utf-8 -*-
 # Copyright: (c) 2019, Jordan Borean (@jborean93) <jborean93@gmail.com>
 # MIT License (see LICENSE or https://opensource.org/licenses/MIT)
 

src/smbclient/_io.py

@@ -1,4 +1,3 @@
-# -*- coding: utf-8 -*-
 # Copyright: (c) 2019, Jordan Borean (@jborean93) <jborean93@gmail.com>
 # MIT License (see LICENSE or https://opensource.org/licenses/MIT)
 
@@ -52,9 +51,8 @@ def _parse_share_access(raw, mode):
         for c in raw:
             access_val = share_access_map.get(c)
             if access_val is None:
-                raise ValueError(
-                    "Invalid share_access char %s, can only be %s" % (c, ", ".join(sorted(share_access_map.keys())))
-                )
+                chars = ", ".join(sorted(share_access_map.keys()))
+                raise ValueError(f"Invalid share_access char {c}, can only be {chars}")
             share_access |= access_val
 
         if "r" in mode:
@@ -82,12 +80,15 @@ def _parse_mode(raw, invalid=""):
     for c in raw:
         dispo_val = disposition_map.get(c)
         if dispo_val is None:
-            raise ValueError("Invalid mode char %s, can only be %s" % (c, ", ".join(sorted(disposition_map.keys()))))
+            chars = ", ".join(sorted(disposition_map.keys()))
+            raise ValueError(
+                f"Invalid mode char {c}, can only be {chars}",
+            )
 
         create_disposition |= dispo_val
 
     if create_disposition == 0:
-        raise ValueError("Invalid mode value %s, must contain at least r, w, x, or a" % raw)
+        raise ValueError(f"Invalid mode value {raw}, must contain at least r, w, x, or a")
 
     return create_disposition
 
@@ -102,7 +103,7 @@ def _chunk_size(connection, length, operation):
     :param operation: The operation the chunk is for: 'read', 'write', 'transact'
     :return: The size of the chunk we can use and the number of credits to request for the next operation.
     """
-    max_size = getattr(connection, "max_%s_size" % operation)
+    max_size = getattr(connection, f"max_{operation}_size")
 
     # Determine the maximum data length we can send for the operation. We do this by checking the available credits and
     # calculating whatever is the smallest; length, negotiated operation size, available credit size).
@@ -141,7 +142,7 @@ def _resolve_dfs(raw_io):
     if not info:
         raise ObjectPathNotFound()
 
-    connection_kwargs = getattr(raw_io, "_%s__kwargs" % SMBRawIO.__name__, {})
+    connection_kwargs = getattr(raw_io, f"_{SMBRawIO.__name__}__kwargs", {})
 
     for target in info:
         new_path = raw_path.replace(info.dfs_path, target.target_path, 1)
@@ -150,7 +151,7 @@ def _resolve_dfs(raw_io):
             tree, fd_path = get_smb_tree(new_path, **connection_kwargs)
 
         except SMBResponseException as link_exc:
-            log.warning("Failed to connect to DFS link target %s: %s", str(target), link_exc)
+            log.warning("Failed to connect to DFS link target %s", target, exc_info=link_exc)
             continue
 
         # Record the target that worked for future reference.
@@ -612,8 +613,7 @@ def query_directory(self, pattern, info_class):
                 break
 
             query_flags = 0  # Only the first request should have set SMB2_RESTART_SCANS
-            for entry in entries:
-                yield entry
+            yield from entries
 
     def readable(self):
         return False

src/smbclient/_os.py

@@ -1,4 +1,3 @@
-# -*- coding: utf-8 -*-
 # Copyright: (c) 2019, Jordan Borean (@jborean93) <jborean93@gmail.com>
 # MIT License (see LICENSE or https://opensource.org/licenses/MIT)
 
@@ -102,7 +101,7 @@
 )
 
 
-def is_remote_path(path):  # type: (str) -> bool
+def is_remote_path(path: str) -> bool:
     """
     Returns True iff the given path is a remote SMB path (rather than a local path).
 
@@ -182,9 +181,8 @@ def copyfile(src, dst, **kwargs):
                     copychunk_response = SMB2SrvCopyChunkResponse()
                     copychunk_response.unpack(result)
                     if copychunk_response["chunks_written"].get_value() != len(batch):
-                        raise IOError(
-                            "Failed to copy all the chunks in a server side copyfile: '%s' -> '%s'"
-                            % (norm_src, norm_dst)
+                        raise OSError(
+                            f"Failed to copy all the chunks in a server side copyfile: '{norm_src}' -> '{norm_dst}'"
                         )
 
 
@@ -433,7 +431,7 @@ def readlink(path, **kwargs):
     reparse_buffer = _get_reparse_point(norm_path, **kwargs)
     reparse_tag = reparse_buffer["reparse_tag"]
     if reparse_tag.get_value() != ReparseTags.IO_REPARSE_TAG_SYMLINK:
-        raise ValueError("Cannot read link of reparse point with tag %s at '%s'" % (str(reparse_tag), norm_path))
+        raise ValueError(f"Cannot read link of reparse point with tag {reparse_tag} at '{norm_path}'")
 
     symlink_buffer = SymbolicLinkReparseDataBuffer()
     symlink_buffer.unpack(reparse_buffer["data_buffer"].get_value())
@@ -549,7 +547,7 @@ def scandir(path, search_pattern="*", **kwargs):
                 continue
 
             dir_entry = SMBDirEntry(
-                SMBRawIO("%s\\%s" % (path, filename), **kwargs), dir_info, connection_cache=connection_cache
+                SMBRawIO(rf"{path}\{filename}", **kwargs), dir_info, connection_cache=connection_cache
             )
             yield dir_entry
 
@@ -696,12 +694,12 @@ def symlink(src, dst, target_is_directory=False, **kwargs):
         norm_src = ntpath.abspath(ntpath.join(dst_dir, norm_src))
     else:
         flags = SymbolicLinkFlags.SYMLINK_FLAG_ABSOLUTE
-        substitute_name = "\\??\\UNC\\%s" % norm_src[2:]
+        substitute_name = "\\??\\UNC\\" + norm_src[2:]
 
     src_drive = ntpath.splitdrive(norm_src)[0]
     dst_drive = ntpath.splitdrive(norm_dst)[0]
     if src_drive.lower() != dst_drive.lower():
-        raise ValueError("Resolved link src root '%s' must be the same as the dst root '%s'" % (src_drive, dst_drive))
+        raise ValueError(f"Resolved link src root '{src_drive}' must be the same as the dst root '{dst_drive}'")
 
     try:
         src_stat = stat(norm_src, **kwargs)
@@ -901,13 +899,11 @@ def walk(top, topdown=True, onerror=None, follow_symlinks=False, **kwargs):
             if not follow_symlinks and py_stat.S_ISLNK(lstat(dirpath, **kwargs).st_mode):
                 continue
 
-            for dir_top, dir_dirs, dir_files in walk(dirpath, **walk_kwargs):
-                yield dir_top, dir_dirs, dir_files
+            yield from walk(dirpath, **walk_kwargs)
     else:
         # On a bottom up approach we yield the sub directories before the top path.
         for dirpath in bottom_up_dirs:
-            for dir_top, dir_dirs, dir_files in walk(dirpath, **walk_kwargs):
-                yield dir_top, dir_dirs, dir_files
+            yield from walk(dirpath, **walk_kwargs)
 
         yield top, dirs, files
 
@@ -1095,7 +1091,7 @@ def _get_reparse_point(path, **kwargs):
 def _rename_information(src, dst, replace_if_exists=False, **kwargs):
     verb = "replace" if replace_if_exists else "rename"
     if not is_remote_path(ntpath.normpath(dst)):
-        raise ValueError("dst must be an absolute path to where the file or directory should be %sd." % verb)
+        raise ValueError(f"dst must be an absolute path to where the file or directory should be {verb}d.")
 
     raw_args = dict(kwargs)
     raw_args.update(
@@ -1126,7 +1122,7 @@ def _rename_information(src, dst, replace_if_exists=False, **kwargs):
         dst_share = ntpath.normpath(dst_raw.fd.tree_connect.share_name).split("\\")[-1]
 
         if src_guid != dst_guid or src_share.lower() != dst_share.lower():
-            raise ValueError("Cannot %s a file to a different root than the src." % verb)
+            raise ValueError(f"Cannot {verb} a file to a different root than the src.")
 
         dst_tree = dst_raw.fd.tree_connect
         dst_path = dst_raw.fd.file_name
@@ -1181,7 +1177,7 @@ def __init__(self, raw, dir_info, connection_cache=None):
         self._connection_cache = connection_cache
 
     def __str__(self):
-        return "<{0}: {1!r}>".format(self.__class__.__name__, self.name)
+        return f"<{self.__class__.__name__}: {self.name!r}>"
 
     @property
     def name(self):