pumpwood_communication.microservice_abc.base
Module with base classes for requests on Pumpwood Server.
41class PumpWoodMicroServiceBase: 42 """Base class for Pumpwood MicroService. 43 44 Enviroment variables can be used to set MicroService parameters: 45 - **PUMPWOOD_COMUNICATION__DEFAULT_TIMEOUT:** Default requests timeout in 46 seconds. 47 - **PUMPWOOD_COMUNICATION__DEBUG:** If object will be initiated using 48 debug parameter. It will have more verbosity and login at each 49 request. Options 'TRUE', 'FALSE'. 50 - **PUMPWOOD_COMUNICATION__VERIFY_SSL:** If requests will validate SSL 51 certificate. 52 """ 53 54 __base_header = {'Content-Type': 'application/json'} 55 """Base header for the requests.""" 56 57 @staticmethod 58 def _adjust_server_url(server_url): 59 """Remove tralling / if present on server URL.""" 60 if server_url is None: 61 return None 62 if server_url[-1] != '/': 63 return server_url + '/' 64 else: 65 return server_url 66 67 def __init__(self, name: str = None, server_url: str = None, 68 username: str = None, password: str = None, 69 verify_ssl: bool = True, debug: bool = None, 70 default_timeout: int = None, **kwargs): 71 """Create new PumpWoodMicroService object. 72 73 Creates a new microservice object. If just name is passed, object must 74 be initiate after with init() method. 75 76 Args: 77 name: 78 Name of the microservice, helps when exceptions 79 are raised. 80 server_url: 81 URL of the server that will be connected. 82 username: 83 Username that will be logged on. 84 password: 85 Variable to be converted to JSON and posted along 86 with the request. 87 verify_ssl: 88 Set if microservice will verify SSL certificate. 89 debug: 90 If microservice will be used as debug mode. This will obrigate 91 auth token refresh for each call. 92 default_timeout: 93 Default timeout for Pumpwood calls. 94 **kwargs: 95 Other parameters used for compatibility between versions. 96 97 Returns: 98 PumpWoodMicroService: New PumpWoodMicroService object 99 100 Raises: 101 No particular Raises. 102 """ 103 # Create attributes to be set at init function 104 self.name = None 105 """Name of the microservice instance.""" 106 self.server_url = None 107 """Pumpwood server URL.""" 108 self._default_timeout: int = None 109 """Default timeout for Pumpwood requests.""" 110 self._debug: bool = None 111 """Name of the microservice instance.""" 112 self._verify_ssl: bool = None 113 """If microservice should check the certificate.""" 114 self._is_mfa_login: bool = None 115 """Set if is MFA login.""" 116 self.__headers: dict = None 117 """Headers to be used on the requests.""" 118 self.__user: dict = None 119 """Information of the logged user.""" 120 self.__auth_header: dict = None 121 """Authenticated auth header.""" 122 self.__token_expiry: pd.Timedelta = None 123 """Expirity datetime of the authetication token.""" 124 self.__username: str = None 125 """Username associated with microservice.""" 126 self.__password: str = None 127 """Password associated with microservice.""" 128 self.init( 129 name=name, server_url=server_url, 130 username=username, password=password, 131 verify_ssl=verify_ssl, debug=debug, 132 default_timeout=default_timeout) 133 134 def init(self, name: str = None, server_url: str = None, 135 username: str = None, password: str = None, 136 verify_ssl: bool = True, debug: bool = None, 137 default_timeout: int = None, **kwargs): 138 """Lazzy initialization of the MicroService of object. 139 140 This function might be usefull to use the object as a singleton at 141 the backends. Using this function it is possible to instanciate an 142 empty object and them set the attributes latter at the systems. 143 144 Args: 145 name: 146 Name of the microservice, helps when exceptions 147 are raised. 148 server_url: 149 URL of the server that will be connected. 150 username: 151 Username that will be logged on. 152 password: 153 Variable to be converted to JSON and posted along 154 with the request. 155 verify_ssl: 156 Set if microservice will verify SSL certificate. 157 debug: 158 If microservice will be used as debug mode. This will obrigate 159 auth token refresh for each call. 160 default_timeout: 161 Default timeout for Pumpwood calls. 162 **kwargs: 163 Other parameters used for compatibility between versions. 164 165 Returns: 166 No return 167 168 Raises: 169 No particular Raises 170 """ 171 self.name = name 172 """Name of the microservice instance.""" 173 self.server_url = self._adjust_server_url(server_url) 174 """Pumpwood server URL.""" 175 176 # Set parameter using arguments or enviroment variables 177 if default_timeout is None: 178 self._default_timeout = int(os.getenv( 179 'PUMPWOOD_COMUNICATION__DEFAULT_TIMEOUT', 60)) 180 else: 181 self._default_timeout = default_timeout 182 183 if debug is None: 184 self._debug = os.getenv( 185 'PUMPWOOD_COMUNICATION__DEBUG', 'FALSE') == 'TRUE' 186 else: 187 self._debug = debug 188 189 if verify_ssl is None: 190 self._verify_ssl = os.getenv( 191 'PUMPWOOD_COMUNICATION__VERIFY_SSL', 'TRUE') == 'TRUE' 192 else: 193 self._verify_ssl = verify_ssl 194 195 self._is_mfa_login = False 196 self.__headers = None 197 self.__user = None 198 self.__auth_header = None 199 self.__token_expiry = None 200 self.__username = username 201 self.__password = password 202 203 @staticmethod 204 def angular_json(request_result) -> Any: 205 r"""Convert text to Json removing any XSSI at the beging of JSON. 206 207 Some backends add `)]}',\n` at the beginning of the JSON data to 208 prevent injection of functions. This function remove this characters 209 if present. 210 211 Args: 212 request_result: 213 JSON request to be converted. 214 215 Returns: 216 No return 217 218 Raises: 219 PumpWoodJSONLoadError: 220 If it is not possible to load JSON from request data. 221 """ 222 if request_result.text == '': 223 return None 224 225 string_start = ")]}',\n" 226 try: 227 if request_result.text[:6] == string_start: 228 return (orjson.loads(request_result.text[6:])) 229 else: 230 return (orjson.loads(request_result.text)) 231 except Exception: 232 msg = "Can not decode to Json" 233 raise PumpWoodJSONLoadError( 234 message=msg, payload={"request_data": request_result.text}) 235 236 def time_to_expiry(self) -> pd.Timedelta: 237 """Return time to token expiry. 238 239 Args: 240 No Args. 241 242 Returns: 243 Return time until token expiration. 244 """ 245 if self.__token_expiry is None: 246 return None 247 248 now_datetime = pd.to_datetime( 249 datetime.datetime.now(datetime.UTC), utc=True) 250 time_to_expiry = self.__token_expiry - now_datetime 251 return time_to_expiry 252 253 def is_credential_set(self) -> bool: 254 """Check if username and password are set on object. 255 256 Args: 257 No Args. 258 259 Returns: 260 True if usename and password were set during object creation or 261 later with init function. 262 """ 263 is_username_not_none = self.__username is not None 264 is_password_not_none = self.__password is not None 265 return is_username_not_none and is_password_not_none 266 267 @classmethod 268 def is_invalid_token_response(cls, response: Response) -> bool: 269 """Check if reponse has invalid token error. 270 271 Args: 272 response: 273 Request reponse to check for invalid token. 274 275 Returns: 276 Return True if response has an invalid token status. 277 """ 278 if response.status_code == 401: 279 return True 280 return False 281 282 def _login_resquest(self) -> Response: 283 """Request login with object credentials. 284 285 Args: 286 No Args 287 288 Returns (Response): 289 Request object. 290 """ 291 login_url = urljoin( 292 self.server_url, 'rest/registration/login/') 293 294 # Make a retry loop for authentication 295 login_result = None 296 for i in range(5): 297 login_result = requests.post( 298 login_url, json={ 299 'username': self.__username, 300 'password': self.__password}, 301 verify=self._verify_ssl, timeout=self._default_timeout) 302 if login_result.ok: 303 try: 304 login_data = self.angular_json(login_result) 305 return login_data 306 except Exception: # NOQA 307 pass 308 309 # Handle Unauthorized responses 310 elif self.is_invalid_token_response(login_result): 311 error_data = self.angular_json(login_result) 312 raise PumpWoodUnauthorized( 313 message="Login is not possible", 314 payload=error_data) 315 316 # Handle Forbidden responses 317 elif login_result.status_code == 403: 318 error_data = self.angular_json(login_result) 319 raise PumpWoodForbidden( 320 message="Login resquest is forbidden", 321 payload=error_data) 322 time.sleep(0.2) 323 324 # If response is not returned then something is not ok 325 error_data = self.angular_json(login_result) 326 raise PumpWoodOtherException( 327 message="Login not possible, server is not responding correctly", 328 payload=error_data) 329 330 def confirm_mfa_code(self, mfa_login_data: dict) -> dict: 331 """Ask user to confirm MFA code to login. 332 333 Open an input interface at terminal for user to validate MFA token. 334 335 Args: 336 mfa_login_data: 337 Result from login request with 'mfa_token' 338 as key. 339 340 Returns: 341 Return login returned with MFA confimation. 342 343 Raise: 344 Raise error if reponse is not valid using error_handler. 345 """ 346 code = input("## Please enter MFA code: ") 347 url = urljoin( 348 self.server_url, 'rest/registration/mfa-validate-code/') 349 mfa_response = requests.post(url, headers={ 350 "X-PUMPWOOD-MFA-Autorization": mfa_login_data['mfa_token']}, 351 json={"mfa_code": code}, timeout=self._default_timeout) 352 self.error_handler(mfa_response) 353 354 # Set _is_mfa_login true to indicate that login required MFA 355 self._is_mfa_login = True 356 return self.angular_json(mfa_response) 357 358 def login(self, force_refresh: bool = False) -> None: 359 """Log microservice in using username and password provided. 360 361 Args: 362 force_refresh (bool): 363 Force token refresh despise still valid 364 according to self.__token_expiry. 365 366 Returns: 367 No return 368 369 Raises: 370 Exception: 371 If login response has status diferent from 200. 372 """ 373 if not self.is_credential_set(): 374 raise PumpWoodUnauthorized( 375 message="Microservice username or/and password not set") 376 377 # Check if expiry time is 1h from now 378 refresh_expiry = False 379 if self.__token_expiry is None: 380 refresh_expiry = True 381 else: 382 time_to_expiry = self.time_to_expiry() 383 if time_to_expiry < datetime.timedelta(hours=1): 384 refresh_expiry = True 385 386 # When if debug always refresh token 387 if refresh_expiry or force_refresh or self._debug: 388 login_data = self._login_resquest() 389 if 'mfa_token' in login_data.keys(): 390 login_data = self.confirm_mfa_code( 391 mfa_login_data=login_data) 392 393 self.set_auth_header( 394 auth_token='Token ' + login_data['token'], 395 token_expiry=pd.to_datetime(login_data['expiry']), 396 user=login_data["user"]) 397 else: 398 # Token is not expired or envicted, them keep same token 399 return None 400 401 def logout(self, auth_header: dict = None) -> bool: 402 """Logout token. 403 404 Args: 405 auth_header: 406 Authentication header. 407 408 Returns: 409 True if logout was ok. 410 """ 411 resp = self.request_post( 412 url='rest/registration/logout/', 413 data={}, auth_header=auth_header) 414 # Set expiry to None to envict the token 415 self.__token_expiry = None 416 return resp is None 417 418 def logout_all(self, auth_header: dict = None) -> bool: 419 """Logout all tokens from user. 420 421 Args: 422 auth_header (dict): 423 Authentication header. 424 425 Returns: 426 True if logout all was ok. 427 """ 428 resp = self.request_post( 429 url='rest/registration/logoutall/', 430 data={}, auth_header=auth_header) 431 # Set expiry to None to envict the token 432 self.__token_expiry = None 433 return resp is None 434 435 def get_auth_header(self) -> dict: 436 """Retrieve auth_header and token_expiry from object. 437 438 Args: 439 No Args. 440 441 Returns: 442 Return authorization header and token_expiry datetime from object. 443 """ 444 # Copy the dictonary to avoid updating the original one 445 return copy.deepcopy({ 446 "auth_header": self.__auth_header, 447 "token_expiry": self.__token_expiry}) 448 449 def set_auth_header(self, auth_token: str, 450 token_expiry: pd.Timestamp, 451 user: dict = None) -> dict: 452 """Retrieve auth_header and token_expiry from object. 453 454 Args: 455 auth_token (str): 456 Auth token that will be set for authentication. 457 token_expiry (pd.Timestamp): 458 Token expiry time. 459 user (dict): 460 User information to be set on authetication. 461 462 Returns: 463 Return authorization header and token_expiry datetime from object. 464 """ 465 # Copy the dictonary to avoid updating the original one 466 self.__auth_header = { 467 'Authorization': auth_token} 468 self.__token_expiry = token_expiry 469 self.__user = user 470 return True 471 472 def _check_auth_header(self, auth_header: dict, 473 multipart: bool = False) -> dict: 474 """Check if auth_header is set or auth_header if provided. 475 476 Args: 477 auth_header (dict): 478 AuthHeader to substitute the microservice original 479 at the request (user impersonation). 480 multipart (dict): 481 Set if call should be made as a multipart instead of JSON. 482 483 Returns (dict): 484 Return a header dict to be used in requests. 485 486 Raises: 487 PumpWoodUnauthorized: 488 If microservice is not logged and a auth_header method 489 argument is not provided. 490 PumpWoodUnauthorized: 491 If microservice is logged and a auth_header method argument 492 is provided. 493 """ 494 if auth_header is None: 495 # Login will refresh token if it is 1h to expire, it will also 496 # check if credentials are set. 497 self.login() 498 auth_header_data = self.get_auth_header() 499 auth_header = auth_header_data['auth_header'] 500 if multipart: 501 return auth_header 502 else: 503 return self.__base_header | auth_header 504 else: 505 if self.is_credential_set(): 506 msg = ( 507 'Microservice [{object_name}] with credentials and ' 508 'auth_header was provided') 509 raise PumpWoodUnauthorized( 510 message=msg, payload={'object_name': self.name}) 511 512 # Set base header as JSON since unserialization is done using 513 # Pumpwood Communication serialization function 514 temp__auth_header = auth_header.copy() 515 if multipart: 516 return temp__auth_header 517 else: 518 return self.__base_header | temp__auth_header 519 520 @classmethod 521 def error_handler(cls, response): 522 """Handle request error. 523 524 Check if is a Json and propagate the error with 525 same type if possible. If not Json raises the content. 526 527 Args: 528 response: 529 response to be handled, it is a PumpWoodException 530 return it will raise the same exception at microservice 531 object. 532 533 Returns: 534 No return. 535 536 Raises: 537 PumpWoodOtherException: 538 If content-type is not application/json. 539 PumpWoodOtherException: 540 If content-type is application/json, but type not 541 present or not recognisable at `exceptions.exceptions_dict`. 542 Other PumpWoodException sub-types: 543 If content-type is application/json if type is present and 544 recognisable. 545 546 Example: 547 No example 548 """ 549 if not response.ok: 550 utcnow = datetime.datetime.now(datetime.UTC) 551 response_content_type = response.headers['content-type'] 552 553 # Request information 554 url = response.url 555 method = response.request.method 556 if 'application/json' not in response_content_type.lower(): 557 # Raise the exception as first in exception deep. 558 exception_dict = [{ 559 "exception_url": url, 560 "exception_method": method, 561 "exception_utcnow": utcnow.isoformat(), 562 "exception_deep": 1}] 563 raise PumpWoodOtherException( 564 message=response.text, payload={ 565 "!exception_stack!": exception_dict}) 566 567 # Build error stack 568 response_dict = cls.angular_json(response) 569 570 # Removing previous error stack 571 payload = copy.deepcopy( 572 response_dict.get("payload", {})) 573 exception_stack = copy.deepcopy( 574 payload.pop("!exception_stack!", [])) 575 576 exception_deep = len(exception_stack) 577 exception_dict = { 578 "exception_url": url, 579 "exception_method": method, 580 "exception_utcnow": utcnow.isoformat(), 581 "exception_deep": exception_deep + 1 582 } 583 exception_stack.insert(0, exception_dict) 584 payload["!exception_stack!"] = exception_stack 585 586 ################### 587 # Propagate error # 588 # get exception using 'type' key at response data and get the 589 # exception from exceptions_dict at exceptions 590 exception_message = response_dict.get("message", "") 591 exception_type = response_dict.get("type", None) 592 TempPumpwoodException = exceptions_dict.get(exception_type) 593 if TempPumpwoodException is not None: 594 raise TempPumpwoodException( 595 message=exception_message, 596 status_code=response.status_code, 597 payload=payload) 598 else: 599 # If token is invalid is at response, return a 600 # PumpWoodUnauthorized error 601 is_invalid_token = cls.is_invalid_token_response(response) 602 response_dict["!exception_stack!"] = exception_stack 603 if is_invalid_token: 604 raise PumpWoodUnauthorized( 605 message="Invalid token", payload=payload) 606 else: 607 # If the error is not mapped return a 608 # PumpWoodOtherException limiting the message size to 1k 609 # characters 610 raise PumpWoodOtherException( 611 message="Not mapped exception JSON", 612 payload=response_dict) 613 614 def _request_post_json(self, post_url: str, data: any, 615 auth_header: dict = None, 616 parameters: None | dict = None) -> any: 617 """Make a POST a request to url with data as JSON payload. 618 619 Args: 620 post_url: 621 URL to make the request, already with server url. 622 data: 623 Data to be used as Json payload. 624 parameters: 625 URL parameters. 626 auth_header: 627 AuthHeader to substitute the microservice original 628 at the request (user impersonation). 629 630 Returns: 631 Return the post response data. 632 633 Raises: 634 PumpWoodException sub-types: 635 Response is passed to error_handler. 636 """ 637 parameters = {} if parameters is None else parameters 638 639 response = None 640 request_header = self._check_auth_header(auth_header=auth_header) 641 dumped_data = pumpJsonDump(data) 642 response = requests.post( 643 url=post_url, data=dumped_data, 644 params=parameters, verify=self._verify_ssl, 645 headers=request_header, timeout=self._default_timeout) 646 647 # Retry request if token is not valid forcing token renew 648 retry_with_login = ( 649 self.is_invalid_token_response(response) and 650 auth_header is None) 651 if not retry_with_login: 652 return response 653 else: 654 # Force token refresh if Unauthorized 655 time.sleep(0.5) 656 self.login(force_refresh=True) 657 request_header = self._check_auth_header(auth_header=auth_header) 658 return requests.post( 659 url=post_url, data=dumped_data, 660 params=parameters, verify=self._verify_ssl, 661 headers=request_header, timeout=self._default_timeout) 662 663 def _request_post_multi(self, post_url: str, data: any, files: list = None, 664 auth_header: dict = None, 665 parameters: None | dict = None) -> any: 666 """Make a POST a request to url with data as multipart payload. 667 668 Args: 669 post_url: 670 URL to make the request, already with server url. 671 data: 672 Data to be used as Json payload. 673 files: 674 A dictonary with file data, files will be set on field 675 corresponding.to dictonary key. 676 `{'file1': open('file1', 'rb'), {'file2': open('file2', 'rb')}` 677 parameters: 678 URL parameters. 679 auth_header: 680 AuthHeader to substitute the microservice original 681 at the request (user impersonation). 682 683 Returns: 684 Return the post response data. 685 686 Raises: 687 PumpWoodException sub-types: 688 Response is passed to error_handler. 689 """ 690 parameters = {} if parameters is None else parameters 691 692 # Request with files are done using multipart serializing all fields 693 # as JSON 694 request_header = self._check_auth_header( 695 auth_header=auth_header, multipart=True) 696 temp_data = {'__json__': pumpJsonDump(data)} 697 698 response = requests.post( 699 url=post_url, data=temp_data, files=files, params=parameters, 700 verify=self._verify_ssl, headers=request_header, 701 timeout=self._default_timeout) 702 retry_with_login = ( 703 self.is_invalid_token_response(response) and 704 auth_header is None) 705 if not retry_with_login: 706 return response 707 else: 708 # Force token refresh if Unauthorized 709 time.sleep(0.5) 710 self.login(force_refresh=True) 711 request_header = self._check_auth_header( 712 auth_header=auth_header, multipart=True) 713 return requests.post( 714 url=post_url, data=temp_data, files=files, 715 params=parameters, verify=self._verify_ssl, 716 headers=request_header, timeout=self._default_timeout) 717 718 @classmethod 719 def _treat_response_for_file(cls, response: Response) -> dict: 720 """Return if response contain a file. 721 722 Args: 723 response (Response): 724 Response to be checked for a file content. 725 726 Returns (bool): 727 Returns if reponse has a file. 728 """ 729 headers = response.headers 730 content_disposition = headers.get('content-disposition') 731 if content_disposition is None: 732 return cls.angular_json(response) 733 else: 734 fname = re.findall("filename=(.+)", content_disposition)[0] 735 return { 736 "__file__": True, 737 "content": response.content, 738 "content-type": response.headers['content-type'], 739 "filename": fname} 740 741 @classmethod 742 def _dump_query_parameters(cls, parameters: dict) -> dict: 743 """Dump query parameters to javascript compatibility. 744 745 Args: 746 parameters (dict): 747 Parameters to be parsed to JSON. 748 749 Returns: 750 pass 751 """ 752 # If parameters are not none convert them to json before 753 # sending information on query string, 'True' is 'true' on javascript 754 # for example 755 if parameters is not None: 756 temp_parameters = copy.deepcopy(parameters) 757 for key in temp_parameters.keys(): 758 # Do not convert str to json, it put extra "" araound string 759 if type(temp_parameters[key]) is not str: 760 temp_parameters[key] = pumpJsonDump(parameters[key]) 761 return temp_parameters 762 else: 763 return None 764 765 def request_post(self, url: str, data: any, files: list = None, 766 auth_header: dict = None, 767 parameters: None | dict = {}) -> any: 768 """Make a POST a request to url with data as multipart/json payload. 769 770 Args: 771 url: 772 URL to make the request, already with server url. 773 data: 774 Data to be used as Json payload. 775 files: 776 A dictonary with file data, files will be set on field 777 corresponding.to dictonary key. 778 `{'file1': open('file1', 'rb'), {'file2': open('file2', 'rb')}` 779 parameters: 780 URL parameters. 781 auth_header: 782 AuthHeader to substitute the microservice original 783 at the request (user impersonation). 784 785 Returns: 786 Return the post response data. 787 788 Raises: 789 PumpWoodException sub-types: 790 Response is passed to error_handler. 791 """ 792 parameters = {} if parameters is None else parameters 793 794 post_url = urljoin(self.server_url, url) 795 dumped_parameters = self._dump_query_parameters(parameters=parameters) 796 response = None 797 if files is None: 798 response = self._request_post_json( 799 post_url=post_url, data=data, auth_header=auth_header, 800 parameters=dumped_parameters) 801 else: 802 response = self._request_post_multi( 803 post_url=post_url, data=data, files=files, 804 auth_header=auth_header, parameters=dumped_parameters) 805 806 # Handle errors and re-raise if Pumpwood Exceptions 807 self.error_handler(response) 808 return self._treat_response_for_file(response=response) 809 810 def request_get(self, url: str, parameters: None | dict = None, 811 auth_header: dict = None, 812 use_disk_cache: bool = False, 813 disk_cache_expire: int = None, 814 disk_cache_tag_dict: dict = None) -> Any: 815 """Make a GET a request to url with data as JSON payload. 816 817 Add the auth_header acording to login information and refresh token 818 if auth_header=None and object token is expired. 819 820 Args: 821 url (str): 822 URL to make the request. 823 parameters (dict): 824 URL parameters to make the request. 825 auth_header (dict): 826 Auth header to substitute the microservice original 827 at the request (user impersonation). 828 use_disk_cache (bool): 829 If set true, get request will use local cache to reduce 830 the requests to the backend. 831 disk_cache_expire (int): 832 Time in seconds to expire the cache, it None it will 833 use de default set be PumpwoodCache. 834 disk_cache_tag_dict (dict): 835 Dictionary to be used as a tag on get request. 836 837 Returns: 838 Return the post reponse data. 839 840 Raises: 841 PumpWoodException sub-types: 842 Raise exception if reponse is not 2XX and if 'type' key on 843 JSON payload if found at exceptions_dict. Use the same 844 exception, message and payload. 845 PumpWoodOtherException: 846 If exception type is not found or return is not a json. 847 """ 848 parameters = {} if parameters is None else parameters 849 850 request_header = self._check_auth_header(auth_header) 851 # If is set to use diskcache, it will create a hash cash using 852 # the query paramerers, url and user access token. The 853 # hash will be used as index, not exposing the token at cache 854 # database 855 hash_dict = None 856 if use_disk_cache: 857 hash_dict = RequestGetCacheHash( 858 context='pumpwood_communication-request_get', 859 authorization=request_header['Authorization'], 860 parameters=parameters, 861 url=url) 862 863 cache_results = default_cache.get(hash_dict=hash_dict) 864 if cache_results is not None: 865 msg = "get from cache url[{url}]".format(url=url) 866 logger.info(msg) 867 return cache_results 868 869 dumped_parameters = self._dump_query_parameters(parameters=parameters) 870 get_url = urljoin(self.server_url, url) 871 response = requests.get( 872 get_url, verify=self._verify_ssl, headers=request_header, 873 params=dumped_parameters, timeout=self._default_timeout) 874 875 # If token is expired, refresh it 876 retry_with_login = ( 877 self.is_invalid_token_response(response) and 878 auth_header is None) 879 if retry_with_login: 880 time.sleep(0.5) 881 self.login(force_refresh=True) 882 request_header = self._check_auth_header(auth_header=auth_header) 883 response = requests.get( 884 get_url, verify=self._verify_ssl, headers=request_header, 885 params=dumped_parameters, timeout=self._default_timeout) 886 887 # Re-raise Pumpwood exceptions 888 self.error_handler(response=response) 889 results = self._treat_response_for_file(response=response) 890 891 # If is set to use cache for this calls, set the local cache 892 if use_disk_cache and not results.get('__file__', False): 893 default_cache.set( 894 hash_dict=hash_dict, value=results, 895 expire=disk_cache_expire, 896 tag_dict=disk_cache_tag_dict) 897 return results 898 899 def request_delete(self, url, parameters: dict = None, 900 auth_header: dict = None): 901 """Make a DELETE a request to url with data as Json payload. 902 903 Args: 904 url: 905 Url to make the request. 906 parameters: 907 Dictionary with Urls parameters. 908 auth_header: 909 Auth header to substitute the microservice original 910 at the request (user impersonation). 911 912 Returns: 913 Return the delete reponse payload. 914 915 Raises: 916 PumpWoodException sub-types: 917 Raise exception if reponse is not 2XX and if 'type' key on 918 JSON payload if found at exceptions_dict. Use the same 919 exception, message and payload. 920 PumpWoodOtherException: 921 If exception type is not found or return is not a json. 922 """ 923 request_header = self._check_auth_header(auth_header) 924 dumped_parameters = self._dump_query_parameters(parameters=parameters) 925 926 post_url = self.server_url + url 927 response = requests.delete( 928 post_url, verify=self._verify_ssl, headers=request_header, 929 params=dumped_parameters, timeout=self._default_timeout) 930 931 # Retry request if token is not valid forcing token renew 932 retry_with_login = ( 933 self.is_invalid_token_response(response) and 934 auth_header is None) 935 if retry_with_login: 936 time.sleep(0.5) 937 self.login(force_refresh=True) 938 request_header = self._check_auth_header(auth_header=auth_header) 939 response = requests.delete( 940 post_url, verify=self._verify_ssl, headers=request_header, 941 params=dumped_parameters, timeout=self._default_timeout) 942 943 # Re-raise Pumpwood Exceptions 944 self.error_handler(response) 945 return self.angular_json(response)
Base class for Pumpwood MicroService.
Enviroment variables can be used to set MicroService parameters:
- PUMPWOOD_COMUNICATION__DEFAULT_TIMEOUT: Default requests timeout in seconds.
- PUMPWOOD_COMUNICATION__DEBUG: If object will be initiated using debug parameter. It will have more verbosity and login at each request. Options 'TRUE', 'FALSE'.
- PUMPWOOD_COMUNICATION__VERIFY_SSL: If requests will validate SSL certificate.
67 def __init__(self, name: str = None, server_url: str = None, 68 username: str = None, password: str = None, 69 verify_ssl: bool = True, debug: bool = None, 70 default_timeout: int = None, **kwargs): 71 """Create new PumpWoodMicroService object. 72 73 Creates a new microservice object. If just name is passed, object must 74 be initiate after with init() method. 75 76 Args: 77 name: 78 Name of the microservice, helps when exceptions 79 are raised. 80 server_url: 81 URL of the server that will be connected. 82 username: 83 Username that will be logged on. 84 password: 85 Variable to be converted to JSON and posted along 86 with the request. 87 verify_ssl: 88 Set if microservice will verify SSL certificate. 89 debug: 90 If microservice will be used as debug mode. This will obrigate 91 auth token refresh for each call. 92 default_timeout: 93 Default timeout for Pumpwood calls. 94 **kwargs: 95 Other parameters used for compatibility between versions. 96 97 Returns: 98 PumpWoodMicroService: New PumpWoodMicroService object 99 100 Raises: 101 No particular Raises. 102 """ 103 # Create attributes to be set at init function 104 self.name = None 105 """Name of the microservice instance.""" 106 self.server_url = None 107 """Pumpwood server URL.""" 108 self._default_timeout: int = None 109 """Default timeout for Pumpwood requests.""" 110 self._debug: bool = None 111 """Name of the microservice instance.""" 112 self._verify_ssl: bool = None 113 """If microservice should check the certificate.""" 114 self._is_mfa_login: bool = None 115 """Set if is MFA login.""" 116 self.__headers: dict = None 117 """Headers to be used on the requests.""" 118 self.__user: dict = None 119 """Information of the logged user.""" 120 self.__auth_header: dict = None 121 """Authenticated auth header.""" 122 self.__token_expiry: pd.Timedelta = None 123 """Expirity datetime of the authetication token.""" 124 self.__username: str = None 125 """Username associated with microservice.""" 126 self.__password: str = None 127 """Password associated with microservice.""" 128 self.init( 129 name=name, server_url=server_url, 130 username=username, password=password, 131 verify_ssl=verify_ssl, debug=debug, 132 default_timeout=default_timeout)
Create new PumpWoodMicroService object.
Creates a new microservice object. If just name is passed, object must be initiate after with init() method.
Arguments:
- name: Name of the microservice, helps when exceptions are raised.
- server_url: URL of the server that will be connected.
- username: Username that will be logged on.
- password: Variable to be converted to JSON and posted along with the request.
- verify_ssl: Set if microservice will verify SSL certificate.
- debug: If microservice will be used as debug mode. This will obrigate auth token refresh for each call.
- default_timeout: Default timeout for Pumpwood calls.
- **kwargs: Other parameters used for compatibility between versions.
Returns:
PumpWoodMicroService: New PumpWoodMicroService object
Raises:
- No particular Raises.
134 def init(self, name: str = None, server_url: str = None, 135 username: str = None, password: str = None, 136 verify_ssl: bool = True, debug: bool = None, 137 default_timeout: int = None, **kwargs): 138 """Lazzy initialization of the MicroService of object. 139 140 This function might be usefull to use the object as a singleton at 141 the backends. Using this function it is possible to instanciate an 142 empty object and them set the attributes latter at the systems. 143 144 Args: 145 name: 146 Name of the microservice, helps when exceptions 147 are raised. 148 server_url: 149 URL of the server that will be connected. 150 username: 151 Username that will be logged on. 152 password: 153 Variable to be converted to JSON and posted along 154 with the request. 155 verify_ssl: 156 Set if microservice will verify SSL certificate. 157 debug: 158 If microservice will be used as debug mode. This will obrigate 159 auth token refresh for each call. 160 default_timeout: 161 Default timeout for Pumpwood calls. 162 **kwargs: 163 Other parameters used for compatibility between versions. 164 165 Returns: 166 No return 167 168 Raises: 169 No particular Raises 170 """ 171 self.name = name 172 """Name of the microservice instance.""" 173 self.server_url = self._adjust_server_url(server_url) 174 """Pumpwood server URL.""" 175 176 # Set parameter using arguments or enviroment variables 177 if default_timeout is None: 178 self._default_timeout = int(os.getenv( 179 'PUMPWOOD_COMUNICATION__DEFAULT_TIMEOUT', 60)) 180 else: 181 self._default_timeout = default_timeout 182 183 if debug is None: 184 self._debug = os.getenv( 185 'PUMPWOOD_COMUNICATION__DEBUG', 'FALSE') == 'TRUE' 186 else: 187 self._debug = debug 188 189 if verify_ssl is None: 190 self._verify_ssl = os.getenv( 191 'PUMPWOOD_COMUNICATION__VERIFY_SSL', 'TRUE') == 'TRUE' 192 else: 193 self._verify_ssl = verify_ssl 194 195 self._is_mfa_login = False 196 self.__headers = None 197 self.__user = None 198 self.__auth_header = None 199 self.__token_expiry = None 200 self.__username = username 201 self.__password = password
Lazzy initialization of the MicroService of object.
This function might be usefull to use the object as a singleton at the backends. Using this function it is possible to instanciate an empty object and them set the attributes latter at the systems.
Arguments:
- name: Name of the microservice, helps when exceptions are raised.
- server_url: URL of the server that will be connected.
- username: Username that will be logged on.
- password: Variable to be converted to JSON and posted along with the request.
- verify_ssl: Set if microservice will verify SSL certificate.
- debug: If microservice will be used as debug mode. This will obrigate auth token refresh for each call.
- default_timeout: Default timeout for Pumpwood calls.
- **kwargs: Other parameters used for compatibility between versions.
Returns:
No return
Raises:
- No particular Raises
203 @staticmethod 204 def angular_json(request_result) -> Any: 205 r"""Convert text to Json removing any XSSI at the beging of JSON. 206 207 Some backends add `)]}',\n` at the beginning of the JSON data to 208 prevent injection of functions. This function remove this characters 209 if present. 210 211 Args: 212 request_result: 213 JSON request to be converted. 214 215 Returns: 216 No return 217 218 Raises: 219 PumpWoodJSONLoadError: 220 If it is not possible to load JSON from request data. 221 """ 222 if request_result.text == '': 223 return None 224 225 string_start = ")]}',\n" 226 try: 227 if request_result.text[:6] == string_start: 228 return (orjson.loads(request_result.text[6:])) 229 else: 230 return (orjson.loads(request_result.text)) 231 except Exception: 232 msg = "Can not decode to Json" 233 raise PumpWoodJSONLoadError( 234 message=msg, payload={"request_data": request_result.text})
Convert text to Json removing any XSSI at the beging of JSON.
Some backends add )]}',\n at the beginning of the JSON data to
prevent injection of functions. This function remove this characters
if present.
Arguments:
- request_result: JSON request to be converted.
Returns:
No return
Raises:
- PumpWoodJSONLoadError: If it is not possible to load JSON from request data.
236 def time_to_expiry(self) -> pd.Timedelta: 237 """Return time to token expiry. 238 239 Args: 240 No Args. 241 242 Returns: 243 Return time until token expiration. 244 """ 245 if self.__token_expiry is None: 246 return None 247 248 now_datetime = pd.to_datetime( 249 datetime.datetime.now(datetime.UTC), utc=True) 250 time_to_expiry = self.__token_expiry - now_datetime 251 return time_to_expiry
Return time to token expiry.
Arguments:
- No Args.
Returns:
Return time until token expiration.
253 def is_credential_set(self) -> bool: 254 """Check if username and password are set on object. 255 256 Args: 257 No Args. 258 259 Returns: 260 True if usename and password were set during object creation or 261 later with init function. 262 """ 263 is_username_not_none = self.__username is not None 264 is_password_not_none = self.__password is not None 265 return is_username_not_none and is_password_not_none
Check if username and password are set on object.
Arguments:
- No Args.
Returns:
True if usename and password were set during object creation or later with init function.
267 @classmethod 268 def is_invalid_token_response(cls, response: Response) -> bool: 269 """Check if reponse has invalid token error. 270 271 Args: 272 response: 273 Request reponse to check for invalid token. 274 275 Returns: 276 Return True if response has an invalid token status. 277 """ 278 if response.status_code == 401: 279 return True 280 return False
Check if reponse has invalid token error.
Arguments:
- response: Request reponse to check for invalid token.
Returns:
Return True if response has an invalid token status.
330 def confirm_mfa_code(self, mfa_login_data: dict) -> dict: 331 """Ask user to confirm MFA code to login. 332 333 Open an input interface at terminal for user to validate MFA token. 334 335 Args: 336 mfa_login_data: 337 Result from login request with 'mfa_token' 338 as key. 339 340 Returns: 341 Return login returned with MFA confimation. 342 343 Raise: 344 Raise error if reponse is not valid using error_handler. 345 """ 346 code = input("## Please enter MFA code: ") 347 url = urljoin( 348 self.server_url, 'rest/registration/mfa-validate-code/') 349 mfa_response = requests.post(url, headers={ 350 "X-PUMPWOOD-MFA-Autorization": mfa_login_data['mfa_token']}, 351 json={"mfa_code": code}, timeout=self._default_timeout) 352 self.error_handler(mfa_response) 353 354 # Set _is_mfa_login true to indicate that login required MFA 355 self._is_mfa_login = True 356 return self.angular_json(mfa_response)
Ask user to confirm MFA code to login.
Open an input interface at terminal for user to validate MFA token.
Arguments:
- mfa_login_data: Result from login request with 'mfa_token' as key.
Returns:
Return login returned with MFA confimation.
Raises:
- Raise error if reponse is not valid using error_handler.
358 def login(self, force_refresh: bool = False) -> None: 359 """Log microservice in using username and password provided. 360 361 Args: 362 force_refresh (bool): 363 Force token refresh despise still valid 364 according to self.__token_expiry. 365 366 Returns: 367 No return 368 369 Raises: 370 Exception: 371 If login response has status diferent from 200. 372 """ 373 if not self.is_credential_set(): 374 raise PumpWoodUnauthorized( 375 message="Microservice username or/and password not set") 376 377 # Check if expiry time is 1h from now 378 refresh_expiry = False 379 if self.__token_expiry is None: 380 refresh_expiry = True 381 else: 382 time_to_expiry = self.time_to_expiry() 383 if time_to_expiry < datetime.timedelta(hours=1): 384 refresh_expiry = True 385 386 # When if debug always refresh token 387 if refresh_expiry or force_refresh or self._debug: 388 login_data = self._login_resquest() 389 if 'mfa_token' in login_data.keys(): 390 login_data = self.confirm_mfa_code( 391 mfa_login_data=login_data) 392 393 self.set_auth_header( 394 auth_token='Token ' + login_data['token'], 395 token_expiry=pd.to_datetime(login_data['expiry']), 396 user=login_data["user"]) 397 else: 398 # Token is not expired or envicted, them keep same token 399 return None
Log microservice in using username and password provided.
Arguments:
- force_refresh (bool): Force token refresh despise still valid according to self.__token_expiry.
Returns:
No return
Raises:
- Exception: If login response has status diferent from 200.
401 def logout(self, auth_header: dict = None) -> bool: 402 """Logout token. 403 404 Args: 405 auth_header: 406 Authentication header. 407 408 Returns: 409 True if logout was ok. 410 """ 411 resp = self.request_post( 412 url='rest/registration/logout/', 413 data={}, auth_header=auth_header) 414 # Set expiry to None to envict the token 415 self.__token_expiry = None 416 return resp is None
Logout token.
Arguments:
- auth_header: Authentication header.
Returns:
True if logout was ok.
418 def logout_all(self, auth_header: dict = None) -> bool: 419 """Logout all tokens from user. 420 421 Args: 422 auth_header (dict): 423 Authentication header. 424 425 Returns: 426 True if logout all was ok. 427 """ 428 resp = self.request_post( 429 url='rest/registration/logoutall/', 430 data={}, auth_header=auth_header) 431 # Set expiry to None to envict the token 432 self.__token_expiry = None 433 return resp is None
Logout all tokens from user.
Arguments:
- auth_header (dict): Authentication header.
Returns:
True if logout all was ok.
435 def get_auth_header(self) -> dict: 436 """Retrieve auth_header and token_expiry from object. 437 438 Args: 439 No Args. 440 441 Returns: 442 Return authorization header and token_expiry datetime from object. 443 """ 444 # Copy the dictonary to avoid updating the original one 445 return copy.deepcopy({ 446 "auth_header": self.__auth_header, 447 "token_expiry": self.__token_expiry})
Retrieve auth_header and token_expiry from object.
Arguments:
- No Args.
Returns:
Return authorization header and token_expiry datetime from object.
449 def set_auth_header(self, auth_token: str, 450 token_expiry: pd.Timestamp, 451 user: dict = None) -> dict: 452 """Retrieve auth_header and token_expiry from object. 453 454 Args: 455 auth_token (str): 456 Auth token that will be set for authentication. 457 token_expiry (pd.Timestamp): 458 Token expiry time. 459 user (dict): 460 User information to be set on authetication. 461 462 Returns: 463 Return authorization header and token_expiry datetime from object. 464 """ 465 # Copy the dictonary to avoid updating the original one 466 self.__auth_header = { 467 'Authorization': auth_token} 468 self.__token_expiry = token_expiry 469 self.__user = user 470 return True
Retrieve auth_header and token_expiry from object.
Arguments:
- auth_token (str): Auth token that will be set for authentication.
- token_expiry (pd.Timestamp): Token expiry time.
- user (dict): User information to be set on authetication.
Returns:
Return authorization header and token_expiry datetime from object.
520 @classmethod 521 def error_handler(cls, response): 522 """Handle request error. 523 524 Check if is a Json and propagate the error with 525 same type if possible. If not Json raises the content. 526 527 Args: 528 response: 529 response to be handled, it is a PumpWoodException 530 return it will raise the same exception at microservice 531 object. 532 533 Returns: 534 No return. 535 536 Raises: 537 PumpWoodOtherException: 538 If content-type is not application/json. 539 PumpWoodOtherException: 540 If content-type is application/json, but type not 541 present or not recognisable at `exceptions.exceptions_dict`. 542 Other PumpWoodException sub-types: 543 If content-type is application/json if type is present and 544 recognisable. 545 546 Example: 547 No example 548 """ 549 if not response.ok: 550 utcnow = datetime.datetime.now(datetime.UTC) 551 response_content_type = response.headers['content-type'] 552 553 # Request information 554 url = response.url 555 method = response.request.method 556 if 'application/json' not in response_content_type.lower(): 557 # Raise the exception as first in exception deep. 558 exception_dict = [{ 559 "exception_url": url, 560 "exception_method": method, 561 "exception_utcnow": utcnow.isoformat(), 562 "exception_deep": 1}] 563 raise PumpWoodOtherException( 564 message=response.text, payload={ 565 "!exception_stack!": exception_dict}) 566 567 # Build error stack 568 response_dict = cls.angular_json(response) 569 570 # Removing previous error stack 571 payload = copy.deepcopy( 572 response_dict.get("payload", {})) 573 exception_stack = copy.deepcopy( 574 payload.pop("!exception_stack!", [])) 575 576 exception_deep = len(exception_stack) 577 exception_dict = { 578 "exception_url": url, 579 "exception_method": method, 580 "exception_utcnow": utcnow.isoformat(), 581 "exception_deep": exception_deep + 1 582 } 583 exception_stack.insert(0, exception_dict) 584 payload["!exception_stack!"] = exception_stack 585 586 ################### 587 # Propagate error # 588 # get exception using 'type' key at response data and get the 589 # exception from exceptions_dict at exceptions 590 exception_message = response_dict.get("message", "") 591 exception_type = response_dict.get("type", None) 592 TempPumpwoodException = exceptions_dict.get(exception_type) 593 if TempPumpwoodException is not None: 594 raise TempPumpwoodException( 595 message=exception_message, 596 status_code=response.status_code, 597 payload=payload) 598 else: 599 # If token is invalid is at response, return a 600 # PumpWoodUnauthorized error 601 is_invalid_token = cls.is_invalid_token_response(response) 602 response_dict["!exception_stack!"] = exception_stack 603 if is_invalid_token: 604 raise PumpWoodUnauthorized( 605 message="Invalid token", payload=payload) 606 else: 607 # If the error is not mapped return a 608 # PumpWoodOtherException limiting the message size to 1k 609 # characters 610 raise PumpWoodOtherException( 611 message="Not mapped exception JSON", 612 payload=response_dict)
Handle request error.
Check if is a Json and propagate the error with same type if possible. If not Json raises the content.
Arguments:
- response: response to be handled, it is a PumpWoodException return it will raise the same exception at microservice object.
Returns:
No return.
Raises:
- PumpWoodOtherException: If content-type is not application/json.
- PumpWoodOtherException: If content-type is application/json, but type not
present or not recognisable at
exceptions.exceptions_dict. - Other PumpWoodException sub-types: If content-type is application/json if type is present and recognisable.
Example:
No example
765 def request_post(self, url: str, data: any, files: list = None, 766 auth_header: dict = None, 767 parameters: None | dict = {}) -> any: 768 """Make a POST a request to url with data as multipart/json payload. 769 770 Args: 771 url: 772 URL to make the request, already with server url. 773 data: 774 Data to be used as Json payload. 775 files: 776 A dictonary with file data, files will be set on field 777 corresponding.to dictonary key. 778 `{'file1': open('file1', 'rb'), {'file2': open('file2', 'rb')}` 779 parameters: 780 URL parameters. 781 auth_header: 782 AuthHeader to substitute the microservice original 783 at the request (user impersonation). 784 785 Returns: 786 Return the post response data. 787 788 Raises: 789 PumpWoodException sub-types: 790 Response is passed to error_handler. 791 """ 792 parameters = {} if parameters is None else parameters 793 794 post_url = urljoin(self.server_url, url) 795 dumped_parameters = self._dump_query_parameters(parameters=parameters) 796 response = None 797 if files is None: 798 response = self._request_post_json( 799 post_url=post_url, data=data, auth_header=auth_header, 800 parameters=dumped_parameters) 801 else: 802 response = self._request_post_multi( 803 post_url=post_url, data=data, files=files, 804 auth_header=auth_header, parameters=dumped_parameters) 805 806 # Handle errors and re-raise if Pumpwood Exceptions 807 self.error_handler(response) 808 return self._treat_response_for_file(response=response)
Make a POST a request to url with data as multipart/json payload.
Arguments:
- url: URL to make the request, already with server url.
- data: Data to be used as Json payload.
- files: A dictonary with file data, files will be set on field
corresponding.to dictonary key.
{'file1': open('file1', 'rb'), {'file2': open('file2', 'rb')} - parameters: URL parameters.
- auth_header: AuthHeader to substitute the microservice original at the request (user impersonation).
Returns:
Return the post response data.
Raises:
- PumpWoodException sub-types: Response is passed to error_handler.
810 def request_get(self, url: str, parameters: None | dict = None, 811 auth_header: dict = None, 812 use_disk_cache: bool = False, 813 disk_cache_expire: int = None, 814 disk_cache_tag_dict: dict = None) -> Any: 815 """Make a GET a request to url with data as JSON payload. 816 817 Add the auth_header acording to login information and refresh token 818 if auth_header=None and object token is expired. 819 820 Args: 821 url (str): 822 URL to make the request. 823 parameters (dict): 824 URL parameters to make the request. 825 auth_header (dict): 826 Auth header to substitute the microservice original 827 at the request (user impersonation). 828 use_disk_cache (bool): 829 If set true, get request will use local cache to reduce 830 the requests to the backend. 831 disk_cache_expire (int): 832 Time in seconds to expire the cache, it None it will 833 use de default set be PumpwoodCache. 834 disk_cache_tag_dict (dict): 835 Dictionary to be used as a tag on get request. 836 837 Returns: 838 Return the post reponse data. 839 840 Raises: 841 PumpWoodException sub-types: 842 Raise exception if reponse is not 2XX and if 'type' key on 843 JSON payload if found at exceptions_dict. Use the same 844 exception, message and payload. 845 PumpWoodOtherException: 846 If exception type is not found or return is not a json. 847 """ 848 parameters = {} if parameters is None else parameters 849 850 request_header = self._check_auth_header(auth_header) 851 # If is set to use diskcache, it will create a hash cash using 852 # the query paramerers, url and user access token. The 853 # hash will be used as index, not exposing the token at cache 854 # database 855 hash_dict = None 856 if use_disk_cache: 857 hash_dict = RequestGetCacheHash( 858 context='pumpwood_communication-request_get', 859 authorization=request_header['Authorization'], 860 parameters=parameters, 861 url=url) 862 863 cache_results = default_cache.get(hash_dict=hash_dict) 864 if cache_results is not None: 865 msg = "get from cache url[{url}]".format(url=url) 866 logger.info(msg) 867 return cache_results 868 869 dumped_parameters = self._dump_query_parameters(parameters=parameters) 870 get_url = urljoin(self.server_url, url) 871 response = requests.get( 872 get_url, verify=self._verify_ssl, headers=request_header, 873 params=dumped_parameters, timeout=self._default_timeout) 874 875 # If token is expired, refresh it 876 retry_with_login = ( 877 self.is_invalid_token_response(response) and 878 auth_header is None) 879 if retry_with_login: 880 time.sleep(0.5) 881 self.login(force_refresh=True) 882 request_header = self._check_auth_header(auth_header=auth_header) 883 response = requests.get( 884 get_url, verify=self._verify_ssl, headers=request_header, 885 params=dumped_parameters, timeout=self._default_timeout) 886 887 # Re-raise Pumpwood exceptions 888 self.error_handler(response=response) 889 results = self._treat_response_for_file(response=response) 890 891 # If is set to use cache for this calls, set the local cache 892 if use_disk_cache and not results.get('__file__', False): 893 default_cache.set( 894 hash_dict=hash_dict, value=results, 895 expire=disk_cache_expire, 896 tag_dict=disk_cache_tag_dict) 897 return results
Make a GET a request to url with data as JSON payload.
Add the auth_header acording to login information and refresh token if auth_header=None and object token is expired.
Arguments:
- url (str): URL to make the request.
- parameters (dict): URL parameters to make the request.
- auth_header (dict): Auth header to substitute the microservice original at the request (user impersonation).
- use_disk_cache (bool): If set true, get request will use local cache to reduce the requests to the backend.
- disk_cache_expire (int): Time in seconds to expire the cache, it None it will use de default set be PumpwoodCache.
- disk_cache_tag_dict (dict): Dictionary to be used as a tag on get request.
Returns:
Return the post reponse data.
Raises:
- PumpWoodException sub-types: Raise exception if reponse is not 2XX and if 'type' key on JSON payload if found at exceptions_dict. Use the same exception, message and payload.
- PumpWoodOtherException: If exception type is not found or return is not a json.
899 def request_delete(self, url, parameters: dict = None, 900 auth_header: dict = None): 901 """Make a DELETE a request to url with data as Json payload. 902 903 Args: 904 url: 905 Url to make the request. 906 parameters: 907 Dictionary with Urls parameters. 908 auth_header: 909 Auth header to substitute the microservice original 910 at the request (user impersonation). 911 912 Returns: 913 Return the delete reponse payload. 914 915 Raises: 916 PumpWoodException sub-types: 917 Raise exception if reponse is not 2XX and if 'type' key on 918 JSON payload if found at exceptions_dict. Use the same 919 exception, message and payload. 920 PumpWoodOtherException: 921 If exception type is not found or return is not a json. 922 """ 923 request_header = self._check_auth_header(auth_header) 924 dumped_parameters = self._dump_query_parameters(parameters=parameters) 925 926 post_url = self.server_url + url 927 response = requests.delete( 928 post_url, verify=self._verify_ssl, headers=request_header, 929 params=dumped_parameters, timeout=self._default_timeout) 930 931 # Retry request if token is not valid forcing token renew 932 retry_with_login = ( 933 self.is_invalid_token_response(response) and 934 auth_header is None) 935 if retry_with_login: 936 time.sleep(0.5) 937 self.login(force_refresh=True) 938 request_header = self._check_auth_header(auth_header=auth_header) 939 response = requests.delete( 940 post_url, verify=self._verify_ssl, headers=request_header, 941 params=dumped_parameters, timeout=self._default_timeout) 942 943 # Re-raise Pumpwood Exceptions 944 self.error_handler(response) 945 return self.angular_json(response)
Make a DELETE a request to url with data as Json payload.
Arguments:
- url: Url to make the request.
- parameters: Dictionary with Urls parameters.
- auth_header: Auth header to substitute the microservice original at the request (user impersonation).
Returns:
Return the delete reponse payload.
Raises:
- PumpWoodException sub-types: Raise exception if reponse is not 2XX and if 'type' key on JSON payload if found at exceptions_dict. Use the same exception, message and payload.
- PumpWoodOtherException: If exception type is not found or return is not a json.