summaryrefslogtreecommitdiffstats
path: root/ms/py-executor/resource_resolution/http/client.py
blob: 8bb1e1be4a319ae4eaca057c23d182ded136931e (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
"""Copyright 2020 Deutsche Telekom.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
from typing import Optional, Tuple

from requests import Session, request, Request, Response, PreparedRequest


class Client:
    """HTTP client class."""

    API_VERSION = "v1"

    def __init__(
        self, server_address: str, server_port: int, auth_user: str = None, auth_pass: str = None, use_ssl: bool = False
    ) -> None:
        """HTTP client class initialization.
        
        Args:
            server_address (str): HTTP server address
            server_port (int): HTTP server port
            auth_user (str, optional): Username used for authorization. Defaults to None.
            auth_pass (str, optional): Password used for authorization. Defaults to None.
            use_ssl (bool, optional): Determines if secure connection has to be used. Defaults to False.
        """
        self.server_address: str = server_address
        self.server_port: int = server_port
        self.use_ssl: bool = use_ssl

        self.auth_user: str = auth_user
        self.auth_pass: str = auth_pass

    @property
    def auth(self) -> Optional[Tuple[str, str]]:
        """Authorization data tuple or None.

        Returns None if not both auth_user and auth_pass values are set.
        
        Returns:
            Optional[Tuple[str, str]]: Authorization tuple (auth_user, auth_pass) or None
        """
        if all([self.auth_user, self.auth_pass]):
            return (self.auth_user, self.auth_pass)
        return None

    @property
    def protocol(self) -> str:
        """Protocol which is going to be used for request call.
        
        Returns:
            str: http or https
        """
        if self.use_ssl:
            return "https"
        return "http"

    @property
    def url(self) -> str:
        """Url to call requests.
        
        Returns:
            str: Url string
        """
        return f"{self.protocol}://{self.server_address}:{self.server_port}/api/{self.API_VERSION}"

    def send_request(self, method: str, endpoint: str, **kwargs) -> Response:
        """Send request to server.

        Send request with `method` method to server. Pass any additional values as **kwargs.

        Args:
            method (str): HTTP method
            endpoint (str): Endpoint to call a request

        Raises:
            requests.HTTPError: An HTTP error occurred.

        Returns:
            Response: `requests.Response` object.
        """
        response: Response = request(
            method=method, url=f"{self.url}/{endpoint}", verify=False, auth=self.auth, **kwargs
        )
        response.raise_for_status()
        return response