Precommit

Author: TrevorBurgoyne

S3MP/__init__.py

@@ -1,2 +1,4 @@
 """S3 MirrorPath package."""
-from ._version import __version__
+from ._version import __version__
+
+__all__ = ["__version__"]

S3MP/async_utils.py

@@ -1,10 +1,13 @@
 """Asynchronous transfer utilities."""
-from S3MP.global_config import S3MPConfig
-import aioboto3
 import asyncio
 from typing import Coroutine, List
+
+import aioboto3
+
+from S3MP.global_config import S3MPConfig
 from S3MP.mirror_path import MirrorPath
 
+
 async def async_upload_from_mirror(mirror_path: MirrorPath):
     """Asynchronously upload a file from a MirrorPath."""
     session = aioboto3.Session()

S3MP/callbacks.py

@@ -1,13 +1,11 @@
-"""
-S3 callbacks to be used for boto3 transfers (uploads, downloads, and copies).
-"""
+"""S3 callbacks to be used for boto3 transfers (uploads, downloads, and copies)."""
 from pathlib import Path
-from S3MP.global_config import S3MPConfig
-import os
+
 import tqdm
 
+from S3MP.global_config import S3MPConfig
 from S3MP.mirror_path import MirrorPath
-from S3MP.types import SList, S3Resource
+from S3MP.types import S3Resource, SList
 
 
 class FileSizeTQDMCallback(tqdm.tqdm):
@@ -29,25 +27,31 @@ def __init__(
         :param is_download: Marker for upload/download transfer.
         """
         if transfer_objs is None:
-            return 
+            return
         if resource is None:
             resource = S3MPConfig.s3_resource
         if bucket_key is None:
             bucket_key = S3MPConfig.default_bucket_key
         if not isinstance(transfer_objs, list):
             transfer_objs = [transfer_objs]
-        
 
         self._total_bytes = 0
         for transfer_mapping in transfer_objs:
             if is_download:
-                s3_key = transfer_mapping.s3_key if isinstance(transfer_mapping, MirrorPath) else transfer_mapping
+                s3_key = (
+                    transfer_mapping.s3_key
+                    if isinstance(transfer_mapping, MirrorPath)
+                    else transfer_mapping
+                )
                 self._total_bytes += resource.Object(bucket_key, s3_key).content_length
             else:
-                local_path = transfer_mapping.local_path if isinstance(transfer_mapping, MirrorPath) else transfer_mapping
+                local_path = (
+                    transfer_mapping.local_path
+                    if isinstance(transfer_mapping, MirrorPath)
+                    else transfer_mapping
+                )
                 self._total_bytes += local_path.stat().st_size
 
-
         transfer_str = "Download" if is_download else "Upload"
         super().__init__(
             self,

S3MP/global_config.py

@@ -1,11 +1,14 @@
 """Set global values for S3MP module."""
+import tempfile
 from configparser import ConfigParser
 from dataclasses import dataclass
 from pathlib import Path
-import tempfile
 from typing import Callable
+
 import boto3
-from S3MP.types import S3Client, S3Resource, S3Bucket, S3TransferConfig
+
+from S3MP.types import S3Bucket, S3Client, S3Resource, S3TransferConfig
+
 
 def get_config_file_path() -> Path:
     """Get the location of the config file."""
@@ -14,17 +17,21 @@ def get_config_file_path() -> Path:
 
 
 class Singleton(type):
-    # Singleton metaclass 
+    """Singleton metaclass."""
+
     _instances = {}
 
     def __call__(cls, *args, **kwargs):
+        """Get instance of class."""
         if cls not in cls._instances:
             cls._instances[cls] = super().__call__(*args, **kwargs)
         return cls._instances[cls]
 
+
 @dataclass
 class S3MPConfig(metaclass=Singleton):
     """Singleton class for S3MP globals."""
+
     # Boto3 Objects
     _s3_client: S3Client = None
     _s3_resource: S3Resource = None
@@ -45,14 +52,14 @@ def s3_client(self) -> S3Client:
         if not self._s3_client:
             self._s3_client = boto3.client("s3")
         return self._s3_client
-    
+
     @property
     def s3_resource(self) -> S3Resource:
         """Get S3 resource."""
         if not self._s3_resource:
             self._s3_resource = boto3.resource("s3")
         return self._s3_resource
-    
+
     @property
     def bucket(self, bucket_key: str = None) -> S3Bucket:
         """Get bucket."""
@@ -63,42 +70,44 @@ def bucket(self, bucket_key: str = None) -> S3Bucket:
                 raise ValueError("No default bucket key set.")
             self._bucket = self.s3_resource.Bucket(self.default_bucket_key)
         return self._bucket
-    
+
     @property
     def mirror_root(self) -> Path:
         """Get mirror root."""
         if self._mirror_root is None:
-            print("Mirror Root not set, a temporary directory will be used as the mirror root.")
+            print(
+                "Mirror Root not set, a temporary directory will be used as the mirror root."
+            )
             self._mirror_root = Path(tempfile.gettempdir())
         return self._mirror_root
-    
+
     def load_config(self, config_file_path: Path = None):
         """Load the config file."""
         config_file_path = config_file_path or get_config_file_path()
         config = ConfigParser()
         config.read(config_file_path)
-        
+
         if "DEFAULT" not in config:
-            return 
-        
+            return
+
         if "default_bucket_key" in config["DEFAULT"]:
             self.default_bucket_key = config["DEFAULT"]["default_bucket_key"]
-        
+
         if "mirror_root" in config["DEFAULT"]:
             self._mirror_root = Path(config["DEFAULT"]["mirror_root"])
-    
+
     def save_config(self, config_file_path: Path = None):
         """Write config file."""
         config_file_path = config_file_path or get_config_file_path()
         config = ConfigParser()
-        config['DEFAULT'] = {}
+        config["DEFAULT"] = {}
         if self.default_bucket_key:
-            config['DEFAULT']['default_bucket_key'] = self.default_bucket_key
+            config["DEFAULT"]["default_bucket_key"] = self.default_bucket_key
         if self._mirror_root:
-            config['DEFAULT']['mirror_root'] = str(self._mirror_root)
-        with open(config_file_path, 'w') as configfile:
+            config["DEFAULT"]["mirror_root"] = str(self._mirror_root)
+        with open(config_file_path, "w") as configfile:
             config.write(configfile)
 
 
-S3MPConfig = S3MPConfig() 
+S3MPConfig = S3MPConfig()
 S3MPConfig.load_config()

S3MP/keys.py

@@ -1,9 +1,10 @@
 """S3 key modification utilities."""
-from enum import Enum
 import itertools
 from dataclasses import dataclass
+from enum import Enum
 from typing import List, Tuple
-from S3MP.prefix_queries import get_folders_within_folder, get_files_within_folder
+
+from S3MP.prefix_queries import get_files_within_folder, get_folders_within_folder
 
 
 @dataclass
@@ -21,16 +22,16 @@ def __call__(self, *args, **kwargs):
         """Set data via calling."""
         self = self.__copy__()
         if len(args) == 1:
-            if type(args[0]) == str:
-                self.name = args[0] 
+            if isinstance(args[0], str):
+                self.name = args[0]
             elif isinstance(args[0], Enum):
                 self.name = args[0].value
             else:
-                try: 
+                try:
                     self.name = str(args[0])
-                except:
+                except Exception:
                     raise TypeError(f"Cannot convert {args[0]} to str.")
-    
+
         for key in self.__dict__.keys():
             if key in kwargs:
                 setattr(self, key, kwargs[key])
@@ -40,11 +41,11 @@ def __call__(self, *args, **kwargs):
     def __copy__(self):
         """Copy."""
         return KeySegment(self.depth, self.name, self.is_file, self.incomplete_name)
-    
+
     def copy(self):
         """Copy."""
         return self.__copy__()
-    
+
     def __repr__(self):
         """Class representation."""
         return f"{self.__class__.__name__}(depth={self.depth}, name={self.name}, is_file={self.is_file}, incomplete_name={self.incomplete_name})"
@@ -77,7 +78,7 @@ def replace_key_segments(
     key: str, segments: List[KeySegment], max_len: int = None
 ) -> str:
     """Replace segments of a key with new segments."""
-    if type(segments) == KeySegment:
+    if isinstance(segments, KeySegment):
         segments = [segments]
     segments = sorted(segments, key=lambda x: x.depth)
     key_segments = key.split("/")
@@ -102,9 +103,10 @@ def replace_key_segments(
 def replace_key_segments_at_relative_depth(key: str, segments: List[KeySegment]) -> str:
     """
     Replace segments of a key with new segments at a relative depth.
+
     0 would be the deepest segment, -1 would be the second deepest, etc.
     """
-    if type(segments) == KeySegment:
+    if isinstance(segments, KeySegment):
         segments = [segments]
     segments = sorted(segments, key=lambda x: x.depth)
     key_segments = [seg for seg in key.split("/") if seg]
@@ -166,7 +168,7 @@ async def dfs_matching_key_gen(
 def sync_dfs_matching_key_gen(
     segments: List[KeySegment], path: str = None, current_depth: int = None
 ):
-    """Synchronous generation of all matching keys from a path, depth first."""
+    """Generate all matching keys synchronously from a path, depth first."""
     if current_depth is None:
         segments = sorted(segments, key=lambda x: x.depth)
         path, current_depth = build_s3_key(segments)

S3MP/multipart_uploads.py

@@ -1,12 +1,11 @@
 """S3MP multipart uploads."""
 import concurrent.futures
 import math
-import S3MP
-from S3MP.async_utils import sync_gather_threads
-from S3MP.global_config import S3MPConfig
-from S3MP.transfer_configs import MB
 
+import S3MP  # noqa: F401
+from S3MP.global_config import S3MPConfig
 from S3MP.mirror_path import MirrorPath
+from S3MP.transfer_configs import MB
 from S3MP.types import S3Bucket
 
 
@@ -53,7 +52,7 @@ def resume_multipart_upload(
     with open(mirror_path.local_path, "rb") as f:
         with concurrent.futures.ThreadPoolExecutor(max_workers=max_threads) as executor:
             # Verify existing parts.
-            assert(all(part.size == part_size for part in mpu_parts[:-1]))
+            assert all(part.size == part_size for part in mpu_parts[:-1])
 
             f.seek(part_size * n_uploaded_parts)
             if S3MPConfig.callback:
@@ -75,13 +74,12 @@ def resume_multipart_upload(
                 if S3MPConfig.callback:
                     S3MPConfig.callback(part_size)
 
-    obj = mpu.complete(
-        MultipartUpload=mpu_dict
-    )
+    obj = mpu.complete(MultipartUpload=mpu_dict)
     if abs(total_size_bytes - obj.content_length) > MB:
         print()
-        print(f"Uploaded size {obj.content_length} does not match local size {total_size_bytes}")
+        print(
+            f"Uploaded size {obj.content_length} does not match local size {total_size_bytes}"
+        )
         obj.delete()
         print("Deleted object, restarting upload.")
         return resume_multipart_upload(mirror_path, max_threads=max_threads)
-    

S3MP/prefix_queries.py

@@ -1,20 +1,20 @@
 """S3 prefix queries.."""
 from __future__ import annotations
+
 from typing import List
+
 from S3MP.global_config import S3MPConfig
 
 
 def get_prefix_paginator(folder_key: str, bucket_key: str = None, delimiter: str = "/"):
     """Get a paginator for a specified prefix."""
     if not bucket_key:
         bucket_key = S3MPConfig.default_bucket_key
-    if folder_key != '' and folder_key[-1] != "/":
+    if folder_key != "" and folder_key[-1] != "/":
         folder_key += "/"
     s3_client = S3MPConfig.s3_client
     paginator = s3_client.get_paginator("list_objects_v2")
-    return paginator.paginate(
-        Bucket=bucket_key, Prefix=folder_key, Delimiter=delimiter
-    )
+    return paginator.paginate(Bucket=bucket_key, Prefix=folder_key, Delimiter=delimiter)
 
 
 def get_files_within_folder(folder_key: str, key_filter: str = None) -> List[str]:
@@ -37,4 +37,3 @@ def get_folders_within_folder(folder_key: str, key_filter: str = None) -> List[s
                 if key_filter and key_filter not in obj:
                     continue
                 yield obj
-

S3MP/transfer_configs.py

@@ -1,21 +1,21 @@
 """Transfer configurations and utilities."""
+from s3transfer.constants import GB, KB, MB
+
 from S3MP.global_config import S3MPConfig
 from S3MP.types import S3TransferConfig
-from s3transfer.constants import KB, MB, GB
 
 
 def get_transfer_config(
-    n_threads: int, 
+    n_threads: int,
     block_size: int = 8 * MB,
     max_ram: int = 4 * GB,
     io_queue_size: int = 10e4,
     io_chunk_size: int = 256 * KB,
     set_global: bool = True,
 ) -> S3TransferConfig:
     """Get transfer config."""
-
     max_in_mem_upload_chunks = (max_ram - (n_threads * block_size)) // block_size
-    max_in_mem_download_chunks = (max_ram // block_size)
+    max_in_mem_download_chunks = max_ram // block_size
 
     config = S3TransferConfig(
         multipart_threshold=block_size,
@@ -27,7 +27,7 @@ def get_transfer_config(
         max_io_queue_size=io_queue_size,
         io_chunksize=io_chunk_size,
     )
-    config.use_threads = (n_threads > 1)
+    config.use_threads = n_threads > 1
     if set_global:
         S3MPConfig.transfer_config = config
-    return config 
+    return config