Samples and views API

.github/workflows/pytest-windows.yml

@@ -11,7 +11,7 @@ jobs:
     runs-on: ${{ matrix.os }}
     strategy:
       matrix:
-        python-version: ["3.10"]
+        python-version: ["3.11"]
         os: [windows-latest]
 
     steps:

.github/workflows/pytest.yml

@@ -11,7 +11,7 @@ jobs:
     runs-on: ${{ matrix.os }}
     strategy:
       matrix:
-        python-version: ["3.8", "3.11"]
+        python-version: ["3.8", "3.12"]
         os: [ubuntu-20.04]
 
     steps:

MANIFEST.in

@@ -1,3 +1,4 @@
 include requirements/*
 include README.md
-include pephubclient/pephub_oauth/*
+include pephubclient/pephub_oauth/*
+include pephubclient/modules/*

Makefile

@@ -5,7 +5,7 @@ run-coverage:
 	coverage run -m pytest
 
 html-report:
-	coverage html
+	coverage html --omit="*/test*"
 
 open-coverage:
 	cd htmlcov && google-chrome index.html

pephubclient/__init__.py

@@ -1,5 +1,7 @@
 from pephubclient.pephubclient import PEPHubClient
 from pephubclient.helpers import is_registry_path, save_pep
+import logging
+import coloredlogs
 
 __app_name__ = "pephubclient"
 __version__ = "0.3.0"
@@ -14,3 +16,11 @@
     "is_registry_path",
     "save_pep",
 ]
+
+
+_LOGGER = logging.getLogger(__app_name__)
+coloredlogs.install(
+    logger=_LOGGER,
+    datefmt="%H:%M:%S",
+    fmt="[%(levelname)s] [%(asctime)s] %(message)s",
+)

pephubclient/constants.py

@@ -12,6 +12,12 @@
 PEPHUB_PEP_SEARCH_URL = f"{PEPHUB_BASE_URL}api/v1/namespaces/{{namespace}}/projects"
 PEPHUB_PUSH_URL = f"{PEPHUB_BASE_URL}api/v1/namespaces/{{namespace}}/projects/json"
 
+PEPHUB_SAMPLE_URL = f"{PEPHUB_BASE_URL}api/v1/projects/{{namespace}}/{{project}}/samples/{{sample_name}}"
+PEPHUB_VIEW_URL = (
+    f"{PEPHUB_BASE_URL}api/v1/projects/{{namespace}}/{{project}}/views/{{view_name}}"
+)
+PEPHUB_VIEW_SAMPLE_URL = f"{PEPHUB_BASE_URL}api/v1/projects/{{namespace}}/{{project}}/views/{{view_name}}/{{sample_name}}"
+
 
 class RegistryPath(BaseModel):
     protocol: Optional[str] = None
@@ -33,3 +39,10 @@ class ResponseStatusCodes(int, Enum):
     NOT_EXIST = 404
     CONFLICT = 409
     INTERNAL_ERROR = 500
+
+
+USER_DATA_FILE_NAME = "jwt.txt"
+HOME_PATH = os.getenv("HOME")
+if not HOME_PATH:
+    HOME_PATH = os.path.expanduser("~")
+PATH_TO_FILE_WITH_JWT = os.path.join(HOME_PATH, ".pephubclient/") + USER_DATA_FILE_NAME

pephubclient/files_manager.py

@@ -6,7 +6,6 @@
 import yaml
 import zipfile
 
-from pephubclient.constants import RegistryPath
 from pephubclient.exceptions import PEPExistsError
 
 

pephubclient/helpers.py

@@ -16,6 +16,7 @@
 
 import requests
 from requests.exceptions import ConnectionError
+from urllib.parse import urlencode
 
 from ubiquerg import parse_registry_path
 from pydantic import ValidationError
@@ -47,20 +48,50 @@ def send_request(
         )
 
     @staticmethod
-    def decode_response(response: requests.Response, encoding: str = "utf-8") -> str:
+    def decode_response(
+        response: requests.Response, encoding: str = "utf-8", output_json: bool = False
+    ) -> Union[str, dict]:
         """
         Decode the response from GitHub and pack the returned data into appropriate model.
 
         :param response: Response from GitHub.
         :param encoding: Response encoding [Default: utf-8]
+        :param output_json: If True, return response in json format
         :return: Response data as an instance of correct model.
         """
 
         try:
-            return response.content.decode(encoding)
+            if output_json:
+                return response.json()
+            else:
+                return response.content.decode(encoding)
         except json.JSONDecodeError as err:
             raise ResponseError(f"Error in response encoding format: {err}")
 
+    @staticmethod
+    def parse_query_param(pep_variables: dict) -> str:
+        """
+        Grab all the variables passed by user (if any) and parse them to match the format specified
+        by PEPhub API for query parameters.
+
+        :param pep_variables: dict of query parameters
+        :return: PEPHubClient variables transformed into string in correct format.
+        """
+        return "?" + urlencode(pep_variables)
+
+    @staticmethod
+    def parse_header(jwt_data: Optional[str] = None) -> dict:
+        """
+        Create Authorization header
+
+        :param jwt_data: jwt string
+        :return: Authorization dict
+        """
+        if jwt_data:
+            return {"Authorization": jwt_data}
+        else:
+            return {}
+
 
 class MessageHandler:
     """

pephubclient/models.py

@@ -1,5 +1,5 @@
 import datetime
-from typing import Optional, List
+from typing import Optional, List, Union
 
 from pydantic import BaseModel, Field, field_validator, ConfigDict
 from peppy.const import CONFIG_KEY, SUBSAMPLE_RAW_LIST_KEY, SAMPLE_RAW_DICT_KEY
@@ -43,6 +43,9 @@ class ProjectAnnotationModel(BaseModel):
     submission_date: datetime.datetime
     digest: str
     pep_schema: str
+    pop: bool = False
+    stars_number: Optional[int] = 0
+    forked_from: Optional[Union[str, None]] = None
 
 
 class SearchReturnModel(BaseModel):

pephubclient/modules/sample.py

@@ -0,0 +1,208 @@
+import logging
+
+from pephubclient.helpers import RequestManager
+from pephubclient.constants import PEPHUB_SAMPLE_URL, ResponseStatusCodes
+from pephubclient.exceptions import ResponseError
+
+_LOGGER = logging.getLogger("pephubclient")
+
+
+class PEPHubSample(RequestManager):
+    """
+    Class for managing samples in PEPhub and provides methods for
+        getting, creating, updating and removing samples.
+    This class is not related to peppy.Sample class.
+    """
+
+    def __init__(self, jwt_data: str = None):
+        """
+        :param jwt_data: jwt token for authorization
+        """
+
+        self.__jwt_data = jwt_data
+
+    def get(
+        self,
+        namespace: str,
+        name: str,
+        tag: str,
+        sample_name: str = None,
+    ) -> dict:
+        """
+        Get sample from project in PEPhub.
+
+        :param namespace: namespace of project
+        :param name: name of project
+        :param tag: tag of project
+        :param sample_name: sample name
+        :return: Sample object
+        """
+        url = self._build_sample_request_url(
+            namespace=namespace, name=name, sample_name=sample_name
+        )
+
+        url = url + self.parse_query_param(pep_variables={"tag": tag})
+
+        response = self.send_request(
+            method="GET", url=url, headers=self.parse_header(self.__jwt_data)
+        )
+        if response.status_code == ResponseStatusCodes.OK:
+            return self.decode_response(response, output_json=True)
+        if response.status_code == ResponseStatusCodes.NOT_EXIST:
+            raise ResponseError(
+                f"Sample does not exist. Project: '{namespace}/{name}:{tag}'. Sample_name: '{sample_name}'"
+            )
+        elif response.status_code == ResponseStatusCodes.INTERNAL_ERROR:
+            raise ResponseError("Internal server error. Unexpected return value.")
+        else:
+            raise ResponseError(
+                f"Unexpected return value. Error: {response.status_code}"
+            )
+
+    def create(
+        self,
+        namespace: str,
+        name: str,
+        tag: str,
+        sample_name: str,
+        sample_dict: dict,
+        overwrite: bool = False,
+    ) -> None:
+        """
+        Create sample in project in PEPhub.
+
+        :param namespace: namespace of project
+        :param name: name of project
+        :param tag: tag of project
+        :param sample_dict: sample dict
+        :param sample_name: sample name
+        :param overwrite: overwrite sample if it exists
+        :return: None
+        """
+        url = self._build_sample_request_url(
+            namespace=namespace,
+            name=name,
+            sample_name=sample_name,
+        )
+
+        url = url + self.parse_query_param(
+            pep_variables={"tag": tag, "overwrite": overwrite}
+        )
+
+        # add sample name to sample_dict if it is not there
+        if sample_name not in sample_dict.values():
+            sample_dict["sample_name"] = sample_name
+
+        response = self.send_request(
+            method="POST",
+            url=url,
+            headers=self.parse_header(self.__jwt_data),
+            json=sample_dict,
+        )
+        if response.status_code == ResponseStatusCodes.ACCEPTED:
+            _LOGGER.info(
+                f"Sample '{sample_name}' added to project '{namespace}/{name}:{tag}' successfully."
+            )
+            return None
+        elif response.status_code == ResponseStatusCodes.NOT_EXIST:
+            raise ResponseError(f"Project '{namespace}/{name}:{tag}' does not exist.")
+        elif response.status_code == ResponseStatusCodes.CONFLICT:
+            raise ResponseError(
+                f"Sample '{sample_name}' already exists. Set overwrite to True to overwrite sample."
+            )
+        else:
+            raise ResponseError(
+                f"Unexpected return value. Error: {response.status_code}"
+            )
+
+    def update(
+        self,
+        namespace: str,
+        name: str,
+        tag: str,
+        sample_name: str,
+        sample_dict: dict,
+    ):
+        """
+        Update sample in project in PEPhub.
+
+        :param namespace: namespace of project
+        :param name: name of project
+        :param tag: tag of project
+        :param sample_name: sample name
+        :param sample_dict: sample dict, that contain elements to update, or
+        :return: None
+        """
+
+        url = self._build_sample_request_url(
+            namespace=namespace, name=name, sample_name=sample_name
+        )
+
+        url = url + self.parse_query_param(pep_variables={"tag": tag})
+
+        response = self.send_request(
+            method="PATCH",
+            url=url,
+            headers=self.parse_header(self.__jwt_data),
+            json=sample_dict,
+        )
+        if response.status_code == ResponseStatusCodes.ACCEPTED:
+            _LOGGER.info(
+                f"Sample '{sample_name}' updated in project '{namespace}/{name}:{tag}' successfully."
+            )
+            return None
+        elif response.status_code == ResponseStatusCodes.NOT_EXIST:
+            raise ResponseError(
+                f"Sample '{sample_name}' or project {namespace}/{name}:{tag} does not exist. Error: {response.status_code}"
+            )
+        else:
+            raise ResponseError(
+                f"Unexpected return value. Error: {response.status_code}"
+            )
+
+    def remove(self, namespace: str, name: str, tag: str, sample_name: str):
+        """
+        Remove sample from project in PEPhub.
+
+        :param namespace: namespace of project
+        :param name: name of project
+        :param tag: tag of project
+        :param sample_name: sample name
+        :return: None
+        """
+        url = self._build_sample_request_url(
+            namespace=namespace, name=name, sample_name=sample_name
+        )
+
+        url = url + self.parse_query_param(pep_variables={"tag": tag})
+
+        response = self.send_request(
+            method="DELETE",
+            url=url,
+            headers=self.parse_header(self.__jwt_data),
+        )
+        if response.status_code == ResponseStatusCodes.ACCEPTED:
+            _LOGGER.info(
+                f"Sample '{sample_name}' removed from project '{namespace}/{name}:{tag}' successfully."
+            )
+            return None
+        elif response.status_code == ResponseStatusCodes.NOT_EXIST:
+            raise ResponseError(
+                f"Sample '{sample_name}' or project {namespace}/{name}:{tag} does not exist. Error: {response.status_code}"
+            )
+        else:
+            raise ResponseError(
+                f"Unexpected return value. Error: {response.status_code}"
+            )
+
+    @staticmethod
+    def _build_sample_request_url(namespace: str, name: str, sample_name: str) -> str:
+        """
+        Build url for sample request.
+
+        :param namespace: namespace where project will be uploaded
+        :return: url string
+        """
+        return PEPHUB_SAMPLE_URL.format(
+            namespace=namespace, project=name, sample_name=sample_name
+        )