Merge pull request #42 from Precioussheep/feature/vsc-refactor

Refactor `vscodeoffline/vsc.py`

Author: LOLINTERNETZ

.gitignore

@@ -2,4 +2,5 @@ artifacts/*
 __pycache__/
 temp/
 venv/
-.vscode/*
+.vscode/*
+.env

vscoffline/server.py

@@ -29,11 +29,10 @@ def on_get(self, req, resp, platform, buildquality, commitid):
             resp.status = falcon.HTTP_204
             return
         name = latest['name']
-        updateglob = os.path.join(updatedir, f'vscode-{name}.*')
-        updatepath = vsc.Utility.first_file(updateglob)
+        updatepath = vsc.Utility.first_file(updatedir, f'vscode-{name}.*')
         if not updatepath:
             resp.content = 'Unable to find update payload'
-            log.warning(f'Unable to find update payload from {updateglob}')
+            log.warning(f'Unable to find update payload from {updatedir}/vscode-{name}.*')
             resp.status = falcon.HTTP_404
             return
         if not vsc.Utility.hash_file_and_check(updatepath, latest['sha256hash']):
@@ -63,10 +62,9 @@ def on_get(self, req, resp, commitid, platform, buildquality):
             resp.status = falcon.HTTP_500
             return
         name = updatejson['name']
-        updateglob = os.path.join(updatedir, f'vscode-{name}.*')
-        updatepath = vsc.Utility.first_file(updateglob)
+        updatepath = vsc.Utility.first_file(updatedir, f'vscode-{name}.*')
         if not updatepath:
-            resp.content = f'Unable to find update payload from {updateglob}'
+            resp.content = f'Unable to find update payload from {updatedir}/vscode-{name}.*'
             log.warning(resp.content)
             resp.status = falcon.HTTP_404
             return

vscoffline/vsc.py

@@ -1,35 +1,35 @@
-import os
-import io
-import json
-import hashlib
-import glob
 import datetime
+import hashlib
+import json
+import os
+import pathlib
 from enum import IntFlag
+from typing import Any, Dict, List, Union
+
 from logzero import logger as log
 
-PLATFORMS = ['win32', 'linux', 'linux-deb', 'linux-rpm',
-             'darwin', 'linux-snap', 'server-linux']
-ARCHITECTURES = ['', 'x64']
-BUILDTYPES = ['', 'archive', 'user']
-QUALITIES = ['stable', 'insider']
+PLATFORMS = ["win32", "linux", "linux-deb", "linux-rpm", "darwin", "linux-snap", "server-linux"]
+ARCHITECTURES = ["", "x64"]
+BUILDTYPES = ["", "archive", "user"]
+QUALITIES = ["stable", "insider"]
 
-URL_BINUPDATES = r'https://update.code.visualstudio.com/api/update/'
-URL_RECOMMENDATIONS = r'https://az764295.vo.msecnd.net/extensions/workspaceRecommendations.json.gz'
-URL_MARKETPLACEQUERY = r'https://marketplace.visualstudio.com/_apis/public/gallery/extensionquery'
-URL_MALICIOUS = r'https://az764295.vo.msecnd.net/extensions/marketplace.json'
+URL_BINUPDATES = r"https://update.code.visualstudio.com/api/update/"
+URL_RECOMMENDATIONS = r"https://az764295.vo.msecnd.net/extensions/workspaceRecommendations.json.gz"
+URL_MARKETPLACEQUERY = r"https://marketplace.visualstudio.com/_apis/public/gallery/extensionquery"
+URL_MALICIOUS = r"https://az764295.vo.msecnd.net/extensions/marketplace.json"
 
-URLROOT = 'https://update.code.visualstudio.com'
-ARTIFACTS = '/artifacts/'
-ARTIFACTS_INSTALLERS = '/artifacts/installers'
-ARTIFACTS_EXTENSIONS = '/artifacts/extensions'
-ARTIFACT_RECOMMENDATION = '/artifacts/recommendations.json'
-ARTIFACT_MALICIOUS = '/artifacts/malicious.json'
+URLROOT = "https://update.code.visualstudio.com"
+ARTIFACTS = "/artifacts/"
+ARTIFACTS_INSTALLERS = "/artifacts/installers"
+ARTIFACTS_EXTENSIONS = "/artifacts/extensions"
+ARTIFACT_RECOMMENDATION = "/artifacts/recommendations.json"
+ARTIFACT_MALICIOUS = "/artifacts/malicious.json"
 
 TIMEOUT = 12
 
 
 class QueryFlags(IntFlag):
-    __no_flags_name__ = 'NoneDefined'
+    __no_flags_name__ = "NoneDefined"
     NoneDefined = 0x0
     IncludeVersions = 0x1
     IncludeFiles = 0x2
@@ -45,7 +45,7 @@ class QueryFlags(IntFlag):
 
 
 class FilterType(IntFlag):
-    __no_flags_name__ = 'Target'
+    __no_flags_name__ = "Target"
     Tag = 1
     ExtensionId = 4
     Category = 5
@@ -58,7 +58,7 @@ class FilterType(IntFlag):
 
 
 class SortBy(IntFlag):
-    __no_flags_name__ = 'NoneOrRelevance'
+    __no_flags_name__ = "NoneOrRelevance"
     NoneOrRelevance = 0
     LastUpdatedDate = 1
     Title = 2
@@ -70,109 +70,118 @@ class SortBy(IntFlag):
 
 
 class SortOrder(IntFlag):
-    __no_flags_name__ = 'Default'
+    __no_flags_name__ = "Default"
     Default = 0
     Ascending = 1
     Descending = 2
 
 
 class MagicJsonEncoder(json.JSONEncoder):
-    def default(self, o):
-        if isinstance(o, datetime.datetime):
-            return o.isoformat()
-        return o.__dict__
+    def default(self, o: Any) -> Union[str, Dict[str, Any]]:
+        try:
+            return super().default(o)
+        except TypeError as err:
+            # could be datetime
+            if isinstance(o, datetime.datetime):
+                return o.isoformat()
+            # could also be cls with slots
+            try:
+                return {key: getattr(o, key, None) for key in o.__slots__}
+            except AttributeError:
+                pass
+            # finally, should have a dict if it is a dataclass or another cls
+            try:
+                return o.__dict__
+            except AttributeError:
+                raise TypeError(
+                    "Can't encode object. Tried isoformat of datetime, class slots and class dict"
+                ) from err
 
 
-class Utility(object):
+class Utility:
     """
     Utility tool
     """
 
     @staticmethod
-    def hash_file_and_check(filepath, expectedchecksum):
+    def hash_file_and_check(filepath: Union[str, pathlib.Path], expectedchecksum: str) -> bool:
         """
-        Hashes a file and checks for the expected checksum
+        Hashes a file and checks for the expected checksum.
+        Checksum is sha256 default implementation.
         """
         h = hashlib.sha256()
-        with open(filepath, 'rb') as f:
-            for chunk in iter(lambda: f.read(4096), b''):
+        with open(filepath, "rb") as f:
+            for chunk in iter(lambda: f.read(4096), b""):
                 h.update(chunk)
-        if expectedchecksum != h.hexdigest():
-            return False
-
-        return True
+        return expectedchecksum == h.hexdigest()
 
     @staticmethod
-    def load_json(filepath):
+    def load_json(filepath: Union[str, pathlib.Path]) -> Union[List[Any], Dict[str, Any]]:
+        if isinstance(filepath, str):
+            filepath: pathlib.Path = pathlib.Path(filepath)
+
         result = []
-        if not os.path.exists(filepath):
-            log.debug(f'Unable to load json from {filepath}')
-            return []
-        with io.open(filepath, 'r', encoding='utf-8-sig') as fp:
+        if not filepath.exists():
+            log.debug(f"Unable to load json from {filepath.absolute()}. Does not exist.")
+            return result
+        elif filepath.is_dir():
+            log.debug(f"Cannot load json at path {filepath.absolute()}. It is a directory")
+            return result
+
+        with open(filepath, "r", encoding="utf-8-sig") as fp:
             try:
                 result = json.load(fp)
                 if not result:
                     return []
-            except json.decoder.JSONDecodeError:
-                log.debug(f'JSONDecodeError while processing {filepath}')
+            except json.decoder.JSONDecodeError as err:
+                log.debug(f"JSONDecodeError while processing {filepath.absolute()} \n error: {str(err)}")
                 return []
         return result
 
     @staticmethod
-    def write_json(filepath, content):
-        with open(filepath, 'w') as outfile:
+    def write_json(filepath: Union[str, pathlib.Path], content: Dict[str, Any]) -> None:
+        with open(filepath, "w") as outfile:
             json.dump(content, outfile, cls=MagicJsonEncoder, indent=4)
 
     @staticmethod
-    def first_file(filepath, reverse=False):
-        results = glob.glob(filepath)
-        if reverse:
+    def first_file(filepath: Union[str, pathlib.Path], pattern: str, reverse: bool = False) -> Union[str, bool]:
+        if isinstance(filepath, str):
+            filepath = pathlib.Path(filepath)
+        results = [*filepath.glob(pattern)]
+        if not results:
+            return False
+        elif len(results) >= 1 and reverse:
             results.sort(reverse=True)
-        # log.info(filepath)
-        if results and len(results) >= 1:
-            return results[0]
-        return False
+        return str(results[0].absolute())
 
     @staticmethod
-    def folders_in_folder(filepath):
+    def folders_in_folder(filepath: str) -> List[str]:
         return [f for f in os.listdir(filepath) if os.path.isdir(os.path.join(filepath, f))]
 
     @staticmethod
-    def files_in_folder(filepath):
+    def files_in_folder(filepath: str) -> List[str]:
         return [f for f in os.listdir(filepath) if os.path.isfile(os.path.join(filepath, f))]
 
     @staticmethod
-    def seconds_to_human_time(seconds):
+    def seconds_to_human_time(seconds: int) -> str:
         return str(datetime.timedelta(seconds=seconds))
 
     @staticmethod
-    def from_json_datetime(jsondate):
-        datetime.datetime.strptime(jsondate, '%Y-%m-%dT%H:%M:%S.%fZ')
+    def from_json_datetime(jsondate: str) -> datetime.datetime:
+        return datetime.datetime.strptime(jsondate, "%Y-%m-%dT%H:%M:%S.%fZ")
 
     @staticmethod
-    def validate_platform(platform):
-        if platform in PLATFORMS:
-            return True
-        else:
-            return False
+    def validate_platform(platform: str) -> bool:
+        return platform in PLATFORMS
 
     @staticmethod
-    def validate_architecture(arch):
-        if arch in ARCHITECTURES:
-            return True
-        else:
-            return False
+    def validate_architecture(arch: str) -> bool:
+        return arch in ARCHITECTURES
 
     @staticmethod
-    def validate_buildtype(buildtype):
-        if buildtype in BUILDTYPES:
-            return True
-        else:
-            return False
+    def validate_buildtype(buildtype: str) -> bool:
+        return buildtype in BUILDTYPES
 
     @staticmethod
-    def validate_quality(quality):
-        if quality in QUALITIES:
-            return True
-        else:
-            return False
+    def validate_quality(quality: str) -> bool:
+        return quality in QUALITIES