From 3441c3813334c20e224b84c45a9616f4fc6b01d1 Mon Sep 17 00:00:00 2001 From: Rathish Cholarajan Date: Wed, 8 Dec 2021 08:08:17 -0500 Subject: [PATCH] Move Program and ProgramJob REST adapters to separate files (#50) * Move Program and ProgramJob REST adapters to separate files * Fix lint --- qiskit_ibm_runtime/api/rest/program.py | 112 ++++++++++++++ qiskit_ibm_runtime/api/rest/program_job.py | 84 +++++++++++ qiskit_ibm_runtime/api/rest/runtime.py | 165 +-------------------- 3 files changed, 199 insertions(+), 162 deletions(-) create mode 100644 qiskit_ibm_runtime/api/rest/program.py create mode 100644 qiskit_ibm_runtime/api/rest/program_job.py diff --git a/qiskit_ibm_runtime/api/rest/program.py b/qiskit_ibm_runtime/api/rest/program.py new file mode 100644 index 000000000..f3c2374db --- /dev/null +++ b/qiskit_ibm_runtime/api/rest/program.py @@ -0,0 +1,112 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2021. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""Program REST adapter.""" + +from typing import Dict, Any, Optional +from concurrent import futures + +from .base import RestAdapterBase +from ..session import RetrySession + + +class Program(RestAdapterBase): + """Rest adapter for program related endpoints.""" + + URL_MAP = { + "self": "", + "data": "/data", + "run": "/jobs", + "private": "/private", + "public": "/public", + } + + _executor = futures.ThreadPoolExecutor() + + def __init__( + self, session: RetrySession, program_id: str, url_prefix: str = "" + ) -> None: + """Job constructor. + + Args: + session: Session to be used in the adapter. + program_id: ID of the runtime program. + url_prefix: Prefix to use in the URL. + """ + super().__init__(session, "{}/programs/{}".format(url_prefix, program_id)) + + def get(self) -> Dict[str, Any]: + """Return program information. + + Returns: + JSON response. + """ + url = self.get_url("self") + return self.session.get(url).json() + + def make_public(self) -> None: + """Sets a runtime program's visibility to public.""" + url = self.get_url("public") + self.session.put(url) + + def make_private(self) -> None: + """Sets a runtime program's visibility to private.""" + url = self.get_url("private") + self.session.put(url) + + def delete(self) -> None: + """Delete this program. + + Returns: + JSON response. + """ + url = self.get_url("self") + self.session.delete(url) + + def update_data(self, program_data: str) -> None: + """Update program data. + + Args: + program_data: Program data (base64 encoded). + """ + url = self.get_url("data") + self.session.put( + url, data=program_data, headers={"Content-Type": "application/octet-stream"} + ) + + def update_metadata( + self, + name: str = None, + description: str = None, + max_execution_time: int = None, + spec: Optional[Dict] = None, + ) -> None: + """Update program metadata. + + Args: + name: Name of the program. + description: Program description. + max_execution_time: Maximum execution time. + spec: Backend requirements, parameters, interim results, return values, etc. + """ + url = self.get_url("self") + payload: Dict = {} + if name: + payload["name"] = name + if description: + payload["description"] = description + if max_execution_time: + payload["cost"] = max_execution_time + if spec: + payload["spec"] = spec + + self.session.patch(url, json=payload) diff --git a/qiskit_ibm_runtime/api/rest/program_job.py b/qiskit_ibm_runtime/api/rest/program_job.py new file mode 100644 index 000000000..0c44149a4 --- /dev/null +++ b/qiskit_ibm_runtime/api/rest/program_job.py @@ -0,0 +1,84 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2021. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""Program Job REST adapter.""" + +from typing import Dict + +from .base import RestAdapterBase +from ..session import RetrySession + + +class ProgramJob(RestAdapterBase): + """Rest adapter for program job related endpoints.""" + + URL_MAP = { + "self": "", + "results": "/results", + "cancel": "/cancel", + "logs": "/logs", + "interim_results": "/interim_results", + } + + def __init__( + self, session: RetrySession, job_id: str, url_prefix: str = "" + ) -> None: + """ProgramJob constructor. + + Args: + session: Session to be used in the adapter. + job_id: ID of the program job. + url_prefix: Prefix to use in the URL. + """ + super().__init__(session, "{}/jobs/{}".format(url_prefix, job_id)) + + def get(self) -> Dict: + """Return program job information. + + Returns: + JSON response. + """ + return self.session.get(self.get_url("self")).json() + + def delete(self) -> None: + """Delete program job.""" + self.session.delete(self.get_url("self")) + + def interim_results(self) -> str: + """Return program job interim results. + + Returns: + Interim results. + """ + response = self.session.get(self.get_url("interim_results")) + return response.text + + def results(self) -> str: + """Return program job results. + + Returns: + Job results. + """ + response = self.session.get(self.get_url("results")) + return response.text + + def cancel(self) -> None: + """Cancel the job.""" + self.session.post(self.get_url("cancel")) + + def logs(self) -> str: + """Retrieve job logs. + + Returns: + Job logs. + """ + return self.session.get(self.get_url("logs")).text diff --git a/qiskit_ibm_runtime/api/rest/runtime.py b/qiskit_ibm_runtime/api/rest/runtime.py index f77c07769..b496aa7d0 100644 --- a/qiskit_ibm_runtime/api/rest/runtime.py +++ b/qiskit_ibm_runtime/api/rest/runtime.py @@ -13,12 +13,12 @@ """Runtime REST adapter.""" import logging -from typing import Dict, List, Any, Union, Optional +from typing import Dict, Any, Union, Optional import json -from concurrent import futures from .base import RestAdapterBase -from ..session import RetrySession +from .program import Program +from .program_job import ProgramJob from ...utils import RuntimeEncoder logger = logging.getLogger(__name__) @@ -184,162 +184,3 @@ def logout(self) -> None: """Clear authorization cache.""" url = self.get_url("logout") self.session.post(url) - - -class Program(RestAdapterBase): - """Rest adapter for program related endpoints.""" - - URL_MAP = { - "self": "", - "data": "/data", - "run": "/jobs", - "private": "/private", - "public": "/public", - } - - _executor = futures.ThreadPoolExecutor() - - def __init__( - self, session: RetrySession, program_id: str, url_prefix: str = "" - ) -> None: - """Job constructor. - - Args: - session: Session to be used in the adapter. - program_id: ID of the runtime program. - url_prefix: Prefix to use in the URL. - """ - super().__init__(session, "{}/programs/{}".format(url_prefix, program_id)) - - def get(self) -> Dict[str, Any]: - """Return program information. - - Returns: - JSON response. - """ - url = self.get_url("self") - return self.session.get(url).json() - - def make_public(self) -> None: - """Sets a runtime program's visibility to public.""" - url = self.get_url("public") - self.session.put(url) - - def make_private(self) -> None: - """Sets a runtime program's visibility to private.""" - url = self.get_url("private") - self.session.put(url) - - def delete(self) -> None: - """Delete this program. - - Returns: - JSON response. - """ - url = self.get_url("self") - self.session.delete(url) - - def update_data(self, program_data: str) -> None: - """Update program data. - - Args: - program_data: Program data (base64 encoded). - """ - url = self.get_url("data") - self.session.put( - url, data=program_data, headers={"Content-Type": "application/octet-stream"} - ) - - def update_metadata( - self, - name: str = None, - description: str = None, - max_execution_time: int = None, - spec: Optional[Dict] = None, - ) -> None: - """Update program metadata. - - Args: - name: Name of the program. - description: Program description. - max_execution_time: Maximum execution time. - spec: Backend requirements, parameters, interim results, return values, etc. - """ - url = self.get_url("self") - payload: Dict = {} - if name: - payload["name"] = name - if description: - payload["description"] = description - if max_execution_time: - payload["cost"] = max_execution_time - if spec: - payload["spec"] = spec - - self.session.patch(url, json=payload) - - -class ProgramJob(RestAdapterBase): - """Rest adapter for program job related endpoints.""" - - URL_MAP = { - "self": "", - "results": "/results", - "cancel": "/cancel", - "logs": "/logs", - "interim_results": "/interim_results", - } - - def __init__( - self, session: RetrySession, job_id: str, url_prefix: str = "" - ) -> None: - """ProgramJob constructor. - - Args: - session: Session to be used in the adapter. - job_id: ID of the program job. - url_prefix: Prefix to use in the URL. - """ - super().__init__(session, "{}/jobs/{}".format(url_prefix, job_id)) - - def get(self) -> Dict: - """Return program job information. - - Returns: - JSON response. - """ - return self.session.get(self.get_url("self")).json() - - def delete(self) -> None: - """Delete program job.""" - self.session.delete(self.get_url("self")) - - def interim_results(self) -> str: - """Return program job interim results. - - Returns: - Interim results. - """ - response = self.session.get(self.get_url("interim_results")) - return response.text - - def results(self) -> str: - """Return program job results. - - Returns: - Job results. - """ - response = self.session.get(self.get_url("results")) - return response.text - - def cancel(self) -> None: - """Cancel the job.""" - self.session.post(self.get_url("cancel")) - - def logs(self) -> str: - """Retrieve job logs. - - Returns: - Job logs. - """ - return self.session.get(self.get_url("logs")).text