cleverlibre/clevercloud-storage
0
0
Fork 0
Code Issues 3 Pull Requests Actions 34 Packages Projects Releases Wiki Activity

style: Additional style fixes

Browse Source
This commit is contained in:
CoreRasurae
2025-09-20 13:11:15 +01:00
committed by Luis Mendes
parent fce62cf3d6
commit 4f409eb0bc
8 changed files with 277 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/clevercloud_storage_framework/client.py
+19 -11
View File
@@ -10,9 +10,7 @@ import logging
import os
from typing import Any, BinaryIO, Dict, Iterator, List, Optional, Union
from clevercloud_storage_framework.exceptions import (
StorageError
)
from clevercloud_storage_framework.exceptions import StorageError
from clevercloud_storage_framework.interface import StorageProviderInterface
@@ -127,7 +125,9 @@ class StorageClient:
return b"".join(chunks)
def write_file(self, path: str, content: Union[bytes, str, BinaryIO, Iterator[bytes]]) -> None:
def write_file(
self, path: str, content: Union[bytes, str, BinaryIO, Iterator[bytes]]
) -> None:
"""
Write content to a file.
@@ -180,7 +180,9 @@ class StorageClient:
# Get provider for the path, passing the SAS token if available
if sas_token:
source_provider = StorageClientFactory.get_provider_for_path(path, sas_token)
source_provider = StorageClientFactory.get_provider_for_path(
path, sas_token
)
else:
source_provider = StorageClientFactory.get_provider_for_path(path)
source_provider.delete_file(path)
@@ -219,7 +221,9 @@ class StorageClient:
return
# If the provider supports the source and can accept files from the source protocol
if source_supported and self.provider.supports_destination_protocol(source_protocol):
if source_supported and self.provider.supports_destination_protocol(
source_protocol
):
self.provider.copy_file(source_path, destination_path)
return
@@ -236,7 +240,9 @@ class StorageClient:
source_provider = self.provider
if not source_supported:
try:
source_provider = StorageClientFactory.get_provider_for_path(source_path)
source_provider = StorageClientFactory.get_provider_for_path(
source_path
)
except StorageError:
# Fall back to the current provider if none found
source_provider = self.provider
@@ -272,9 +278,9 @@ class StorageClient:
destination_path (str): The destination file path
"""
# Check if the current provider supports both paths
both_supported = self.provider.supports_path(source_path) and self.provider.supports_path(
destination_path
)
both_supported = self.provider.supports_path(
source_path
) and self.provider.supports_path(destination_path)
# If the current provider supports both paths, use its move_file method
if both_supported:
@@ -291,7 +297,9 @@ class StorageClient:
if not self.provider.supports_path(source_path):
try:
source_provider = StorageClientFactory.get_provider_for_path(source_path)
source_provider = StorageClientFactory.get_provider_for_path(
source_path
)
except StorageError:
source_provider = self.provider
src/clevercloud_storage_framework/factory.py
+9 -7
View File
@@ -95,8 +95,8 @@ class StorageClientFactory:
ProviderConfiguration: provider configuration instance for the requested provider and path, or None,
if no configuration could be found.
"""
candidate_configs: List[ProviderConfiguration] = cls._registered_configurations.get(
provider_type, []
candidate_configs: List[ProviderConfiguration] = (
cls._registered_configurations.get(provider_type, [])
)
for candidate_config in candidate_configs:
if candidate_config.matches_path_prefix(path):
@@ -115,8 +115,8 @@ class StorageClientFactory:
provider_type: str = provider_config.get_provider_type()
if provider_type not in StorageClientFactory.get_registered_providers_ids():
raise StorageError(f"Unsupported provider type: {provider_type}")
configs_for_provider: List[ProviderConfiguration] = cls._registered_configurations.get(
provider_type, []
configs_for_provider: List[ProviderConfiguration] = (
cls._registered_configurations.get(provider_type, [])
)
if configs_for_provider:
configs_for_provider.append(provider_config)
@@ -161,8 +161,8 @@ class StorageClientFactory:
from clevercloud_storage_framework.builder import ProviderBuilder
if apply_provider_config:
config: ProviderConfiguration = cls.get_registered_provider_config_by_provider_type(
provider_type, path
config: ProviderConfiguration = (
cls.get_registered_provider_config_by_provider_type(provider_type, path)
)
if config:
params: Dict[str, Any] = config.get_config_params()
@@ -234,7 +234,9 @@ class StorageClientFactory:
if provider_class.supports_path_format(path):
if apply_provider_config:
config: ProviderConfiguration = (
cls.get_registered_provider_config_by_provider_type(provider_type, path)
cls.get_registered_provider_config_by_provider_type(
provider_type, path
)
)
if config:
params: Dict[str, Any] = config.get_config_params()
src/clevercloud_storage_framework/interface.py
+6 -2
View File
@@ -33,7 +33,9 @@ class StorageProviderInterface(ABC):
pass
@abstractmethod
def list_files(self, path: str, recursive: bool = False) -> Iterator[Dict[str, Any]]:
def list_files(
self, path: str, recursive: bool = False
) -> Iterator[Dict[str, Any]]:
"""
List files in the specified path.
@@ -132,7 +134,9 @@ class StorageProviderInterface(ABC):
pass
@abstractmethod
def write_file(self, path: str, content: Union[bytes, BinaryIO, Iterator[bytes]]) -> None:
def write_file(
self, path: str, content: Union[bytes, BinaryIO, Iterator[bytes]]
) -> None:
"""
Write content to a file.
src/clevercloud_storage_framework/provider_configuration.py
+3 -1
View File
@@ -24,7 +24,9 @@ class ProviderConfiguration(object):
self._configuration: Dict[str, Any] = {}
if target_uri_prefix:
uri_provider_type = StorageClientFactory.get_provider_type_for_path(target_uri_prefix)
uri_provider_type = StorageClientFactory.get_provider_type_for_path(
target_uri_prefix
)
if uri_provider_type != provider_type:
raise InvalidPathError(
target_uri_prefix,
src/clevercloud_storage_framework/providers/efs_provider.py
+67 -22
View File
@@ -81,7 +81,9 @@ class EFSStorageProvider(StorageProviderInterface):
self.logger.debug("EFSStorageProvider initialized successfully")
except Exception as e:
self.logger.error(f"Failed to initialize EFSStorageProvider: {str(e)}")
raise StorageError(f"Failed to initialize EFS client: {str(e)}", provider="efs")
raise StorageError(
f"Failed to initialize EFS client: {str(e)}", provider="efs"
)
def _get_local_path(self, filesystem_id: str, path: str) -> str:
"""
@@ -148,7 +150,9 @@ class EFSStorageProvider(StorageProviderInterface):
)
else:
self.logger.error(f"EFS describe_file_systems error: {str(e)}")
raise StorageError(f"Failed to describe filesystem: {str(e)}", provider="efs")
raise StorageError(
f"Failed to describe filesystem: {str(e)}", provider="efs"
)
# Mount the filesystem
try:
@@ -170,7 +174,9 @@ class EFSStorageProvider(StorageProviderInterface):
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
raise StorageError(f"Failed to mount filesystem: {result.stderr}", provider="efs")
raise StorageError(
f"Failed to mount filesystem: {result.stderr}", provider="efs"
)
except ClientError as e:
error_code = e.response.get("Error", {}).get("Code", "")
@@ -181,7 +187,9 @@ class EFSStorageProvider(StorageProviderInterface):
)
else:
self.logger.error(f"EFS mount error: {str(e)}")
raise StorageError(f"Failed to mount filesystem: {str(e)}", provider="efs")
raise StorageError(
f"Failed to mount filesystem: {str(e)}", provider="efs"
)
except Exception as e:
self.logger.error(f"EFS mount error: {str(e)}")
raise StorageError(f"Failed to mount filesystem: {str(e)}", provider="efs")
@@ -189,7 +197,9 @@ class EFSStorageProvider(StorageProviderInterface):
def get_chunk_size(self) -> int:
return self.chunk_size
def list_files(self, path: str, recursive: bool = False) -> Iterator[Dict[str, Any]]:
def list_files(
self, path: str, recursive: bool = False
) -> Iterator[Dict[str, Any]]:
"""
List files in an EFS filesystem with the given path.
@@ -230,13 +240,16 @@ class EFSStorageProvider(StorageProviderInterface):
# Yield directories
for dir_name in dirs:
dir_path = os.path.join(root, dir_name)
rel_dir_path = os.path.join(rel_path, dir_name) if rel_path else dir_name
rel_dir_path = (
os.path.join(rel_path, dir_name) if rel_path else dir_name
)
stat = os.stat(dir_path)
yield {
"name": dir_name,
"path": f"efs://{filesystem_id}/{os.path.join(fs_path, rel_dir_path) if fs_path else rel_dir_path}",
"path": f"efs://{filesystem_id}/"
f"{os.path.join(fs_path, rel_dir_path) if fs_path else rel_dir_path}",
"size": 0,
"type": "directory",
"last_modified": stat.st_mtime,
@@ -246,13 +259,16 @@ class EFSStorageProvider(StorageProviderInterface):
# Yield files
for file_name in files:
file_path = os.path.join(root, file_name)
rel_file_path = os.path.join(rel_path, file_name) if rel_path else file_name
rel_file_path = (
os.path.join(rel_path, file_name) if rel_path else file_name
)
stat = os.stat(file_path)
yield {
"name": file_name,
"path": f"efs://{filesystem_id}/{os.path.join(fs_path, rel_file_path) if fs_path else rel_file_path}",
"path": f"efs://{filesystem_id}/"
f"{os.path.join(fs_path, rel_file_path) if fs_path else rel_file_path}",
"size": stat.st_size,
"type": "file",
"last_modified": stat.st_mtime,
@@ -314,7 +330,9 @@ class EFSStorageProvider(StorageProviderInterface):
return False
except Exception as e:
self.logger.error(f"EFS file_exists error: {str(e)}")
raise StorageError(f"Failed to check if file exists: {str(e)}", provider="efs")
raise StorageError(
f"Failed to check if file exists: {str(e)}", provider="efs"
)
def get_file_size(self, path: str) -> int:
"""
@@ -473,7 +491,9 @@ class EFSStorageProvider(StorageProviderInterface):
self.logger.error(f"EFS read_file error: {str(e)}")
raise StorageError(f"Failed to read file: {str(e)}", provider="efs")
def write_file(self, path: str, content: Union[bytes, BinaryIO, Iterator[bytes]]) -> None:
def write_file(
self, path: str, content: Union[bytes, BinaryIO, Iterator[bytes]]
) -> None:
"""
Write content to a file in EFS.
@@ -610,7 +630,9 @@ class EFSStorageProvider(StorageProviderInterface):
self._ensure_filesystem_mounted(dest_filesystem_id)
# Get the local paths
source_local_path = self._get_local_path(source_filesystem_id, source_fs_path)
source_local_path = self._get_local_path(
source_filesystem_id, source_fs_path
)
dest_local_path = self._get_local_path(dest_filesystem_id, dest_fs_path)
# Check if the source file exists
@@ -621,13 +643,20 @@ class EFSStorageProvider(StorageProviderInterface):
# Check if it's a file
if not os.path.isfile(source_local_path):
raise ResourceNotFoundError(f"Not a file: {source_path}", provider="efs")
raise ResourceNotFoundError(
f"Not a file: {source_path}", provider="efs"
)
# Ensure the destination directory exists
os.makedirs(os.path.dirname(os.path.abspath(dest_local_path)), exist_ok=True)
os.makedirs(
os.path.dirname(os.path.abspath(dest_local_path)), exist_ok=True
)
# Copy the file
with open(source_local_path, "rb") as src, open(dest_local_path, "wb") as dst:
with (
open(source_local_path, "rb") as src,
open(dest_local_path, "wb") as dst,
):
while True:
chunk = src.read(self.chunk_size)
if not chunk:
@@ -655,7 +684,9 @@ class EFSStorageProvider(StorageProviderInterface):
self._ensure_filesystem_mounted(source_filesystem_id)
# Get the local path
source_local_path = self._get_local_path(source_filesystem_id, source_fs_path)
source_local_path = self._get_local_path(
source_filesystem_id, source_fs_path
)
# Check if the source file exists
if not os.path.exists(source_local_path):
@@ -665,7 +696,9 @@ class EFSStorageProvider(StorageProviderInterface):
# Check if it's a file
if not os.path.isfile(source_local_path):
raise ResourceNotFoundError(f"Not a file: {source_path}", provider="efs")
raise ResourceNotFoundError(
f"Not a file: {source_path}", provider="efs"
)
# Ensure the destination directory exists
os.makedirs(
@@ -674,7 +707,10 @@ class EFSStorageProvider(StorageProviderInterface):
)
# Copy the file
with open(source_local_path, "rb") as src, open(destination_path, "wb") as dst:
with (
open(source_local_path, "rb") as src,
open(destination_path, "wb") as dst,
):
while True:
chunk = src.read(self.chunk_size)
if not chunk:
@@ -702,7 +738,9 @@ class EFSStorageProvider(StorageProviderInterface):
self._ensure_filesystem_mounted(dest_filesystem_id)
# Get the local path
dest_local_path = self._get_local_path(dest_filesystem_id, dest_fs_path)
dest_local_path = self._get_local_path(
dest_filesystem_id, dest_fs_path
)
# Check if the source file exists
if not os.path.exists(source_path):
@@ -711,10 +749,15 @@ class EFSStorageProvider(StorageProviderInterface):
)
# Ensure the destination directory exists
os.makedirs(os.path.dirname(os.path.abspath(dest_local_path)), exist_ok=True)
os.makedirs(
os.path.dirname(os.path.abspath(dest_local_path)), exist_ok=True
)
# Copy the file
with open(source_path, "rb") as src, open(dest_local_path, "wb") as dst:
with (
open(source_path, "rb") as src,
open(dest_local_path, "wb") as dst,
):
while True:
chunk = src.read(self.chunk_size)
if not chunk:
@@ -835,7 +878,9 @@ class EFSStorageProvider(StorageProviderInterface):
# Check if the directory exists
if not os.path.exists(local_path):
raise ResourceNotFoundError(f"Directory not found: {path}", provider="efs")
raise ResourceNotFoundError(
f"Directory not found: {path}", provider="efs"
)
# Check if it's a directory
if not os.path.isdir(local_path):
src/clevercloud_storage_framework/providers/glacier_provider.py
+13 -5
View File
@@ -14,7 +14,7 @@ from ..exceptions import (
InvalidPathError,
OperationNotSupportedError,
ResourceNotFoundError,
StorageError
StorageError,
)
from ..interface import StorageProviderInterface
from ..utils.stream import ChunkedReader
@@ -82,7 +82,9 @@ class GlacierStorageProvider(StorageProviderInterface):
InvalidPathError: If the path is not a valid Glacier path
"""
if not path or not self.supports_path(path):
raise InvalidPathError(path, "Path must be in the format 'glacier://vault/archive-id'")
raise InvalidPathError(
path, "Path must be in the format 'glacier://vault/archive-id'"
)
# Remove the 'glacier://' prefix
path = path[10:]
@@ -128,7 +130,9 @@ class GlacierStorageProvider(StorageProviderInterface):
"""
return path.startswith("glacier://")
def list_files(self, path: str, recursive: bool = False) -> Iterator[Dict[str, Any]]:
def list_files(
self, path: str, recursive: bool = False
) -> Iterator[Dict[str, Any]]:
"""
List files in a vault.
@@ -227,7 +231,9 @@ class GlacierStorageProvider(StorageProviderInterface):
except self.client.exceptions.ResourceNotFoundException:
raise ResourceNotFoundError(path, "glacier")
def write_file(self, path: str, content: Union[bytes, BinaryIO, Iterator[bytes]]) -> None:
def write_file(
self, path: str, content: Union[bytes, BinaryIO, Iterator[bytes]]
) -> None:
"""
Write a file.
@@ -255,7 +261,9 @@ class GlacierStorageProvider(StorageProviderInterface):
chunks = list(content)
file_obj = io.BytesIO(b"".join(chunks))
else:
raise ValueError("Content must be bytes, a file-like object, or an iterator of bytes")
raise ValueError(
"Content must be bytes, a file-like object, or an iterator of bytes"
)
# Upload the archive
self.client.upload_archive(vaultName=vault_name, body=file_obj)
src/clevercloud_storage_framework/providers/local_provider.py
+67 -23
View File
@@ -14,7 +14,7 @@ from typing import Any, BinaryIO, Dict, Iterator, Optional, Tuple, Union
from clevercloud_storage_framework.exceptions import (
InvalidPathError,
ResourceNotFoundError,
StorageError
StorageError,
)
from clevercloud_storage_framework.interface import StorageProviderInterface
@@ -50,7 +50,9 @@ class LocalStorageProvider(StorageProviderInterface):
def get_chunk_size(self) -> int:
return self.chunk_size
def list_files(self, path: str, recursive: bool = False) -> Iterator[Dict[str, Any]]:
def list_files(
self, path: str, recursive: bool = False
) -> Iterator[Dict[str, Any]]:
"""
List files in the specified local directory.
@@ -73,7 +75,9 @@ class LocalStorageProvider(StorageProviderInterface):
# Check if it's a directory
if not os.path.isdir(path):
raise ResourceNotFoundError(f"Not a directory: {path}", provider="local")
raise ResourceNotFoundError(
f"Not a directory: {path}", provider="local"
)
# List files
if recursive:
@@ -86,7 +90,9 @@ class LocalStorageProvider(StorageProviderInterface):
# Yield directories
for dir_name in dirs:
dir_path = os.path.join(root, dir_name)
rel_dir_path = os.path.join(rel_path, dir_name) if rel_path else dir_name
rel_dir_path = (
os.path.join(rel_path, dir_name) if rel_path else dir_name
)
stat = os.stat(dir_path)
@@ -102,7 +108,9 @@ class LocalStorageProvider(StorageProviderInterface):
# Yield files
for file_name in files:
file_path = os.path.join(root, file_name)
rel_file_path = os.path.join(rel_path, file_name) if rel_path else file_name
rel_file_path = (
os.path.join(rel_path, file_name) if rel_path else file_name
)
stat = os.stat(file_path)
@@ -134,7 +142,9 @@ class LocalStorageProvider(StorageProviderInterface):
except PermissionError:
# Re-raise permission errors
raise PermissionError(f"Permission denied for listing: {path}", provider="local")
raise PermissionError(
f"Permission denied for listing: {path}", provider="local"
)
except ResourceNotFoundError:
# Re-raise resource not found errors
raise
@@ -183,7 +193,9 @@ class LocalStorageProvider(StorageProviderInterface):
except PermissionError:
# Re-raise permission errors
raise PermissionError(f"Permission denied for accessing: {path}", provider="local")
raise PermissionError(
f"Permission denied for accessing: {path}", provider="local"
)
except ResourceNotFoundError:
# Re-raise resource not found errors
raise
@@ -231,13 +243,17 @@ class LocalStorageProvider(StorageProviderInterface):
except PermissionError:
# Re-raise permission errors
raise PermissionError(f"Permission denied for accessing: {path}", provider="local")
raise PermissionError(
f"Permission denied for accessing: {path}", provider="local"
)
except ResourceNotFoundError:
# Re-raise resource not found errors
raise
except Exception as e:
self.logger.error(f"Local get_file_metadata error: {str(e)}")
raise StorageError(f"Failed to get file metadata: {str(e)}", provider="local")
raise StorageError(
f"Failed to get file metadata: {str(e)}", provider="local"
)
def read_file(
self, path: str, offset: int = 0, length: Optional[int] = None
@@ -290,7 +306,9 @@ class LocalStorageProvider(StorageProviderInterface):
except PermissionError:
# Re-raise permission errors
raise PermissionError(f"Permission denied for reading: {path}", provider="local")
raise PermissionError(
f"Permission denied for reading: {path}", provider="local"
)
except ResourceNotFoundError:
# Re-raise resource not found errors
raise
@@ -298,7 +316,9 @@ class LocalStorageProvider(StorageProviderInterface):
self.logger.error(f"Local read_file error: {str(e)}")
raise StorageError(f"Failed to read file: {str(e)}", provider="local")
def write_file(self, path: str, content: Union[bytes, BinaryIO, Iterator[bytes]]) -> None:
def write_file(
self, path: str, content: Union[bytes, BinaryIO, Iterator[bytes]]
) -> None:
"""
Write content to a file in the local filesystem.
@@ -342,7 +362,9 @@ class LocalStorageProvider(StorageProviderInterface):
except PermissionError:
# Re-raise permission errors
raise PermissionError(f"Permission denied for writing to: {path}", provider="local")
raise PermissionError(
f"Permission denied for writing to: {path}", provider="local"
)
except Exception as e:
self.logger.error(f"Local write_file error: {str(e)}")
raise StorageError(f"Failed to write file: {str(e)}", provider="local")
@@ -373,7 +395,9 @@ class LocalStorageProvider(StorageProviderInterface):
except PermissionError:
# Re-raise permission errors
raise PermissionError(f"Permission denied for deleting: {path}", provider="local")
raise PermissionError(
f"Permission denied for deleting: {path}", provider="local"
)
except ResourceNotFoundError:
# Re-raise resource not found errors
raise
@@ -412,10 +436,14 @@ class LocalStorageProvider(StorageProviderInterface):
# Check if it's a file
if not os.path.isfile(source_path):
raise ResourceNotFoundError(f"Not a file: {source_path}", provider="local")
raise ResourceNotFoundError(
f"Not a file: {source_path}", provider="local"
)
# Ensure the destination directory exists
os.makedirs(os.path.dirname(os.path.abspath(destination_path)), exist_ok=True)
os.makedirs(
os.path.dirname(os.path.abspath(destination_path)), exist_ok=True
)
# Copy the file
shutil.copy2(source_path, destination_path)
@@ -428,7 +456,9 @@ class LocalStorageProvider(StorageProviderInterface):
StorageClientFactory,
)
source_provider = StorageClientFactory.get_provider_for_path(source_path)
source_provider = StorageClientFactory.get_provider_for_path(
source_path
)
# Read from source provider and write to local
content = source_provider.read_file(source_path)
@@ -476,10 +506,14 @@ class LocalStorageProvider(StorageProviderInterface):
# Check if it's a file
if not os.path.isfile(source_path):
raise ResourceNotFoundError(f"Not a file: {source_path}", provider="local")
raise ResourceNotFoundError(
f"Not a file: {source_path}", provider="local"
)
# Ensure the destination directory exists
os.makedirs(os.path.dirname(os.path.abspath(destination_path)), exist_ok=True)
os.makedirs(
os.path.dirname(os.path.abspath(destination_path)), exist_ok=True
)
# Move the file
shutil.move(source_path, destination_path)
@@ -492,7 +526,9 @@ class LocalStorageProvider(StorageProviderInterface):
StorageClientFactory,
)
source_provider = StorageClientFactory.get_provider_for_path(source_path)
source_provider = StorageClientFactory.get_provider_for_path(
source_path
)
# Copy the file
self.copy_file(source_path, destination_path)
@@ -533,7 +569,9 @@ class LocalStorageProvider(StorageProviderInterface):
)
except Exception as e:
self.logger.error(f"Local create_directory error: {str(e)}")
raise StorageError(f"Failed to create directory: {str(e)}", provider="local")
raise StorageError(
f"Failed to create directory: {str(e)}", provider="local"
)
def delete_directory(self, path: str, recursive: bool = False) -> None:
"""
@@ -551,11 +589,15 @@ class LocalStorageProvider(StorageProviderInterface):
try:
# Check if the directory exists
if not os.path.exists(path):
raise ResourceNotFoundError(f"Directory not found: {path}", provider="local")
raise ResourceNotFoundError(
f"Directory not found: {path}", provider="local"
)
# Check if it's a directory
if not os.path.isdir(path):
raise ResourceNotFoundError(f"Not a directory: {path}", provider="local")
raise ResourceNotFoundError(
f"Not a directory: {path}", provider="local"
)
# Delete the directory
if recursive:
@@ -581,7 +623,9 @@ class LocalStorageProvider(StorageProviderInterface):
raise
except Exception as e:
self.logger.error(f"Local delete_directory error: {str(e)}")
raise StorageError(f"Failed to delete directory: {str(e)}", provider="local")
raise StorageError(
f"Failed to delete directory: {str(e)}", provider="local"
)
def get_provider_name(self) -> str:
"""
src/clevercloud_storage_framework/providers/s3_provider.py
+93 -31
View File
@@ -94,7 +94,9 @@ class S3StorageProvider(StorageProviderInterface):
self.logger.debug("S3StorageProvider initialized successfully")
except Exception as e:
self.logger.error(f"Failed to initialize S3StorageProvider: {str(e)}")
raise StorageError(f"Failed to initialize S3 client: {str(e)}", provider="s3")
raise StorageError(
f"Failed to initialize S3 client: {str(e)}", provider="s3"
)
def _append_sas_token(self, url: str) -> str:
"""
@@ -121,7 +123,9 @@ class S3StorageProvider(StorageProviderInterface):
def get_chunk_size(self) -> int:
return self.chunk_size
def list_files(self, path: str, recursive: bool = False) -> Iterator[Dict[str, Any]]:
def list_files(
self, path: str, recursive: bool = False
) -> Iterator[Dict[str, Any]]:
"""
List files in an S3 bucket with the given prefix.
@@ -201,7 +205,9 @@ class S3StorageProvider(StorageProviderInterface):
error_code = e.response.get("Error", {}).get("Code", "")
if error_code == "NoSuchBucket":
raise ResourceNotFoundError(f"Bucket not found: {bucket}", provider="s3")
raise ResourceNotFoundError(
f"Bucket not found: {bucket}", provider="s3"
)
elif error_code in ("AccessDenied", "AllAccessDisabled"):
raise StoragePermissionError(
f"Permission denied for bucket: {bucket}", provider="s3"
@@ -240,10 +246,14 @@ class S3StorageProvider(StorageProviderInterface):
return False
elif error_code in ("403", "AccessDenied"):
# If we get access denied, we can't determine if the file exists
raise StoragePermissionError(f"Permission denied for object: {path}", provider="s3")
raise StoragePermissionError(
f"Permission denied for object: {path}", provider="s3"
)
else:
self.logger.error(f"S3 file_exists error: {str(e)}")
raise StorageError(f"Failed to check if file exists: {str(e)}", provider="s3")
raise StorageError(
f"Failed to check if file exists: {str(e)}", provider="s3"
)
def get_file_size(self, path: str) -> int:
"""
@@ -272,7 +282,9 @@ class S3StorageProvider(StorageProviderInterface):
if error_code in ("404", "NoSuchKey", "NotFound"):
raise ResourceNotFoundError(f"File not found: {path}", provider="s3")
elif error_code in ("403", "AccessDenied"):
raise StoragePermissionError(f"Permission denied for object: {path}", provider="s3")
raise StoragePermissionError(
f"Permission denied for object: {path}", provider="s3"
)
else:
self.logger.error(f"S3 get_file_size error: {str(e)}")
raise StorageError(f"Failed to get file size: {str(e)}", provider="s3")
@@ -318,10 +330,14 @@ class S3StorageProvider(StorageProviderInterface):
if error_code in ("404", "NoSuchKey", "NotFound"):
raise ResourceNotFoundError(f"File not found: {path}", provider="s3")
elif error_code in ("403", "AccessDenied"):
raise StoragePermissionError(f"Permission denied for object: {path}", provider="s3")
raise StoragePermissionError(
f"Permission denied for object: {path}", provider="s3"
)
else:
self.logger.error(f"S3 get_file_metadata error: {str(e)}")
raise StorageError(f"Failed to get file metadata: {str(e)}", provider="s3")
raise StorageError(
f"Failed to get file metadata: {str(e)}", provider="s3"
)
def read_file(
self, path: str, offset: int = 0, length: Optional[int] = None
@@ -371,7 +387,9 @@ class S3StorageProvider(StorageProviderInterface):
headers["Range"] = range_header
# Stream the response without loading everything into memory
response = self.http_session.get(presigned_url, headers=headers, stream=True)
response = self.http_session.get(
presigned_url, headers=headers, stream=True
)
response.raise_for_status()
# Create a custom iterator to yield chunks of the requested size
@@ -399,7 +417,9 @@ class S3StorageProvider(StorageProviderInterface):
if error_code in ("404", "NoSuchKey", "NotFound"):
raise ResourceNotFoundError(f"File not found: {path}", provider="s3")
elif error_code in ("403", "AccessDenied"):
raise StoragePermissionError(f"Permission denied for object: {path}", provider="s3")
raise StoragePermissionError(
f"Permission denied for object: {path}", provider="s3"
)
else:
self.logger.error(f"S3 read_file error: {str(e)}")
raise StorageError(f"Failed to read file: {str(e)}", provider="s3")
@@ -441,9 +461,13 @@ class S3StorageProvider(StorageProviderInterface):
return url
except ClientError as e:
self.logger.error(f"S3 generate_presigned_url error: {str(e)}")
raise StorageError(f"Failed to generate pre-signed URL: {str(e)}", provider="s3")
raise StorageError(
f"Failed to generate pre-signed URL: {str(e)}", provider="s3"
)
def write_file(self, path: str, content: Union[bytes, BinaryIO, Iterator[bytes]]) -> None:
def write_file(
self, path: str, content: Union[bytes, BinaryIO, Iterator[bytes]]
) -> None:
"""
Write content to a file in S3.
@@ -470,7 +494,9 @@ class S3StorageProvider(StorageProviderInterface):
and not isinstance(content, (bytes, bytearray))
):
# Initialize multipart upload
response = self.client.create_multipart_upload(Bucket=bucket, Key=key)
response = self.client.create_multipart_upload(
Bucket=bucket, Key=key
)
upload_id = response["UploadId"]
try:
@@ -479,7 +505,9 @@ class S3StorageProvider(StorageProviderInterface):
part_number = 1
# Use ChunkedWriter with preserve_chunks=True to maintain individual chunks
writer = ChunkedWriter(content, self.chunk_size, preserve_chunks=True)
writer = ChunkedWriter(
content, self.chunk_size, preserve_chunks=True
)
for chunk in writer:
if not chunk: # Skip empty chunks
@@ -530,7 +558,9 @@ class S3StorageProvider(StorageProviderInterface):
# Complete the multipart upload
complete_body = {"Parts": parts}
response = self.http_session.post(complete_url, json=complete_body)
response = self.http_session.post(
complete_url, json=complete_body
)
response.raise_for_status()
except Exception as e:
@@ -551,7 +581,9 @@ class S3StorageProvider(StorageProviderInterface):
# Abort the multipart upload
self.http_session.delete(abort_url)
raise StorageError(f"Failed to write file: {str(e)}", provider="s3")
raise StorageError(
f"Failed to write file: {str(e)}", provider="s3"
)
return
else:
@@ -600,7 +632,9 @@ class S3StorageProvider(StorageProviderInterface):
part_number = 1
# Use ChunkedWriter with preserve_chunks=True to maintain individual chunks
writer = ChunkedWriter(content, self.chunk_size, preserve_chunks=True)
writer = ChunkedWriter(
content, self.chunk_size, preserve_chunks=True
)
# Upload each chunk as a separate part
for chunk in writer:
@@ -617,7 +651,9 @@ class S3StorageProvider(StorageProviderInterface):
)
# Add the part to the list
parts.append({"PartNumber": part_number, "ETag": response["ETag"]})
parts.append(
{"PartNumber": part_number, "ETag": response["ETag"]}
)
part_number += 1
@@ -630,7 +666,9 @@ class S3StorageProvider(StorageProviderInterface):
)
except Exception as e:
# Abort the multipart upload on error
self.client.abort_multipart_upload(Bucket=bucket, Key=key, UploadId=upload_id)
self.client.abort_multipart_upload(
Bucket=bucket, Key=key, UploadId=upload_id
)
raise e
else:
raise ValueError(f"Unsupported content type: {type(content)}")
@@ -639,7 +677,9 @@ class S3StorageProvider(StorageProviderInterface):
error_code = e.response.get("Error", {}).get("Code", "")
if error_code in ("403", "AccessDenied"):
raise StoragePermissionError(f"Permission denied for object: {path}", provider="s3")
raise StoragePermissionError(
f"Permission denied for object: {path}", provider="s3"
)
else:
self.logger.error(f"S3 write_file error: {str(e)}")
raise StorageError(f"Failed to write file: {str(e)}", provider="s3")
@@ -671,7 +711,9 @@ class S3StorageProvider(StorageProviderInterface):
error_code = e.response.get("Error", {}).get("Code", "")
if error_code in ("403", "AccessDenied"):
raise StoragePermissionError(f"Permission denied for object: {path}", provider="s3")
raise StoragePermissionError(
f"Permission denied for object: {path}", provider="s3"
)
else:
self.logger.error(f"S3 delete_file error: {str(e)}")
raise StorageError(f"Failed to delete file: {str(e)}", provider="s3")
@@ -727,7 +769,9 @@ class S3StorageProvider(StorageProviderInterface):
copy_source = {"Bucket": source_bucket, "Key": source_key}
# Initiate multipart upload
response = self.client.create_multipart_upload(Bucket=dest_bucket, Key=dest_key)
response = self.client.create_multipart_upload(
Bucket=dest_bucket, Key=dest_key
)
upload_id = response["UploadId"]
try:
@@ -788,7 +832,9 @@ class S3StorageProvider(StorageProviderInterface):
)
# Ensure the destination directory exists
os.makedirs(os.path.dirname(os.path.abspath(destination_path)), exist_ok=True)
os.makedirs(
os.path.dirname(os.path.abspath(destination_path)), exist_ok=True
)
# Download the file
with open(destination_path, "wb") as f:
@@ -820,9 +866,13 @@ class S3StorageProvider(StorageProviderInterface):
error_code = e.response.get("Error", {}).get("Code", "")
if error_code in ("404", "NoSuchKey", "NotFound"):
raise ResourceNotFoundError(f"Source file not found: {source_path}", provider="s3")
raise ResourceNotFoundError(
f"Source file not found: {source_path}", provider="s3"
)
elif error_code in ("403", "AccessDenied"):
raise StoragePermissionError(f"Permission denied for copy operation", provider="s3")
raise StoragePermissionError(
"Permission denied for copy operation", provider="s3"
)
else:
self.logger.error(f"S3 copy_file error: {str(e)}")
raise StorageError(f"Failed to copy file: {str(e)}", provider="s3")
@@ -886,7 +936,9 @@ class S3StorageProvider(StorageProviderInterface):
)
else:
self.logger.error(f"S3 create_directory error: {str(e)}")
raise StorageError(f"Failed to create directory: {str(e)}", provider="s3")
raise StorageError(
f"Failed to create directory: {str(e)}", provider="s3"
)
def delete_directory(self, path: str, recursive: bool = False) -> None:
"""
@@ -921,17 +973,23 @@ class S3StorageProvider(StorageProviderInterface):
break
if not has_objects:
raise ResourceNotFoundError(f"Directory not found: {path}", provider="s3")
raise ResourceNotFoundError(
f"Directory not found: {path}", provider="s3"
)
# If not recursive, check if there are objects other than the directory marker
if not recursive:
paginator = self.client.get_paginator("list_objects_v2")
page_iterator = paginator.paginate(Bucket=bucket, Prefix=key, Delimiter="/")
page_iterator = paginator.paginate(
Bucket=bucket, Prefix=key, Delimiter="/"
)
for page in page_iterator:
# Check if there are any objects other than the directory marker
contents = page.get("Contents", [])
if len(contents) > 1 or (len(contents) == 1 and contents[0].get("Key") != key):
if len(contents) > 1 or (
len(contents) == 1 and contents[0].get("Key") != key
):
raise StorageError(
f"Directory not empty: {path}. Use recursive=True to delete non-empty directories.",
provider="s3",
@@ -954,7 +1012,9 @@ class S3StorageProvider(StorageProviderInterface):
if "Contents" in page:
objects = [{"Key": obj["Key"]} for obj in page["Contents"]]
if objects:
self.client.delete_objects(Bucket=bucket, Delete={"Objects": objects})
self.client.delete_objects(
Bucket=bucket, Delete={"Objects": objects}
)
else:
# Delete only the directory marker
self.client.delete_object(Bucket=bucket, Key=key)
@@ -968,7 +1028,9 @@ class S3StorageProvider(StorageProviderInterface):
)
else:
self.logger.error(f"S3 delete_directory error: {str(e)}")
raise StorageError(f"Failed to delete directory: {str(e)}", provider="s3")
raise StorageError(
f"Failed to delete directory: {str(e)}", provider="s3"
)
def get_provider_name(self) -> str:
"""