ops_utils.request_util

Module to handle web requests.

  1"""Module to handle web requests."""
  2from typing import Any, Optional
  3import requests
  4import backoff
  5
  6from .token_util import Token
  7from .vars import ARG_DEFAULTS
  8
  9GET = "GET"
 10"""Method used for API GET endpoints"""
 11POST = "POST"
 12"""Method used for API POST endpoints"""
 13DELETE = "DELETE"
 14"""Method used for API DELETE endpoints"""
 15PATCH = "PATCH"
 16"""Method used for API PATCH endpoints"""
 17PUT = "PUT"
 18"""Method used for API PUT endpoints"""
 19
 20
 21class RunRequest:
 22    """Class to handle web requests with retries and backoff."""
 23
 24    def __init__(
 25            self,
 26            token: Token,
 27            max_retries: int = ARG_DEFAULTS["max_retries"],  # type: ignore[assignment]
 28            max_backoff_time: int = ARG_DEFAULTS["max_backoff_time"],  # type: ignore[assignment]
 29    ):
 30        """
 31        Initialize the RunRequest class.
 32
 33        **Args:**
 34        - token (`ops_utils.token_util.Token`): The token used for authentication
 35        - max_retries (int, optional): Maximum number of retries for a request. Defaults to `5`.
 36        - max_backoff_time (int, optional): Maximum backoff time for a request (in seconds). Defaults to `300`.
 37        """
 38        self.token = token
 39        """@private"""
 40        self.max_retries = max_retries
 41        """@private"""
 42        self.max_backoff_time = max_backoff_time
 43        """@private"""
 44
 45    @staticmethod
 46    def _create_backoff_decorator(max_tries: int, factor: int, max_time: int) -> Any:
 47        """
 48        Create a backoff decorator with the specified parameters.
 49
 50        Args:
 51            max_tries (int): The maximum number of tries.
 52            factor (int): The exponential backoff factor.
 53            max_time (int): The maximum backoff time in seconds.
 54
 55        Returns:
 56            Any: The backoff decorator.
 57        """
 58        return backoff.on_exception(
 59            backoff.expo,
 60            requests.exceptions.RequestException,
 61            max_tries=max_tries,
 62            factor=factor,
 63            max_time=max_time
 64        )
 65
 66    def run_request(
 67            self,
 68            uri: str,
 69            method: str,
 70            data: Any = None,
 71            files: Any = None,
 72            params: Optional[dict] = None,
 73            factor: int = 15,
 74            accept: Optional[str] = "application/json",
 75            content_type: Optional[str] = None,
 76            accept_return_codes: list[int] = []
 77    ) -> requests.Response:
 78        """
 79        Run an HTTP request with retries and backoff.
 80
 81        **Args:**
 82        - uri (str): The URI for the request.
 83        - method (str): The HTTP method (must be one of `GET`, `POST`, `DELETE`, `PATCH`, `PUT`).
 84        - data (Any, optional): The data to send in the request body. Defaults to None.
 85        - params (dict, optional): The query parameters for the request. Defaults to None.
 86        - factor (int, optional): The exponential backoff factor. Defaults to 15.
 87        - accept (str, optional): The accept header for the request. Defaults to "application/json".
 88        - content_type (str, optional): The content type for the request. Defaults to None.
 89        - accept_return_codes (list[int], optional): List of acceptable return codes. Defaults to [].
 90
 91        **Returns:**
 92        - requests.Response: The response from the request.
 93        """
 94        # Create a custom backoff decorator with the provided parameters
 95        backoff_decorator = self._create_backoff_decorator(
 96            max_tries=self.max_retries,
 97            factor=factor,
 98            max_time=self.max_backoff_time
 99        )
100
101        # Apply decorators to request execution
102        @backoff_decorator
103        def _make_request() -> requests.Response:
104            headers = self.create_headers(content_type=content_type, accept=accept)
105            if method == GET:
106                response = requests.get(
107                    uri,
108                    headers=headers,
109                    params=params
110                )
111            elif method == POST:
112                if files:
113                    response = requests.post(
114                        uri,
115                        headers=headers,
116                        files=files
117                    )
118                else:
119                    response = requests.post(
120                        uri,
121                        headers=headers,
122                        data=data
123                    )
124            elif method == DELETE:
125                response = requests.delete(
126                    uri,
127                    headers=headers
128                )
129            elif method == PATCH:
130                response = requests.patch(
131                    uri,
132                    headers=headers,
133                    data=data
134                )
135            elif method == PUT:
136                response = requests.put(
137                    uri,
138                    headers=headers,
139                    data=data
140                )
141            else:
142                raise ValueError(f"Method {method} is not supported")
143            # Raise an exception for non-200 status codes and codes not in acceptable_return_codes
144            if ((300 <= response.status_code or response.status_code < 200)
145                    and response.status_code not in accept_return_codes):
146                print(response.text)
147                response.raise_for_status()  # Raise an exception for non-200 status codes
148            return response
149
150        return _make_request()
151
152    def create_headers(self, content_type: Optional[str] = None, accept: Optional[str] = "application/json") -> dict:
153        """
154        Create headers for API calls.
155
156        **Args:**
157        - content_type (str, optional): The content type for the request. Defaults to None.
158        - accept (str, optional): The accept header for the request. Defaults to "application/json".
159
160        **Returns:**
161        - dict: The headers for the request.
162        """
163        self.token.get_token()
164        headers = {
165            "Authorization": f"Bearer {self.token.token_string}",
166            "accept": accept
167        }
168        if content_type:
169            headers["Content-Type"] = content_type
170        if accept:
171            headers["accept"] = accept
172        return headers
173
174    def upload_file(self, uri: str, data: dict, accept: str = "*/*") -> requests.Response:
175        """
176        Run a POST request with files parameter.
177
178        **Args:**
179        - uri (str): The URI for the request.
180        - data (dict): The files data to upload.
181        - accept (str, optional): The accept header for the request. Defaults to "*/*".
182
183        **Returns:**
184        - requests.Response: The response from the request.
185        """
186        return self.run_request(
187            uri=uri,
188            method=POST,
189            files=data,
190            accept=accept,
191        )
GET = 'GET'

Method used for API GET endpoints

POST = 'POST'

Method used for API POST endpoints

DELETE = 'DELETE'

Method used for API DELETE endpoints

PATCH = 'PATCH'

Method used for API PATCH endpoints

PUT = 'PUT'

Method used for API PUT endpoints

class RunRequest:
 22class RunRequest:
 23    """Class to handle web requests with retries and backoff."""
 24
 25    def __init__(
 26            self,
 27            token: Token,
 28            max_retries: int = ARG_DEFAULTS["max_retries"],  # type: ignore[assignment]
 29            max_backoff_time: int = ARG_DEFAULTS["max_backoff_time"],  # type: ignore[assignment]
 30    ):
 31        """
 32        Initialize the RunRequest class.
 33
 34        **Args:**
 35        - token (`ops_utils.token_util.Token`): The token used for authentication
 36        - max_retries (int, optional): Maximum number of retries for a request. Defaults to `5`.
 37        - max_backoff_time (int, optional): Maximum backoff time for a request (in seconds). Defaults to `300`.
 38        """
 39        self.token = token
 40        """@private"""
 41        self.max_retries = max_retries
 42        """@private"""
 43        self.max_backoff_time = max_backoff_time
 44        """@private"""
 45
 46    @staticmethod
 47    def _create_backoff_decorator(max_tries: int, factor: int, max_time: int) -> Any:
 48        """
 49        Create a backoff decorator with the specified parameters.
 50
 51        Args:
 52            max_tries (int): The maximum number of tries.
 53            factor (int): The exponential backoff factor.
 54            max_time (int): The maximum backoff time in seconds.
 55
 56        Returns:
 57            Any: The backoff decorator.
 58        """
 59        return backoff.on_exception(
 60            backoff.expo,
 61            requests.exceptions.RequestException,
 62            max_tries=max_tries,
 63            factor=factor,
 64            max_time=max_time
 65        )
 66
 67    def run_request(
 68            self,
 69            uri: str,
 70            method: str,
 71            data: Any = None,
 72            files: Any = None,
 73            params: Optional[dict] = None,
 74            factor: int = 15,
 75            accept: Optional[str] = "application/json",
 76            content_type: Optional[str] = None,
 77            accept_return_codes: list[int] = []
 78    ) -> requests.Response:
 79        """
 80        Run an HTTP request with retries and backoff.
 81
 82        **Args:**
 83        - uri (str): The URI for the request.
 84        - method (str): The HTTP method (must be one of `GET`, `POST`, `DELETE`, `PATCH`, `PUT`).
 85        - data (Any, optional): The data to send in the request body. Defaults to None.
 86        - params (dict, optional): The query parameters for the request. Defaults to None.
 87        - factor (int, optional): The exponential backoff factor. Defaults to 15.
 88        - accept (str, optional): The accept header for the request. Defaults to "application/json".
 89        - content_type (str, optional): The content type for the request. Defaults to None.
 90        - accept_return_codes (list[int], optional): List of acceptable return codes. Defaults to [].
 91
 92        **Returns:**
 93        - requests.Response: The response from the request.
 94        """
 95        # Create a custom backoff decorator with the provided parameters
 96        backoff_decorator = self._create_backoff_decorator(
 97            max_tries=self.max_retries,
 98            factor=factor,
 99            max_time=self.max_backoff_time
100        )
101
102        # Apply decorators to request execution
103        @backoff_decorator
104        def _make_request() -> requests.Response:
105            headers = self.create_headers(content_type=content_type, accept=accept)
106            if method == GET:
107                response = requests.get(
108                    uri,
109                    headers=headers,
110                    params=params
111                )
112            elif method == POST:
113                if files:
114                    response = requests.post(
115                        uri,
116                        headers=headers,
117                        files=files
118                    )
119                else:
120                    response = requests.post(
121                        uri,
122                        headers=headers,
123                        data=data
124                    )
125            elif method == DELETE:
126                response = requests.delete(
127                    uri,
128                    headers=headers
129                )
130            elif method == PATCH:
131                response = requests.patch(
132                    uri,
133                    headers=headers,
134                    data=data
135                )
136            elif method == PUT:
137                response = requests.put(
138                    uri,
139                    headers=headers,
140                    data=data
141                )
142            else:
143                raise ValueError(f"Method {method} is not supported")
144            # Raise an exception for non-200 status codes and codes not in acceptable_return_codes
145            if ((300 <= response.status_code or response.status_code < 200)
146                    and response.status_code not in accept_return_codes):
147                print(response.text)
148                response.raise_for_status()  # Raise an exception for non-200 status codes
149            return response
150
151        return _make_request()
152
153    def create_headers(self, content_type: Optional[str] = None, accept: Optional[str] = "application/json") -> dict:
154        """
155        Create headers for API calls.
156
157        **Args:**
158        - content_type (str, optional): The content type for the request. Defaults to None.
159        - accept (str, optional): The accept header for the request. Defaults to "application/json".
160
161        **Returns:**
162        - dict: The headers for the request.
163        """
164        self.token.get_token()
165        headers = {
166            "Authorization": f"Bearer {self.token.token_string}",
167            "accept": accept
168        }
169        if content_type:
170            headers["Content-Type"] = content_type
171        if accept:
172            headers["accept"] = accept
173        return headers
174
175    def upload_file(self, uri: str, data: dict, accept: str = "*/*") -> requests.Response:
176        """
177        Run a POST request with files parameter.
178
179        **Args:**
180        - uri (str): The URI for the request.
181        - data (dict): The files data to upload.
182        - accept (str, optional): The accept header for the request. Defaults to "*/*".
183
184        **Returns:**
185        - requests.Response: The response from the request.
186        """
187        return self.run_request(
188            uri=uri,
189            method=POST,
190            files=data,
191            accept=accept,
192        )

Class to handle web requests with retries and backoff.

RunRequest( token: ops_utils.token_util.Token, max_retries: int = 5, max_backoff_time: int = 300)
25    def __init__(
26            self,
27            token: Token,
28            max_retries: int = ARG_DEFAULTS["max_retries"],  # type: ignore[assignment]
29            max_backoff_time: int = ARG_DEFAULTS["max_backoff_time"],  # type: ignore[assignment]
30    ):
31        """
32        Initialize the RunRequest class.
33
34        **Args:**
35        - token (`ops_utils.token_util.Token`): The token used for authentication
36        - max_retries (int, optional): Maximum number of retries for a request. Defaults to `5`.
37        - max_backoff_time (int, optional): Maximum backoff time for a request (in seconds). Defaults to `300`.
38        """
39        self.token = token
40        """@private"""
41        self.max_retries = max_retries
42        """@private"""
43        self.max_backoff_time = max_backoff_time
44        """@private"""

Initialize the RunRequest class.

Args:

  • token (ops_utils.token_util.Token): The token used for authentication
  • max_retries (int, optional): Maximum number of retries for a request. Defaults to 5.
  • max_backoff_time (int, optional): Maximum backoff time for a request (in seconds). Defaults to 300.
def run_request( self, uri: str, method: str, data: Any = None, files: Any = None, params: Optional[dict] = None, factor: int = 15, accept: Optional[str] = 'application/json', content_type: Optional[str] = None, accept_return_codes: list[int] = []) -> requests.models.Response:
 67    def run_request(
 68            self,
 69            uri: str,
 70            method: str,
 71            data: Any = None,
 72            files: Any = None,
 73            params: Optional[dict] = None,
 74            factor: int = 15,
 75            accept: Optional[str] = "application/json",
 76            content_type: Optional[str] = None,
 77            accept_return_codes: list[int] = []
 78    ) -> requests.Response:
 79        """
 80        Run an HTTP request with retries and backoff.
 81
 82        **Args:**
 83        - uri (str): The URI for the request.
 84        - method (str): The HTTP method (must be one of `GET`, `POST`, `DELETE`, `PATCH`, `PUT`).
 85        - data (Any, optional): The data to send in the request body. Defaults to None.
 86        - params (dict, optional): The query parameters for the request. Defaults to None.
 87        - factor (int, optional): The exponential backoff factor. Defaults to 15.
 88        - accept (str, optional): The accept header for the request. Defaults to "application/json".
 89        - content_type (str, optional): The content type for the request. Defaults to None.
 90        - accept_return_codes (list[int], optional): List of acceptable return codes. Defaults to [].
 91
 92        **Returns:**
 93        - requests.Response: The response from the request.
 94        """
 95        # Create a custom backoff decorator with the provided parameters
 96        backoff_decorator = self._create_backoff_decorator(
 97            max_tries=self.max_retries,
 98            factor=factor,
 99            max_time=self.max_backoff_time
100        )
101
102        # Apply decorators to request execution
103        @backoff_decorator
104        def _make_request() -> requests.Response:
105            headers = self.create_headers(content_type=content_type, accept=accept)
106            if method == GET:
107                response = requests.get(
108                    uri,
109                    headers=headers,
110                    params=params
111                )
112            elif method == POST:
113                if files:
114                    response = requests.post(
115                        uri,
116                        headers=headers,
117                        files=files
118                    )
119                else:
120                    response = requests.post(
121                        uri,
122                        headers=headers,
123                        data=data
124                    )
125            elif method == DELETE:
126                response = requests.delete(
127                    uri,
128                    headers=headers
129                )
130            elif method == PATCH:
131                response = requests.patch(
132                    uri,
133                    headers=headers,
134                    data=data
135                )
136            elif method == PUT:
137                response = requests.put(
138                    uri,
139                    headers=headers,
140                    data=data
141                )
142            else:
143                raise ValueError(f"Method {method} is not supported")
144            # Raise an exception for non-200 status codes and codes not in acceptable_return_codes
145            if ((300 <= response.status_code or response.status_code < 200)
146                    and response.status_code not in accept_return_codes):
147                print(response.text)
148                response.raise_for_status()  # Raise an exception for non-200 status codes
149            return response
150
151        return _make_request()

Run an HTTP request with retries and backoff.

Args:

  • uri (str): The URI for the request.
  • method (str): The HTTP method (must be one of GET, POST, DELETE, PATCH, PUT).
  • data (Any, optional): The data to send in the request body. Defaults to None.
  • params (dict, optional): The query parameters for the request. Defaults to None.
  • factor (int, optional): The exponential backoff factor. Defaults to 15.
  • accept (str, optional): The accept header for the request. Defaults to "application/json".
  • content_type (str, optional): The content type for the request. Defaults to None.
  • accept_return_codes (list[int], optional): List of acceptable return codes. Defaults to [].

Returns:

  • requests.Response: The response from the request.
def create_headers( self, content_type: Optional[str] = None, accept: Optional[str] = 'application/json') -> dict:
153    def create_headers(self, content_type: Optional[str] = None, accept: Optional[str] = "application/json") -> dict:
154        """
155        Create headers for API calls.
156
157        **Args:**
158        - content_type (str, optional): The content type for the request. Defaults to None.
159        - accept (str, optional): The accept header for the request. Defaults to "application/json".
160
161        **Returns:**
162        - dict: The headers for the request.
163        """
164        self.token.get_token()
165        headers = {
166            "Authorization": f"Bearer {self.token.token_string}",
167            "accept": accept
168        }
169        if content_type:
170            headers["Content-Type"] = content_type
171        if accept:
172            headers["accept"] = accept
173        return headers

Create headers for API calls.

Args:

  • content_type (str, optional): The content type for the request. Defaults to None.
  • accept (str, optional): The accept header for the request. Defaults to "application/json".

Returns:

  • dict: The headers for the request.
def upload_file( self, uri: str, data: dict, accept: str = '*/*') -> requests.models.Response:
175    def upload_file(self, uri: str, data: dict, accept: str = "*/*") -> requests.Response:
176        """
177        Run a POST request with files parameter.
178
179        **Args:**
180        - uri (str): The URI for the request.
181        - data (dict): The files data to upload.
182        - accept (str, optional): The accept header for the request. Defaults to "*/*".
183
184        **Returns:**
185        - requests.Response: The response from the request.
186        """
187        return self.run_request(
188            uri=uri,
189            method=POST,
190            files=data,
191            accept=accept,
192        )

Run a POST request with files parameter.

Args:

  • uri (str): The URI for the request.
  • data (dict): The files data to upload.
  • accept (str, optional): The accept header for the request. Defaults to "/".

Returns:

  • requests.Response: The response from the request.