capsule some of the request calls

src/cript/api/api.py

@@ -3,7 +3,6 @@
 import logging
 import os
 import uuid
-import warnings
 from pathlib import Path
 from typing import Any, Dict, Optional, Union
 
@@ -17,9 +16,7 @@
     APIError,
     CRIPTAPIRequiredError,
     CRIPTAPISaveError,
-    CRIPTConnectionError,
     CRIPTDuplicateNameError,
-    InvalidHostError,
 )
 from cript.api.paginator import Paginator
 from cript.api.utils.aws_s3_utils import get_s3_client
@@ -208,7 +205,7 @@ def __init__(self, host: Union[str, None] = None, api_token: Union[str, None] =
             api_token = authentication_dict["api_token"]
             storage_token = authentication_dict["storage_token"]
 
-        self._host = self._prepare_host(host=host)  # type: ignore
+        self._host: str = host.rstrip("/")
         self._api_token = api_token  # type: ignore
         self._storage_token = storage_token  # type: ignore
 
@@ -220,7 +217,7 @@ def __init__(self, host: Union[str, None] = None, api_token: Union[str, None] =
 
         # set a logger instance to use for the class logs
         self._init_logger(default_log_level)
-        self._db_schema = DataSchema(self.host, self.logger)
+        self._db_schema = DataSchema(self)
 
     def __str__(self) -> str:
         """
@@ -283,46 +280,6 @@ def _init_logger(self, log_level=logging.INFO) -> None:
     def logger(self):
         return self._logger
 
-    @beartype
-    def _prepare_host(self, host: str) -> str:
-        """
-        Takes the host URL provided by the user during API object construction (e.g., `https://api.criptapp.org`)
-        and standardizes it for internal use. Performs any required string manipulation to ensure uniformity.
-
-        Parameters
-        ----------
-        host: str
-            The host URL specified during API initialization, typically in the form `https://api.criptapp.org`.
-
-        Warnings
-        --------
-        If the specified host uses the unsafe "http://" protocol, a warning will be raised to consider using HTTPS.
-
-        Raises
-        ------
-        InvalidHostError
-            If the host string does not start with either "http" or "https", an InvalidHostError will be raised.
-            Only HTTP protocol is acceptable at this time.
-
-        Returns
-        -------
-        str
-            A standardized host string formatted for internal use.
-
-        """
-        # strip ending slash to make host always uniform
-        host = host.rstrip("/")
-        host = f"{host}/{self._api_prefix}/{self._api_version}"
-
-        # if host is using unsafe "http://" then give a warning
-        if host.startswith("http://"):
-            warnings.warn("HTTP is an unsafe protocol please consider using HTTPS.")
-
-        if not host.startswith("http"):
-            raise InvalidHostError()
-
-        return host
-
     # Use a property to ensure delayed init of s3_client
     @property
     def _s3_client(self) -> boto3.client:  # type: ignore
@@ -414,24 +371,6 @@ def host(self):
         """
         return self._host
 
-    def _check_initial_host_connection(self) -> None:
-        """
-        tries to create a connection with host and if the host does not respond or is invalid it raises an error
-
-        Raises
-        -------
-        CRIPTConnectionError
-            raised when the host does not give the expected response
-
-        Returns
-        -------
-        None
-        """
-        try:
-            pass
-        except Exception as exc:
-            raise CRIPTConnectionError(self.host, self._api_token) from exc
-
     def save(self, project: Project) -> None:
         """
         This method takes a project node, serializes the class into JSON
@@ -497,7 +436,7 @@ def _internal_save(self, node, save_values: Optional[_InternalSaveValues] = None
 
             # This checks if the current node exists on the back end.
             # if it does exist we use `patch` if it doesn't `post`.
-            test_get_response: Dict = requests.get(url=f"{self._host}/{node.node_type_snake_case}/{str(node.uuid)}/", headers=self._http_headers, timeout=_API_TIMEOUT).json()
+            test_get_response: Dict = self._capsule_request(url_path=f"/{node.node_type_snake_case}/{str(node.uuid)}/", method="GET").json()
             patch_request = test_get_response["code"] == 200
 
             # TODO remove once get works properly
@@ -512,13 +451,15 @@ def _internal_save(self, node, save_values: Optional[_InternalSaveValues] = None
                 response = {"code": 200}
                 break
 
+            method = "POST"
             if patch_request:
-                response: Dict = requests.patch(url=f"{self._host}/{node.node_type_snake_case}/{str(node.uuid)}/", headers=self._http_headers, data=json_data, timeout=_API_TIMEOUT).json()  # type: ignore
-            else:
-                response: Dict = requests.post(url=f"{self._host}/{node.node_type_snake_case}/", headers=self._http_headers, data=json_data, timeout=_API_TIMEOUT).json()  # type: ignore
-                # if node.node_type != "Project":
-                #     test_success: Dict = requests.get(url=f"{self._host}/{node.node_type_snake_case}/{str(node.uuid)}/", headers=self._http_headers, timeout=_API_TIMEOUT).json()
-                #     print("XYZ", json_data, save_values, response, test_success)
+                method = "PATCH"
+
+            response: Dict = self._capsule_request(url=f"/{node.node_type_snake_case}/{str(node.uuid)}/", method=method, data=json_data).json()  # type: ignore
+
+            # if node.node_type != "Project":
+            #     test_success: Dict = requests.get(url=f"{self._host}/{node.node_type_snake_case}/{str(node.uuid)}/", headers=self._http_headers, timeout=_API_TIMEOUT).json()
+            #     print("XYZ", json_data, save_values, response, test_success)
 
             # print(json_data, patch_request, response, save_values)
             # If we get an error we may be able to fix, we to handle this extra and save the bad node first.
@@ -997,11 +938,59 @@ def delete_node_by_uuid(self, node_type: str, node_uuid: str) -> None:
         -------
         None
         """
-        delete_node_api_url: str = f"{self._host}/{node_type.lower()}/{node_uuid}/"
 
-        response: Dict = requests.delete(headers=self._http_headers, url=delete_node_api_url, timeout=_API_TIMEOUT).json()
+        response: Dict = self._capsule_request(url_path=f"/{node_type.lower()}/{node_uuid}/", method="DELETE").json()
 
         if response["code"] != 200:
-            raise APIError(api_error=str(response), http_method="DELETE", api_url=delete_node_api_url)
+            raise APIError(api_error=str(response), http_method="DELETE", api_url=f"/{node_type.lower()}/{node_uuid}/")
 
         self.logger.info(f"Deleted '{node_type.title()}' with UUID of '{node_uuid}' from CRIPT API.")
+
+    def _capsule_request(self, url_path: str, method: str, api_request: bool = True, headers: Optional[Dict] = None, timeout: int = _API_TIMEOUT, **kwargs) -> requests.Response:
+        """Helper function that capsules every request call we make against the backend.
+
+        Please *always* use this methods instead of `requests` directly.
+        We can log all request calls this way, which can help debugging immensely.
+
+        Parameters
+        ----------
+        url_path:str
+          URL path that we want to request from. So every thing that follows api.host. You can omit the api prefix and api version if you use api_request=True they are automatically added.
+
+        method: str
+          One of `GET`, `OPTIONS`, `HEAD`, `POST`, `PUT, `PATCH`, or `DELETE` as this will directly passed to `requests.request(...)`. See https://docs.python-requests.org/en/latest/api/ for details.
+
+        headers: Dict
+          HTTPS headers to use for the request.
+          If None (default) use the once associated with this API object for authentication.
+
+        timeout:int
+          Time out to be used for the request call.
+
+        kwargs
+          additional keyword arguments that are passed to `request.request`
+        """
+
+        extra_debug_info: bool = True
+
+        if headers is None:
+            headers = self._http_headers
+
+        url: str = self.host
+        if api_request:
+            url += f"/{self._api_prefix}/{self._api_version}"
+        url += url_path
+
+        pre_log_message: str = f"Requesting {method} from {url}"
+        if extra_debug_info:
+            pre_log_message += f"headers {headers} kwargs {kwargs}"
+        pre_log_message += "..."
+        self.logger.debug(pre_log_message)
+
+        response: requests.Response = requests.request(url=url, method=method, headers=headers, timeout=timeout, **kwargs)
+        post_log_message: str = f"Request return with {response.status_code}"
+        if extra_debug_info:
+            post_log_message += f" {response}"
+        self.logger.debug(post_log_message)
+
+        return response

src/cript/api/data_schema.py

@@ -2,10 +2,8 @@
 from typing import Union
 
 import jsonschema
-import requests
 from beartype import beartype
 
-from cript.api.api_config import _API_TIMEOUT
 from cript.api.exceptions import APIError, InvalidVocabulary
 from cript.api.utils.helper_functions import _get_node_type_from_json
 from cript.api.vocabulary_categories import VocabCategories
@@ -20,14 +18,30 @@ class DataSchema:
 
     _vocabulary: dict = {}
     _db_schema: dict = {}
-    _logger = None
+    _api = None
     # Advanced User Tip: Disabling Node Validation
     # For experienced users, deactivating node validation during creation can be a time-saver.
     # Note that the complete node graph will still undergo validation before being saved to the back end.
     # Caution: It's advisable to keep validation active while debugging scripts, as disabling it can delay error notifications and complicate the debugging process.
     skip_validation: bool = False
 
-    def _get_db_schema(self, host: str) -> dict:
+    def __init__(self, api):
+        """
+        Initialize DataSchema class with a full hostname to fetch the node validation schema.
+
+        Examples
+        --------
+        ### Create a stand alone DataSchema instance.
+        >>> import cript
+        >>> with cript.API(host="https://api.criptapp.org/") as api:
+        ...    data_schema = cript.api.DataSchema(api)
+        """
+
+        self._db_schema = self._get_db_schema()
+        self._vocabulary = self._get_vocab()
+        self._api = api
+
+    def _get_db_schema(self) -> dict:
         """
         Sends a GET request to CRIPT to get the database schema and returns it.
         The database schema can be used for validating the JSON request
@@ -45,15 +59,13 @@ def _get_db_schema(self, host: str) -> dict:
             return self._db_schema
 
         # fetch db_schema from API
-        if self._logger:
-            self._logger.info(f"Loading node validation schema from {host}/schema/")
+        self._api.logger.info(f"Loading node validation schema from {self._api.host}/schema/")
         # fetch db schema from API
-        response: requests.Response = requests.get(url=f"{host}/schema/", timeout=_API_TIMEOUT)
+        response = self._api._capsule_request(url_path="/schema/", method="GET")
 
         # raise error if not HTTP 200
         response.raise_for_status()
-        if self._logger:
-            self._logger.info(f"Loading node validation schema from {host}/schema/ was successful.")
+        self._api.logger.info(f"Loading node validation schema from {self._api.host}/schema/ was successful.")
 
         # if no error, take the JSON from the API response
         response_dict: dict = response.json()
@@ -63,22 +75,6 @@ def _get_db_schema(self, host: str) -> dict:
 
         return db_schema
 
-    def __init__(self, host: str, logger=None):
-        """
-        Initialize DataSchema class with a full hostname to fetch the node validation schema.
-
-        Examples
-        --------
-        ### Create a stand alone DataSchema instance.
-        >>> import cript
-        >>> with cript.API(host="https://api.criptapp.org/") as api:
-        ...    data_schema = cript.api.DataSchema(api.host)
-        """
-
-        self._db_schema = self._get_db_schema(host)
-        self._vocabulary = self._get_vocab(host)
-        self._logger = logger
-
     def _get_vocab(self, host: str) -> dict:
         """
         gets the entire CRIPT controlled vocabulary and stores it in _vocabulary
@@ -107,10 +103,10 @@ def _get_vocab(self, host: str) -> dict:
         # loop through all vocabulary categories and make a request to each vocabulary category
         # and put them all inside of self._vocab with the keys being the vocab category name
         for category in VocabCategories:
-            vocabulary_category_url: str = f"{host}/cv/{category.value}/"
+            vocabulary_category_url: str = f"/cv/{category.value}/"
 
             # if vocabulary category is not in cache, then get it from API and cache it
-            response: dict = requests.get(url=vocabulary_category_url, timeout=_API_TIMEOUT).json()
+            response: dict = self._api._capsule_request(url_path=vocabulary_category_url, method="GET").json()
 
             if response["code"] != 200:
                 raise APIError(api_error=str(response), http_method="GET", api_url=vocabulary_category_url)
@@ -247,8 +243,7 @@ def is_node_schema_valid(self, node_json: str, is_patch: bool = False, force_val
         else:
             log_message += " (Can be disabled by setting `cript.API.skip_validation = True`.)"
 
-        if self._logger:
-            self._logger.info(log_message)
+        self._api.logger.info(log_message)
 
         # set the schema to test against http POST or PATCH of DB Schema
         schema_http_method: str