ms/py-executor/resource_resolution/http/client.py

   1 """Copyright 2020 Deutsche Telekom.
   2
   3 Licensed under the Apache License, Version 2.0 (the "License");
   4 you may not use this file except in compliance with the License.
   5 You may obtain a copy of the License at
   6
   7     http://www.apache.org/licenses/LICENSE-2.0
   8
   9 Unless required by applicable law or agreed to in writing, software
  10 distributed under the License is distributed on an "AS IS" BASIS,
  11 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 See the License for the specific language governing permissions and
  13 limitations under the License.
  14 """
  15 from typing import Optional, Tuple
  16
  17 from requests import Session, request, Request, Response, PreparedRequest
  18
  19
  20 class Client:
  21     """HTTP client class."""
  22
  23     API_VERSION = "v1"
  24
  25     def __init__(
  26         self, server_address: str, server_port: int, auth_user: str = None, auth_pass: str = None, use_ssl: bool = False
  27     ) -> None:
  28         """HTTP client class initialization.
  29
  30         Args:
  31             server_address (str): HTTP server address
  32             server_port (int): HTTP server port
  33             auth_user (str, optional): Username used for authorization. Defaults to None.
  34             auth_pass (str, optional): Password used for authorization. Defaults to None.
  35             use_ssl (bool, optional): Determines if secure connection has to be used. Defaults to False.
  36         """
  37         self.server_address: str = server_address
  38         self.server_port: int = server_port
  39         self.use_ssl: bool = use_ssl
  40
  41         self.auth_user: str = auth_user
  42         self.auth_pass: str = auth_pass
  43
  44     @property
  45     def auth(self) -> Optional[Tuple[str, str]]:
  46         """Authorization data tuple or None.
  47
  48         Returns None if not both auth_user and auth_pass values are set.
  49
  50         Returns:
  51             Optional[Tuple[str, str]]: Authorization tuple (auth_user, auth_pass) or None
  52         """
  53         if all([self.auth_user, self.auth_pass]):
  54             return (self.auth_user, self.auth_pass)
  55         return None
  56
  57     @property
  58     def protocol(self) -> str:
  59         """Protocol which is going to be used for request call.
  60
  61         Returns:
  62             str: http or https
  63         """
  64         if self.use_ssl:
  65             return "https"
  66         return "http"
  67
  68     @property
  69     def url(self) -> str:
  70         """Url to call requests.
  71
  72         Returns:
  73             str: Url string
  74         """
  75         return f"{self.protocol}://{self.server_address}:{self.server_port}/api/{self.API_VERSION}"
  76
  77     def send_request(self, method: str, endpoint: str, **kwargs) -> Response:
  78         """Send request to server.
  79
  80         Send request with `method` method to server. Pass any additional values as **kwargs.
  81
  82         Args:
  83             method (str): HTTP method
  84             endpoint (str): Endpoint to call a request
  85
  86         Raises:
  87             requests.HTTPError: An HTTP error occurred.
  88
  89         Returns:
  90             Response: `requests.Response` object.
  91         """
  92         response: Response = request(
  93             method=method, url=f"{self.url}/{endpoint}", verify=False, auth=self.auth, **kwargs
  94         )
  95         response.raise_for_status()
  96         return response