Skip to content

ark_client

ArkClient

Source code in ark_sdk_python/common/ark_client.py 
 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
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
class ArkClient:
    def __init__(
        self,
        base_url: Optional[str] = None,
        token: Optional[str] = None,
        token_type: str = 'Bearer',
        cookies: Optional[List] = None,
        auth_header_name: str = 'Authorization',
        auth: Optional[Tuple[str, str]] = None,
        cookie_jar: Optional[RequestsCookieJar] = None,
        verify: Optional[Union[str, bool]] = None,
        refresh_connection_callback: Optional[Callable[['ArkClient'], None]] = None,
    ) -> None:
        from fake_useragent import UserAgent

        self.__session = Session()
        self.__base_url = base_url
        self.__token = token
        self.__token_type = token_type
        self.__auth_header_name = auth_header_name
        self.__refresh_connection_callback = refresh_connection_callback
        if self.__base_url and not self.__base_url.startswith('https://'):
            self.__base_url = f'https://{self.__base_url}'
        if auth:
            self.__session.auth = auth
        self.update_token(token)
        self.update_cookies(cookies, cookie_jar)
        if verify is None:
            if ArkSystemConfig.trusted_certificate() is not None:
                verify = ArkSystemConfig.trusted_certificate()
            else:
                verify = ArkSystemConfig.is_verifiying_certificates()
        self.__session.verify = verify
        self.__session.headers['User-Agent'] = UserAgent(browsers=['chrome']).googlechrome

    @property
    def base_url(self) -> Optional[str]:
        return self.__base_url

    @property
    def session(self) -> Session:
        return self.__session

    @property
    def session_token(self) -> Optional[str]:
        return self.__token

    @property
    def refresh_connection_callback(self) -> Optional[Callable[['ArkClient'], None]]:
        return self.__refresh_connection_callback

    def add_header(self, key: str, value: str) -> None:
        self.__session.headers.update({key: value})

    def add_headers(self, headers: Dict[str, str]) -> None:
        self.__session.headers.update(headers)

    def add_cookie(self, key: str, value: str) -> None:
        self.__session.cookies[key] = value

    def generic_http_method_request(self, method: str, route: str, **kwargs) -> Response:
        url = route
        if self.__base_url:
            url = f'{self.__base_url}'
            if route and route != '':
                if self.__base_url.endswith('/') or route.startswith('/'):
                    url = f'{self.__base_url}{route}'
                else:
                    url = f'{self.__base_url}/{route}'
        http_method = getattr(self.__session, method)
        response: Response = http_method(url, **kwargs)
        return response

    def get(self, route: str, **kwargs) -> Response:
        """
        Performs a GET request with the session details and given headers and tokens.

        Args:
            route (str): _description_

        Returns:
            Response: _description_
        """
        return self.generic_http_method_request('get', route, **kwargs)

    def post(self, route: str, **kwargs) -> Response:
        """
        Performs a POST request with the session details and given headers and tokens.

        Args:
            route (str): _description_

        Returns:
            Response: _description_
        """
        return self.generic_http_method_request('post', route, **kwargs)

    def put(self, route: str, **kwargs) -> Response:
        """
        Performs a PUT request with the session details and given headers and tokens.

        Args:
            route (str): _description_

        Returns:
            Response: _description_
        """
        return self.generic_http_method_request('put', route, **kwargs)

    def delete(self, route: str, **kwargs) -> Response:
        """
        Performs a DELETE request with the session details and given headers and tokens.

        Args:
            route (str): _description_

        Returns:
            Response: _description_
        """
        return self.generic_http_method_request('delete', route, **kwargs)

    def patch(self, route: str, **kwargs) -> Response:
        """
        Performs a PATCH request with the session details and given headers and tokens.

        Args:
            route (str): _description_

        Returns:
            Response: _description_
        """
        return self.generic_http_method_request('patch', route, **kwargs)

    def options(self, route: str, **kwargs) -> Response:
        """
        Performs a OPTIONS request with the session details and given headers and tokens.

        Args:
            route (str): _description_

        Returns:
            Response: _description_
        """
        return self.generic_http_method_request('options', route, **kwargs)

    def update_token(self, token: Optional[str] = None) -> None:
        """
        Updates a session token.

        Args:
            token (Optional[str], optional): _description_. Defaults to None.
        """
        self.__token = token
        if token:
            if self.__token_type == 'Basic':
                user, password = b64decode(token.encode('ascii')).decode('ascii').split(':')
                self.__session.auth = (user, password)
            else:
                if len(self.__token_type) == 0:
                    self.__session.headers.update({self.__auth_header_name: f'{self.__token}'})
                else:
                    self.__session.headers.update({self.__auth_header_name: f'{self.__token_type} {self.__token}'})

    def update_cookies(self, cookies: Optional[List] = None, cookie_jar: Optional[RequestsCookieJar] = None) -> None:
        """
        Updates session cookies.

        Args:
            cookies (Optional[List], optional): _description_. Defaults to None.
            cookie_jar (Optional[RequestsCookieJar], optional): _description_. Defaults to None.
        """
        if cookies:
            for c in cookies:
                self.__session.cookies.set_cookie(c)
        if cookie_jar:
            self.__session.cookies.update(cookie_jar)

delete(route, **kwargs)

Performs a DELETE request with the session details and given headers and tokens.

Parameters:

Name Type Description Default
route str

description

 required

Returns:

Name Type Description
Response Response

description
Source code in ark_sdk_python/common/ark_client.py 
128
129
130
131
132
133
134
135
136
137
138
def delete(self, route: str, **kwargs) -> Response:
    """
    Performs a DELETE request with the session details and given headers and tokens.

    Args:
        route (str): _description_

    Returns:
        Response: _description_
    """
    return self.generic_http_method_request('delete', route, **kwargs)

get(route, **kwargs)

Performs a GET request with the session details and given headers and tokens.

Parameters:

Name Type Description Default
route str

description

 required

Returns:

Name Type Description
Response Response

description
Source code in ark_sdk_python/common/ark_client.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def get(self, route: str, **kwargs) -> Response:
    """
    Performs a GET request with the session details and given headers and tokens.

    Args:
        route (str): _description_

    Returns:
        Response: _description_
    """
    return self.generic_http_method_request('get', route, **kwargs)

options(route, **kwargs)

Performs a OPTIONS request with the session details and given headers and tokens.

Parameters:

Name Type Description Default
route str

description

 required

Returns:

Name Type Description
Response Response

description
Source code in ark_sdk_python/common/ark_client.py 
152
153
154
155
156
157
158
159
160
161
162
def options(self, route: str, **kwargs) -> Response:
    """
    Performs a OPTIONS request with the session details and given headers and tokens.

    Args:
        route (str): _description_

    Returns:
        Response: _description_
    """
    return self.generic_http_method_request('options', route, **kwargs)

patch(route, **kwargs)

Performs a PATCH request with the session details and given headers and tokens.

Parameters:

Name Type Description Default
route str

description

 required

Returns:

Name Type Description
Response Response

description
Source code in ark_sdk_python/common/ark_client.py 
140
141
142
143
144
145
146
147
148
149
150
def patch(self, route: str, **kwargs) -> Response:
    """
    Performs a PATCH request with the session details and given headers and tokens.

    Args:
        route (str): _description_

    Returns:
        Response: _description_
    """
    return self.generic_http_method_request('patch', route, **kwargs)

post(route, **kwargs)

Performs a POST request with the session details and given headers and tokens.

Parameters:

Name Type Description Default
route str

description

 required

Returns:

Name Type Description
Response Response

description
Source code in ark_sdk_python/common/ark_client.py 
104
105
106
107
108
109
110
111
112
113
114
def post(self, route: str, **kwargs) -> Response:
    """
    Performs a POST request with the session details and given headers and tokens.

    Args:
        route (str): _description_

    Returns:
        Response: _description_
    """
    return self.generic_http_method_request('post', route, **kwargs)

put(route, **kwargs)

Performs a PUT request with the session details and given headers and tokens.

Parameters:

Name Type Description Default
route str

description

 required

Returns:

Name Type Description
Response Response

description
Source code in ark_sdk_python/common/ark_client.py 
116
117
118
119
120
121
122
123
124
125
126
def put(self, route: str, **kwargs) -> Response:
    """
    Performs a PUT request with the session details and given headers and tokens.

    Args:
        route (str): _description_

    Returns:
        Response: _description_
    """
    return self.generic_http_method_request('put', route, **kwargs)

update_cookies(cookies=None, cookie_jar=None)

Updates session cookies.

Parameters:

Name Type Description Default
cookies Optional[List]

description. Defaults to None.

 None
cookie_jar Optional[RequestsCookieJar]

description. Defaults to None.

 None
Source code in ark_sdk_python/common/ark_client.py 
182
183
184
185
186
187
188
189
190
191
192
193
194
def update_cookies(self, cookies: Optional[List] = None, cookie_jar: Optional[RequestsCookieJar] = None) -> None:
    """
    Updates session cookies.

    Args:
        cookies (Optional[List], optional): _description_. Defaults to None.
        cookie_jar (Optional[RequestsCookieJar], optional): _description_. Defaults to None.
    """
    if cookies:
        for c in cookies:
            self.__session.cookies.set_cookie(c)
    if cookie_jar:
        self.__session.cookies.update(cookie_jar)

update_token(token=None)

Updates a session token.

Parameters:

Name Type Description Default
token Optional[str]

description. Defaults to None.

 None
Source code in ark_sdk_python/common/ark_client.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def update_token(self, token: Optional[str] = None) -> None:
    """
    Updates a session token.

    Args:
        token (Optional[str], optional): _description_. Defaults to None.
    """
    self.__token = token
    if token:
        if self.__token_type == 'Basic':
            user, password = b64decode(token.encode('ascii')).decode('ascii').split(':')
            self.__session.auth = (user, password)
        else:
            if len(self.__token_type) == 0:
                self.__session.headers.update({self.__auth_header_name: f'{self.__token}'})
            else:
                self.__session.headers.update({self.__auth_header_name: f'{self.__token_type} {self.__token}'})