Added license metadata endpoint

socketdev/__init__.py

@@ -17,6 +17,7 @@
 from socketdev.utils import Utils, IntegrationType, INTEGRATION_TYPES
 from socketdev.version import __version__
 from socketdev.labels import Labels
+from socketdev.licensemetadata import LicenseMetadata
 from socketdev.log import log
 
 
@@ -59,6 +60,7 @@ def __init__(self, token: str, timeout: int = 1200):
         self.triage = Triage(self.api)
         self.utils = Utils()
         self.labels = Labels(self.api)
+        self.licensemetadata = LicenseMetadata(self.api)
 
     @staticmethod
     def set_timeout(timeout: int):

socketdev/core/dedupe.py

@@ -1,5 +1,6 @@
 from collections import defaultdict
 from typing import Dict, List, Any
+from socketdev.log import log
 
 
 class Dedupe:
@@ -100,5 +101,5 @@ def consolidate_by_order(packages: List[Dict[str, Any]]) -> dict[int, list[dict[
                 pkg["batchIndex"] = batch_index
                 grouped[pkg["batchIndex"]].append(pkg)
         except Exception as error:
-            print(error)
+            log.error(error)
         return grouped

socketdev/export/__init__.py

@@ -47,7 +47,7 @@ def cdx_bom(
             # TODO: Add typed response when types are defined
 
         log.error(f"Error exporting CDX BOM: {response.status_code}")
-        print(response.text)
+        log.error(response.text)
         return {}
 
     def spdx_bom(
@@ -71,5 +71,5 @@ def spdx_bom(
             # TODO: Add typed response when types are defined
 
         log.error(f"Error exporting SPDX BOM: {response.status_code}")
-        print(response.text)
+        log.error(response.text)
         return {}

socketdev/labels/__init__.py

@@ -73,7 +73,7 @@ def post(self, org_slug: str, label_name: str) -> dict:
             return result
 
         error_message = response.json().get("error", {}).get("message", "Unknown error")
-        print(f"Failed to create repository label: {response.status_code}, message: {error_message}")
+        log.error(f"Failed to create repository label: {response.status_code}, message: {error_message}")
         return {}
 
     def get(self, org_slug: str, label_id: str) -> dict:
@@ -84,7 +84,7 @@ def get(self, org_slug: str, label_id: str) -> dict:
             return result
 
         error_message = response.json().get("error", {}).get("message", "Unknown error")
-        print(f"Failed to get repository label: {response.status_code}, message: {error_message}")
+        log.error(f"Failed to get repository label: {response.status_code}, message: {error_message}")
         return {}
 
     def delete(self, org_slug: str, label_id: str) -> dict:

socketdev/licensemetadata/__init__.py

@@ -0,0 +1,27 @@
+import json
+import logging
+import urllib.parse
+
+log = logging.getLogger("socketdev")
+
+
+class LicenseMetadata:
+    def __init__(self, api):
+        self.api = api
+
+    def post(self, licenses: list, params: dict = None) -> dict:
+        path = f"license-metadata"
+        if params:
+            query_args = urllib.parse.urlencode(params)
+            path += f"?{query_args}"
+        payload = json.dumps(licenses)
+        response = self.api.do_request(path=path, method="POST", payload=payload)
+
+        if response.status_code == 200:
+            result = response.json()
+            return result
+
+        error_message = response.json().get("error", {}).get("message", "Unknown error")
+        log.error(f"Failed to create license metadata: {response.status_code}, message: {error_message}")
+        return {}
+

socketdev/npm/__init__.py

@@ -15,7 +15,7 @@ def issues(self, package: str, version: str) -> list:
         if response.status_code == 200:
             return response.json()
         log.error(f"Error getting npm issues: {response.status_code}")
-        print(response.text)
+        log.error(response.text)
         return []
 
     def score(self, package: str, version: str) -> list:
@@ -24,5 +24,5 @@ def score(self, package: str, version: str) -> list:
         if response.status_code == 200:
             return response.json()
         log.error(f"Error getting npm score: {response.status_code}")
-        print(response.text)
+        log.error(response.text)
         return []

socketdev/openapi/__init__.py

@@ -15,5 +15,5 @@ def get(self) -> dict:
         if response.status_code == 200:
             return response.json()
         log.error(f"Error getting OpenAPI spec: {response.status_code}")
-        print(response.text)
+        log.error(response.text)
         return {}

socketdev/org/__init__.py

@@ -30,5 +30,5 @@ def get(self, use_types: bool = False) -> OrganizationsResponse:
                 return OrganizationsResponse(result)
             return result
         log.error(f"Error getting organizations: {response.status_code}")
-        print(response.text)
+        log.error(response.text)
         return {"organizations": {}}

socketdev/purl/__init__.py

@@ -37,5 +37,5 @@ def post(self, license: str = "false", components: list = None, **kwargs) -> lis
             return purl_deduped
 
         log.error(f"Error posting {components} to the Purl API: {response.status_code}")
-        print(response.text)
+        log.error(response.text)
         return []

socketdev/quota/__init__.py

@@ -15,5 +15,5 @@ def get(self) -> dict:
         if response.status_code == 200:
             return response.json()
         log.error(f"Error getting quota: {response.status_code}")
-        print(response.text)
+        log.error(response.text)
         return {}

socketdev/report/__init__.py

@@ -27,7 +27,7 @@ def list(self, from_time: int = None) -> dict:
         if response.status_code == 200:
             return response.json()
         log.error(f"Error listing reports: {response.status_code}")
-        print(response.text)
+        log.error(response.text)
         return {}
 
     def delete(self, report_id: str) -> bool:
@@ -36,7 +36,7 @@ def delete(self, report_id: str) -> bool:
         if response.status_code == 200:
             return True
         log.error(f"Error deleting report: {response.status_code}")
-        print(response.text)
+        log.error(response.text)
         return False
 
     def view(self, report_id) -> dict:
@@ -45,7 +45,7 @@ def view(self, report_id) -> dict:
         if response.status_code == 200:
             return response.json()
         log.error(f"Error viewing report: {response.status_code}")
-        print(response.text)
+        log.error(response.text)
         return {}
 
     def supported(self) -> dict:
@@ -54,7 +54,7 @@ def supported(self) -> dict:
         if response.status_code == 200:
             return response.json()
         log.error(f"Error getting supported reports: {response.status_code}")
-        print(response.text)
+        log.error(response.text)
         return {}
 
     def create(self, files: list) -> dict:
@@ -68,5 +68,5 @@ def create(self, files: list) -> dict:
         if response.status_code == 200:
             return response.json()
         log.error(f"Error creating report: {response.status_code}")
-        print(response.text)
+        log.error(response.text)
         return {}

socketdev/repos/__init__.py

@@ -108,7 +108,7 @@ def repo(self, org_slug: str, repo_name: str, use_types: bool = False) -> Union[
             return result
 
         error_message = response.json().get("error", {}).get("message", "Unknown error")
-        print(f"Failed to get repository: {response.status_code}, message: {error_message}")
+        log.error(f"Failed to get repository: {response.status_code}, message: {error_message}")
         if use_types:
             return GetRepoResponse.from_dict(
                 {"success": False, "status": response.status_code, "message": error_message}

socketdev/repositories/__init__.py

@@ -27,5 +27,5 @@ def list(self, use_types: bool = False) -> Union[dict, list[Repo]]:
             return result
 
         log.error(f"Error listing repositories: {response.status_code}")
-        print(response.text)
+        log.error(response.text)
         return []

socketdev/sbom/__init__.py

@@ -32,7 +32,7 @@ def view(self, report_id: str) -> dict[str, dict]:
                 sbom_dict[val["id"]] = val
         else:
             log.error(f"Error viewing SBOM: {response.status_code}")
-            print(response.text)
+            log.error(response.text)
             sbom_dict = {}
         return sbom_dict
 
@@ -48,7 +48,7 @@ def create_packages_dict(self, sbom: dict[str, dict]) -> dict[str, Package]:
             item = sbom[package_id]
             package = Package(**item)
             if package.id in packages:
-                print(f"Duplicate package_id: {package_id}")
+                log.error(f"Duplicate package_id: {package_id}")
             else:
                 packages[package.id] = package
                 for top_id in package.topLevelAncestors:

socketdev/settings/__init__.py

@@ -91,7 +91,7 @@ def get(
             return rules
 
         error_message = response.json().get("error", {}).get("message", "Unknown error")
-        print(f"Failed to get security policy: {response.status_code}, message: {error_message}")
+        log.error(f"Failed to get security policy: {response.status_code}, message: {error_message}")
         if use_types:
             return OrgSecurityPolicyResponse.from_dict(
                 {"securityPolicyRules": {}, "success": False, "status": response.status_code, "message": error_message}

socketdev/version.py

@@ -1 +1 @@
-__version__ = "2.1.0"
+__version__ = "2.1.3"