Issue 162 | Repository: client integration

client/quantum_serverless/core/constrants.py

@@ -21,3 +21,9 @@
 OT_SPAN_DEFAULT_NAME = "entrypoint"
 OT_ATTRIBUTE_PREFIX = "qs"
 OT_LABEL_CALL_LOCATION = "qs.location"
+
+
+# repository
+REPO_HOST_KEY = "REPO_HOST_KEY"
+REPO_PORT_KEY = "REPO_PORT_KEY"
+REPO_PREFIX_KEY = "REPO_PREFIX_KEY"

client/quantum_serverless/core/program.py

@@ -26,17 +26,31 @@
 
     Program
 """
+import dataclasses
+import json
+import logging
+import os
+import tarfile
 from abc import ABC
-from typing import Optional, Dict, List
 from dataclasses import dataclass
+from pathlib import Path
+from typing import Optional, Dict, List, Any
+
+import requests
+
+from quantum_serverless.core.constrants import (
+    REPO_HOST_KEY,
+    REPO_PORT_KEY,
+)
+from quantum_serverless.exception import QuantumServerlessException
 
 
 @dataclass
 class Program:  # pylint: disable=too-many-instance-attributes
     """Serverless programs.
 
     Args:
-        name: program name
+        title: program name
         entrypoint: is a script that will be executed as a job
             ex: job.py
         arguments: arguments for entrypoint script
@@ -47,14 +61,21 @@ class Program:  # pylint: disable=too-many-instance-attributes
         version: version of a program
     """
 
-    name: str
+    title: str
     entrypoint: str
     working_dir: str = "./"
     arguments: Optional[Dict[str, str]] = None
     env_vars: Optional[Dict[str, str]] = None
     dependencies: Optional[List[str]] = None
     description: Optional[str] = None
     version: Optional[str] = None
+    tags: Optional[List[str]] = None
+
+    @classmethod
+    def from_json(cls, data: Dict[str, Any]):
+        """Reconstructs Program from dictionary."""
+        field_names = set(f.name for f in dataclasses.fields(Program))
+        return Program(**{k: v for k, v in data.items() if k in field_names})
 
 
 class ProgramStorage(ABC):
@@ -71,22 +92,130 @@ def save_program(self, program: Program) -> bool:
         """
         raise NotImplementedError
 
-    def get_programs(self) -> List[str]:
+    def get_programs(self, **kwargs) -> List[str]:
         """Returns list of available programs to get.
 
+        Args:
+            kwargs: filtering criteria
+
         Returns:
             List of names of programs
         """
         raise NotImplementedError
 
-    def get_program(self, name: str, **kwargs) -> Program:
+    def get_program(self, program: str, **kwargs) -> Optional[Program]:
         """Returns program by name of other query criterieas.
 
         Args:
-            name: name of program
+            program: name of the program
             **kwargs: other args
 
         Returns:
             Program
         """
         raise NotImplementedError
+
+
+class ProgramRepository(ProgramStorage):
+    """ProgramRepository."""
+
+    def __init__(
+        self,
+        host: Optional[str] = None,
+        port: Optional[int] = None,
+        token: Optional[str] = None,
+        root: Optional[str] = None,
+    ):
+        """Program repository implementation.
+
+        Args:
+            host: host of backend
+            port: port of backend
+            token: authentication token
+            root: path to directory where program files will be stored
+        """
+        self.root = root or os.path.dirname(os.path.abspath(__file__))
+        self._host = host or os.environ.get(REPO_HOST_KEY, "http://localhost")
+        self._port = port or os.environ.get(REPO_PORT_KEY, 80)
+        self._token = token
+        self._base_url = f"{self._host}:{self._port}/v1/api/nested-programs/"
+
+    def save_program(self, program: Program) -> bool:
+        raise NotImplementedError("Not implemented yet.")
+
+    def get_programs(self, **kwargs) -> List[str]:
+        result = []
+        response = requests.get(url=self._base_url, params=kwargs, timeout=10)
+        if response.ok:
+            response_data = json.loads(response.text)
+            result = [entry.get("title") for entry in response_data.get("results", [])]
+        return result
+
+    def get_program(self, program: str, **kwargs) -> Optional[Program]:
+        result = None
+        response = requests.get(
+            url=f"{self._base_url}",
+            params={"name": program},
+            allow_redirects=True,
+            timeout=10,
+        )
+        if response.ok:
+            response_data = json.loads(response.text)
+            results = response_data.get("results", [])
+            if len(results) > 0:
+                artifact = results[0].get("artifact")
+                result = Program.from_json(results[0])
+                result.working_dir = download_and_unpack_artifact(
+                    artifact_url=artifact, program=result, root=self.root
+                )
+            else:
+                logging.warning("No entries were found for your request.")
+        return result
+
+
+def download_and_unpack_artifact(
+    artifact_url: str,
+    program: Program,
+    root: str,
+    headers: Optional[Dict[str, Any]] = None,
+) -> str:
+    """Downloads and extract artifact files into destination folder.
+
+    Args:
+        artifact_url: url to get artifact from
+        program: program object artifact belongs to
+        root: root of programs folder a.k.a unpack destination
+        headers: optional headers needed for download requests
+
+    Returns:
+        workdir for program
+    """
+    program_folder_path = os.path.join(root, program.title)
+    artifact_file_name = "artifact"
+    tarfile_path = os.path.join(program_folder_path, artifact_file_name)
+
+    # check if program already exist on the disc
+    if os.path.exists(program_folder_path):
+        logging.warning("Program folder already exist. Will be overwritten.")
+
+    # download file
+    response = requests.get(artifact_url, stream=True, headers=headers, timeout=100)
+    if not response.ok:
+        raise QuantumServerlessException(
+            f"Error during fetch of [{artifact_url}] file."
+        )
+
+    Path(program_folder_path).mkdir(parents=True, exist_ok=True)
+
+    with open(tarfile_path, "wb") as file:
+        for data in response.iter_content():
+            file.write(data)
+
+    # unpack tarfile
+    with tarfile.open(tarfile_path, "r") as file_obj:
+        file_obj.extractall(program_folder_path)
+
+    # delete artifact
+    if os.path.exists(tarfile_path):
+        os.remove(tarfile_path)
+    return program_folder_path

client/quantum_serverless/core/provider.py

@@ -245,7 +245,7 @@ def run_program(self, program: Program) -> Job:
         entrypoint = f"python {program.entrypoint} {arguments}"
 
         # set program name so OT can use it as parent span name
-        env_vars = {**(program.env_vars or {}), **{OT_PROGRAM_NAME: program.name}}
+        env_vars = {**(program.env_vars or {}), **{OT_PROGRAM_NAME: program.title}}
 
         job_id = job_client.submit_job(
             entrypoint=entrypoint,

client/tests/core/test_program.py

@@ -41,7 +41,7 @@ def test_program():
         wait_for_job_client(serverless)
 
         program = Program(
-            name="simple_job",
+            title="simple_job",
             entrypoint="job.py",
             working_dir=resources_path,
             description="description",

client/tests/core/test_program_repository.py

@@ -0,0 +1,27 @@
+"""Test program repository."""
+import os.path
+from unittest import TestCase
+
+from quantum_serverless.core.program import ProgramRepository
+
+
+class TestRepository(TestCase):
+    """Tests for program."""
+
+    def test_repository_get_programs(self):
+        """Tests program repository."""
+
+        repository = ProgramRepository(host="http://127.0.0.1", port=8000)
+        print(repository.get_programs())
+
+    def test_repository_get_program(self):
+        """Tests single program fetch."""
+        repository = ProgramRepository(
+            host="http://127.0.0.1",
+            port=8000,
+            root=os.path.join(
+                os.path.dirname(os.path.abspath(__file__)), "..", "resources"
+            ),
+        )
+        program = repository.get_program("hello_program")
+        print(program)

repository/api/views.py

@@ -7,6 +7,9 @@
 
 from rest_framework import viewsets
 from rest_framework import permissions
+from rest_framework.decorators import action
+from rest_framework.response import Response
+
 from .models import NestedProgram
 
 
@@ -19,3 +22,13 @@ class NestedProgramViewSet(viewsets.ModelViewSet):  # pylint: disable=too-many-a
 
     queryset = NestedProgram.objects.all().order_by("created")
     permission_classes = [permissions.IsAuthenticatedOrReadOnly]
+
+    def get_queryset(self):
+        query_params = self.request.query_params
+        queryset = NestedProgram.objects.all().order_by("created")
+
+        # if name is specified in query parameters
+        name = query_params.get('name', None)
+        if name:
+            queryset = queryset.filter(title__exact=name)
+        return queryset