Improve unzip_file function. Reduce cognitive complexity.

Removed all unspecified exceptions. Removed unneeded calls to other functions.
Reduced size of try catch blocks to just parts of code that could throw an error.
This commit is contained in:
Wout Bouckaert 2024-08-16 15:25:04 -06:00
parent 890e84cd5e
commit bd021e0d43
No known key found for this signature in database
2 changed files with 63 additions and 41 deletions

View File

@ -13,6 +13,7 @@ import urllib.request
import ssl import ssl
import time import time
import certifi import certifi
from jsonschema.exceptions import ValidationError
from app.classes.helpers.helpers import Helpers from app.classes.helpers.helpers import Helpers
from app.classes.shared.console import Console from app.classes.shared.console import Console
@ -394,48 +395,69 @@ class FileHelpers:
return True return True
@staticmethod @staticmethod
def unzip_file(zip_path, server_update=False): def unzip_file(zip_path, server_update: bool = False) -> None:
ignored_names = ["server.properties", "permissions.json", "allowlist.json"] """
Unzips zip file at zip_path to location generated at new_dir based on zip
contents.
Args:
zip_path: Path to zip file to unzip.
server_update: Will skip ignored items list if not set to true. Used for
updating bedrock servers.
Returns: None
"""
ignored_names: list = [
"server.properties",
"permissions.json",
"allowlist.json",
]
# Get directory without zipfile name # Get directory without zipfile name
new_dir = pathlib.Path(zip_path).parents[0] new_dir = pathlib.Path(zip_path).parents[0]
# make sure we're able to access the zip file # make sure the directory we're unzipping this to exists
if Helpers.check_file_perms(zip_path) and os.path.isfile(zip_path): Helpers.ensure_dir_exists(new_dir)
# make sure the directory we're unzipping this to exists # we'll make a temporary directory to unzip this to.
Helpers.ensure_dir_exists(new_dir) temp_dir = tempfile.mkdtemp()
# we'll make a temporary directory to unzip this to. try:
temp_dir = tempfile.mkdtemp() with zipfile.ZipFile(zip_path, "r") as zip_ref:
try: # we'll extract this to the temp dir using zipfile module
with zipfile.ZipFile(zip_path, "r") as zip_ref: zip_ref.extractall(temp_dir)
# we'll extract this to the temp dir using zipfile module # Catch zipfile extract all error or file open errors.
zip_ref.extractall(temp_dir) except ValueError as why:
# we'll iterate through the top level directory moving everything Console.error(f"Unzip failed with information: {why}")
# out of the temp directory and into it's final home. raise RuntimeError(f"Unzip failed for path: {zip_path}") from why
for item in os.listdir(temp_dir): except FileNotFoundError as why:
# if the file is one of our ignored names we'll skip it Console.error(f"Unzip failed file not found: {zip_path}")
if item in ignored_names and server_update: raise FileNotFoundError(f"Unable to find file at path: {zip_path}") from why
continue except PermissionError as why:
# we handle files and dirs differently or we'll crash out. Console.error(f"Bad permissions for file at: {zip_path}")
if os.path.isdir(os.path.join(temp_dir, item)): raise PermissionError(f"Bad permissions for file at: {zip_path}") from why
try:
FileHelpers.move_dir_exist( # we'll iterate through the top level directory moving everything
os.path.join(temp_dir, item), # out of the temp directory and into it's final home.
os.path.join(new_dir, item), for item in os.listdir(temp_dir):
) # if the file is one of our ignored names we'll skip it
except Exception as ex: if item in ignored_names and server_update:
logger.error(f"ERROR IN ZIP IMPORT: {ex}") continue
else:
try: # we handle files and dirs differently or we'll crash out.
FileHelpers.move_file( if os.path.isdir(os.path.join(temp_dir, item)):
os.path.join(temp_dir, item), try:
os.path.join(new_dir, item), FileHelpers.move_dir_exist(
) os.path.join(temp_dir, item),
except Exception as ex: os.path.join(new_dir, item),
logger.error(f"ERROR IN ZIP IMPORT: {ex}") )
except Exception as ex: except shutil.Error as ex:
Console.error(ex) logger.error(f"ERROR IN ZIP IMPORT: {ex}")
else: else:
return "false" try:
return FileHelpers.move_file(
os.path.join(temp_dir, item),
os.path.join(new_dir, item),
)
except shutil.Error as ex:
logger.error(f"ERROR IN ZIP IMPORT: {ex}")
def unzip_server(self, zip_path, user_id): def unzip_server(self, zip_path, user_id):
if Helpers.check_file_perms(zip_path): if Helpers.check_file_perms(zip_path):

View File

@ -565,7 +565,7 @@ class ApiServersServerFilesZipHandler(BaseApiHandler):
}, },
) )
if Helpers.check_file_exists(folder): if Helpers.check_file_exists(folder):
folder = self.file_helper.unzip_file(folder, user_id) self.file_helper.unzip_file(folder, user_id)
else: else:
if user_id: if user_id:
return self.finish_json( return self.finish_json(