Enhancement: customers will use the rest api for the project entity.

README.md

@@ -274,3 +274,28 @@ response = rest_client.post(f"orgs/{snyk_org}/invites", body={"email": "some.bod
 ```
 
 For backwards compatibility the get_rest_pages method has an alternative name of get_v3_pages to not break code already rewritten replatformed to the 0.9.0 pysnyk module.
+
+## Migrating from old project to new project entity
+
+The rest API introduces a new representation for the project class, which offers more information. There are some changes necessary to keep the same functionality
+of the application:
+- The old project attributes can be accessed using the attributes class. The below example 
+```python
+project.name -> project.attributes.name
+project.created -> project.attributes.created
+project.origin -> project.attributes.origin
+project.type -> project.attributes.type
+project.readOnly -> project.attributes.read_only
+project.attributes.criticality -> project.attributes.business_criticality
+project.totalDependencies -> project.meta.latest_dependency_total.total
+project.issueCountsBySeverity -> project.meta.latest_issue_counts
+project.lastTestedDate -> project.meta.cli_monitored_at
+project.importingUser.id -> project.relationships.importer.data.id
+project.isMonitored -> project.attributes.status
+project.targetReference -> project.attributes.target_reference
+```
+
+To update an existing project, the update method from `ProjectManager` can be used, as in the following example:
+```python
+client.organizations.get(org_id).projects.update(project_id, tags=[{"key":"added_tag_key2", "value":"added_tag_value2"}], environment=[], business_criticality=["critical","low","medium"], lifecycle=["development","production"], test_frequency="daily")
+```

examples/api-demo-1-list-projects.py

@@ -19,8 +19,8 @@ def parse_command_line_args():
 
 client = SnykClient(token=snyk_token)
 for proj in client.organizations.get(org_id).projects.all():
-    print("\nProject name: %s" % proj.name)
+    print("\nProject name: %s" % proj.attributes.name)
     print("  Issues Found:")
-    print("      High  : %s" % proj.issueCountsBySeverity.high)
-    print("      Medium: %s" % proj.issueCountsBySeverity.medium)
-    print("      Low   : %s" % proj.issueCountsBySeverity.low)
+    print("      High  : %s" % proj.meta.latest_issue_counts.high)
+    print("      Medium: %s" % proj.meta.latest_issue_counts.medium)
+    print("      Low   : %s" % proj.meta.latest_issue_counts.low)

examples/api-demo-10-project-deps-licenses-report.py

@@ -100,8 +100,8 @@ def get_flat_dependencies(dep_list):
 client = SnykClient(snyk_token)
 projects = client.organizations.get(org_id).projects.all()
 for proj in projects:
-    if proj.origin in allowed_origins:
-        all_projects_list.append({"project_id": proj.id, "project_name": proj.name})
+    if proj.attributes.origin in allowed_origins:
+        all_projects_list.append({"project_id": proj.id, "project_name": proj.attributes.name})
 
 project_trees = []
 flattened_project_dependencies_lists = []
@@ -112,6 +112,7 @@ def get_flat_dependencies(dep_list):
     next_project_id = next_project["project_id"]
 
     if args.projectId == "all" or next_project_id == args.projectId:
+        # Deprecated, check api-demo-2c-list-issues-aggregated
         next_project_tree = ProjectDependenciesReport.get_project_tree(
             snyk_token, org_id, next_project_id
         )

examples/api-demo-11-update-github-checks.py

@@ -37,9 +37,9 @@ def parse_command_line_args():
 projects = client.organizations.get(org_id).projects.all()
 
 github_projects = [
-    {"id": p.id, "name": p.name}
+    {"id": p.id, "name": p.attributes.name}
     for p in projects
-    if p.origin == "github"
+    if p.attributes.origin == "github"
 ]
 
 def get_project_by_id(projects, project_id):

examples/api-demo-2-list-issues.py

@@ -57,6 +57,7 @@ def output_excel(vulns, output_path):
 
 
 client = SnykClient(snyk_token)
+#Deprecated, check api-demo-2c-list-issues-aggregated
 issue_set = client.organizations.get(org_id).projects.get(project_id).issueset.all()
 
 lst_output = []

examples/api-demo-8-delete-projects-by-name.py

@@ -42,8 +42,8 @@ def parse_command_line_args():
 
 client = SnykClient(snyk_token)
 for project in client.organizations.get(org_id).projects.all():
-    if project_name == project.name and (
-        project.origin == project_origin or not project_origin
+    if project_name == project.attributes.name and (
+        project.attributes.origin == project_origin or not project_origin
     ):
         if project.delete():
             print("Project ID %s deleted" % project.id)

examples/api-demo-9-list-ignores.py

@@ -23,7 +23,7 @@ def parse_command_line_args():
 
 for proj in projects:
     project_id = proj.id
-    project_name = proj.name
+    project_name = proj.attributes.name
     ignores = proj.ignores.all()
 
     if len(ignores) > 0:

examples/api-demo-9b-ignore-vulns-by-id.py

@@ -66,12 +66,12 @@ def parse_command_line_args():
 # API call to collect every project in all of a customers orgs
 http = urllib3.PoolManager()
 for proj in client.organizations.get(org_id).projects.all():
-    print("\nProject name: %s" % proj.name)
+    print("\nProject name: %s" % proj.attributes.name)
     print("  Issues Found:")
-    print("      High  : %s" % proj.issueCountsBySeverity.high)
-    print("      Medium: %s" % proj.issueCountsBySeverity.medium)
-    print("      Low   : %s" % proj.issueCountsBySeverity.low)
-
+    print("      High  : %s" % proj.meta.latest_issue_counts.high)
+    print("      Medium: %s" % proj.meta.latest_issue_counts.medium)
+    print("      Low   : %s" % proj.meta.latest_issue_counts.low)
+    # Deprecated, check api-demo-2c-list-issues-aggregated
     url = "org/" + org_id + "/project/" + proj.id + "/issues"
 
     print(url)

examples/api-demo-9c-bulk-ignore-vulns-by-issueIdList.py

@@ -57,11 +57,12 @@ def parse_command_line_args():
 # API call to collect every project in all of a customers orgs
 
 for proj in client.organizations.get(org_id).projects.all():
-    print("\nProject name: %s" % proj.name)
+    print("\nProject name: %s" % proj.attributes.name)
     print("  Issues Found:")
-    print("      High  : %s" % proj.issueCountsBySeverity.high)
-    print("      Medium: %s" % proj.issueCountsBySeverity.medium)
-    print("      Low   : %s" % proj.issueCountsBySeverity.low)
+    print("      High  : %s" % proj.meta.latest_issue_counts.high)
+    print("      Medium: %s" % proj.meta.latest_issue_counts.medium)
+    print("      Low   : %s" % proj.meta.latest_issue_counts.low)
+    #Deprecated, check api-demo-2c-list-issues-aggregated
     url = "org/" + org_id + "/project/" + proj.id + "/issues"
     print(url)
     # API call to grab all of the issue

snyk/client.py

@@ -1,6 +1,8 @@
+import copy
 import logging
+import re
 import urllib.parse
-from typing import Any, List, Optional
+from typing import Any, Dict, List, Optional, Pattern
 from urllib.parse import parse_qs, urlparse
 
 import requests
@@ -40,13 +42,16 @@ def __init__(
             "Authorization": "token %s" % self.api_token,
             "User-Agent": user_agent,
         }
-        self.api_post_headers = self.api_headers
+        self.api_post_headers = dict(self.api_headers)
         self.api_post_headers["Content-Type"] = "application/json"
+        self.api_patch_headers = dict(self.api_headers)
+        self.api_patch_headers["Content-Type"] = "application/vnd.api+json"
         self.tries = tries
         self.backoff = backoff
         self.delay = delay
         self.verify = verify
         self.version = version
+        self.__latest_version = "2024-06-21"
 
         # Ensure we don't have a trailing /
         if self.api_url[-1] == "/":
@@ -82,14 +87,31 @@ def request(
             raise SnykHTTPError(resp)
         return resp
 
-    def post(self, path: str, body: Any, headers: dict = {}) -> requests.Response:
-        url = f"{self.api_url}/{path}"
+    def post(
+        self,
+        path: str,
+        body: Dict[str, Any],
+        headers: Dict = {},
+        params: Dict[str, Any] = {},
+        use_rest: bool = False,
+    ) -> requests.Response:
+        url = f"{self.rest_api_url if use_rest else self.api_url}/{path}"
         logger.debug(f"POST: {url}")
 
+        request_headers = {**self.api_post_headers, **headers}
+        if use_rest:
+            if "version" not in params:
+                params["version"] = self.version or self.__latest_version
+            request_headers["Content-Type"] = "application/vnd.api+json"
+
         resp = retry_call(
             self.request,
             fargs=[requests.post, url],
-            fkwargs={"json": body, "headers": {**self.api_post_headers, **headers}},
+            fkwargs={
+                "json": body,
+                "headers": request_headers,
+                "params": params,
+            },
             tries=self.tries,
             delay=self.delay,
             backoff=self.backoff,
@@ -103,14 +125,68 @@ def post(self, path: str, body: Any, headers: dict = {}) -> requests.Response:
 
         return resp
 
-    def put(self, path: str, body: Any, headers: dict = {}) -> requests.Response:
-        url = "%s/%s" % (self.api_url, path)
+    def patch(
+        self,
+        path: str,
+        body: Dict[str, Any],
+        headers: Dict[str, str] = {},
+        params: Dict[str, Any] = {},
+    ) -> requests.Response:
+        url = f"{self.rest_api_url}/{path}"
+        logger.debug(f"PATCH: {url}")
+
+        if "version" not in params:
+            params["version"] = self.version or self.__latest_version
+
+        resp = retry_call(
+            self.request,
+            fargs=[requests.patch, url],
+            fkwargs={
+                "json": body,
+                "headers": {**self.api_patch_headers, **headers},
+                "params": params,
+            },
+            tries=self.tries,
+            delay=self.delay,
+            backoff=self.backoff,
+            logger=logger,
+        )
+
+        if not resp.ok:
+            logger.error(resp.text)
+            raise SnykHTTPError(resp)
+
+        return resp
+
+    def put(
+        self,
+        path: str,
+        body: Any,
+        headers: Dict = {},
+        params: Dict[str, Any] = {},
+        use_rest: bool = False,
+    ) -> requests.Response:
+        url = "%s/%s" % (
+            self.rest_api_url if use_rest else self.api_url,
+            path,
+        )
         logger.debug("PUT: %s" % url)
 
+        request_headers = {**self.api_post_headers, **headers}
+
+        if use_rest:
+            if "version" not in params:
+                params["version"] = self.version or self.__latest_version
+            request_headers["Content-Type"] = "application/vnd.api+json"
+
         resp = retry_call(
             self.request,
             fargs=[requests.put, url],
-            fkwargs={"json": body, "headers": {**self.api_post_headers, **headers}},
+            fkwargs={
+                "json": body,
+                "headers": request_headers,
+                "params": params,
+            },
             tries=self.tries,
             delay=self.delay,
             backoff=self.backoff,
@@ -125,7 +201,7 @@ def put(self, path: str, body: Any, headers: dict = {}) -> requests.Response:
     def get(
         self,
         path: str,
-        params: dict = None,
+        params: Dict = None,
         version: str = None,
         exclude_version: bool = False,
         exclude_params: bool = False,
@@ -163,10 +239,12 @@ def get(
 
             # Python Bools are True/False, JS Bools are true/false
             # Snyk REST API is strictly case sensitive at the moment
-
+            # List elements are separated using comma
             for k, v in params.items():
                 if isinstance(v, bool):
                     params[k] = str(v).lower()
+                elif isinstance(v, list):
+                    params[k] = ",".join(v)
 
             # the limit is returned in the url, and if two limits are passed
             # the API interprets as an array and throws an error
@@ -196,14 +274,19 @@ def get(
 
         return resp
 
-    def delete(self, path: str) -> requests.Response:
-        url = f"{self.api_url}/{path}"
+    def delete(self, path: str, use_rest: bool = False) -> requests.Response:
+        url = f"{self.rest_api_url if use_rest else self.api_url}/{path}"
+
+        params = {}
+        if use_rest:
+            params["version"] = self.version or self.__latest_version
+
         logger.debug(f"DELETE: {url}")
 
         resp = retry_call(
             self.request,
             fargs=[requests.delete, url],
-            fkwargs={"headers": self.api_headers},
+            fkwargs={"headers": self.api_headers, "params": params},
             tries=self.tries,
             delay=self.delay,
             backoff=self.backoff,
@@ -215,7 +298,7 @@ def delete(self, path: str) -> requests.Response:
 
         return resp
 
-    def get_rest_pages(self, path: str, params: dict = {}) -> List:
+    def get_rest_pages(self, path: str, params: Dict = {}) -> List:
         """
         Helper function to collect paginated responses from the rest API into a single
         list.