diff --git a/src/cal_tools/calcat_interface2.py b/src/cal_tools/calcat_interface2.py
index 4bddc2b7f753ca47ce857291635b8f8956d5778c..53b01f8436c7f596f2a3090b94b477925041c581 100644
--- a/src/cal_tools/calcat_interface2.py
+++ b/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
diff --git a/src/cal_tools/constants.py b/src/cal_tools/constants.py
index 98bc70a1505795f0fbe6be3565d0c65d082c7979..b7d4f8fc9b87112cd604a39c6c4957eb18d9e594 100644
--- a/src/cal_tools/constants.py
+++ b/src/cal_tools/constants.py
@@ -1,4 +1,5 @@
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(