Refactor: move post/inject methods to InjectAPI class based on CalCatAPIClient

src/cal_tools/calcat_interface2.py

@@ -30,34 +30,6 @@ class ModuleNameError(KeyError):
 class CalCatAPIError(requests.HTTPError):
     """Used when the response includes error details as JSON"""
 
-
-class InjectAPIError(CalCatAPIError):
-    ...
-
-
-class CCVAlreadyInjectedError(InjectAPIError):
-    ...
-
-
-CALIBRATION_CONSTANT_VERSIONS = "calibration_constant_versions"
-
-def _failed_response(resp):
-    # TODO: Add more errors if needed
-    if CALIBRATION_CONSTANT_VERSIONS in resp.url.lstrip("/"):
-        if resp.status_code == 422:
-            raise CCVAlreadyInjectedError
-    elif resp.status_code >= 400:
-        try:
-            d = json.loads(resp.content.decode("utf-8"))
-        except Exception:
-            resp.raise_for_status()
-        else:
-            raise InjectAPIError(
-                f"Error {resp.status_code} from API: "
-                f"{d.get('info', 'missing details')}"
-            )
-
-
 class CalCatAPIClient:
     def __init__(self, base_api_url, oauth_client=None, user_email=""):
         if oauth_client is not None:
@@ -106,17 +78,6 @@ class CalCatAPIClient:
             _headers.update(headers)
         return self.session.get(url, params=params, headers=_headers, **kwargs)
 
-    def post_request(self, relative_url, data=None, headers=None, **kwargs):
-        """Make a POST request, return the HTTP response object"""
-        # Base URL may include e.g. '/api/'. This is a prefix for all URLs;
-        # even if they look like an absolute path.
-        url = urljoin(self.base_api_url, relative_url.lstrip("/"))
-        _headers = self.default_headers()
-        if headers:
-            _headers.update(headers)
-        return self.session.post(
-            url, data=json.dumps(data), headers=_headers, **kwargs)
-
     @staticmethod
     def _parse_response(resp: requests.Response):
         if resp.status_code >= 400:
@@ -135,27 +96,11 @@ class CalCatAPIClient:
         else:
             return json.loads(resp.content.decode("utf-8"))
 
-    @staticmethod
-    def _parse_post_response(resp: requests.Response):
-
-        if resp.status_code >= 400:
-            _failed_response(resp)
-
-        if resp.content == b"":
-            return None
-        else:
-            return json.loads(resp.content.decode("utf-8"))
-
     def get(self, relative_url, params=None, **kwargs):
         """Make a GET request, return response content from JSON"""
         resp = self.get_request(relative_url, params, **kwargs)
         return self._parse_response(resp)
 
-    def post(self, relative_url, params=None, **kwargs):
-        """Make a POST request, return response content from JSON"""
-        resp = self.post_request(relative_url, params, **kwargs)
-        return self._parse_post_response(resp)
-
     _pagination_headers = (
         "X-Total-Pages",
         "X-Count-Per-Page",
@@ -230,20 +175,6 @@ class CalCatAPIClient:
         return self.get_by_name(
             "physical_detector_units", name, name_key="physical_name")
 
-    # Posting APIs
-    def set_expected_condition(self, conditions: dict):
-        return self.post("conditions/set_expected_condition", conditions)
-
-    def create_calibration_constant(self, calibration_constant: dict):
-        return self.post("calibration_constants", calibration_constant)
-
-    def get_or_create_report(self, report: dict):
-        # Based on create or get API
-        return self.post("reports/set", report)
-
-    def create_calibration_constant_version(self, ccv: dict):
-        return self.post(CALIBRATION_CONSTANT_VERSIONS, ccv)
-
 
 global_client = None
 

src/cal_tools/constants.py

 import binascii
+import json
 import time
 from dataclasses import asdict, dataclass
 
@@ -8,11 +9,14 @@ from pathlib import Path
 from shutil import copyfile
 from struct import pack, unpack
 from typing import List, Tuple, Union
+from urllib.parse import urljoin
 
 import h5py
 import numpy as np
+import requests
 
 from cal_tools.calcat_interface2 import (
+    CalCatAPIClient,
     CalCatAPIError,
     get_default_caldb_root,
 )
@@ -20,6 +24,79 @@ from cal_tools.calcat_interface2 import (
 CONDITION_NAME_MAX_LENGTH = 60
 
 
+class InjectAPIError(CalCatAPIError):
+    ...
+
+
+class CCVAlreadyInjectedError(InjectAPIError):
+    ...
+
+
+def _failed_response(resp):
+    # TODO: Add more errors if needed
+    if "calibration_constant_versions" in resp.url.lstrip("/"):
+        if resp.status_code == 422:
+            raise CCVAlreadyInjectedError
+    elif resp.status_code >= 400:
+        try:
+            d = json.loads(resp.content.decode("utf-8"))
+        except Exception:
+            resp.raise_for_status()
+        else:
+            raise InjectAPIError(
+                f"Error {resp.status_code} from API: "
+                f"{d.get('info', 'missing details')}"
+            )
+
+
+class InjectAPI(CalCatAPIClient):
+    def __init__(self, base_api_url, oauth_client=None, user_email=""):
+        # Call the parent class initializer
+        super().__init__(base_api_url, oauth_client, user_email)
+        self.base_api_url = base_api_url
+
+    def post_request(self, relative_url, data=None, headers=None, **kwargs):
+        """Make a POST request, return the HTTP response object"""
+        # Base URL may include e.g. '/api/'. This is a prefix for all URLs;
+        # even if they look like an absolute path.
+        url = urljoin(self.base_api_url, relative_url.lstrip("/"))
+        _headers = self.default_headers()
+        if headers:
+            _headers.update(headers)
+        return self.session.post(
+            url, data=json.dumps(data), headers=_headers, **kwargs)
+
+    @staticmethod
+    def _parse_post_response(resp: requests.Response):
+
+        if resp.status_code >= 400:
+            _failed_response(resp)
+
+        if resp.content == b"":
+            return None
+        else:
+            return json.loads(resp.content.decode("utf-8"))
+
+    def post(self, relative_url, params=None, **kwargs):
+        """Make a POST request, return response content from JSON"""
+        resp = self.post_request(relative_url, params, **kwargs)
+        return self._parse_post_response(resp)
+
+    # Posting APIs
+    def set_expected_condition(self, conditions: dict):
+        return self.post("conditions/set_expected_condition", conditions)
+
+    def create_calibration_constant(self, calibration_constant: dict):
+        return self.post("calibration_constants", calibration_constant)
+
+    def get_or_create_report(self, report: dict):
+        # Based on create or get API
+        return self.post("reports/set", report)
+
+    def create_calibration_constant_version(self, ccv: dict):
+        return self.post("calibration_constant_versions", ccv)
+
+
 @dataclass
 class ParameterConditionAttribute:
     value: str
@@ -55,14 +132,6 @@ def create_unique_cc_name(det_type, calibration, condition_name):
     return f'{cc_name_1[:40]}_{condition_name}'
 
 
-class InjectAPIError(CalCatAPIError):
-    ...
-
-
-class CCVAlreadyInjectedError(InjectAPIError):
-    ...
-
-
 def create_unique_ccv_name(start_idx):
     # Generate unique name if it doesn't exist
     datetime_str = datetime_str = datetime.now(