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.