mirror of
https://github.com/flobz/psa_car_controller.git
synced 2026-08-22 17:36:15 +00:00
79 lines
3.4 KiB
Python
79 lines
3.4 KiB
Python
from http import HTTPStatus
|
|
|
|
from oauth2_client.credentials_manager import CredentialManager
|
|
from requests import Response
|
|
|
|
import psa_connectedcar as psac
|
|
from mylogger import logger
|
|
from psa_connectedcar import ApiClient
|
|
from psa_connectedcar.rest import ApiException
|
|
|
|
|
|
class OpenIdCredentialManager(CredentialManager):
|
|
def _grant_password_request_realm(self, login: str, password: str, realm: str) -> dict:
|
|
return dict(grant_type='password',
|
|
username=login,
|
|
scope=' '.join(self.service_information.scopes),
|
|
password=password, realm=realm)
|
|
|
|
def init_with_user_credentials_realm(self, login: str, password: str, realm: str):
|
|
self._token_request(self._grant_password_request_realm(login, password, realm), True)
|
|
|
|
@staticmethod
|
|
def _is_token_expired(response: Response) -> bool:
|
|
if response.status_code == HTTPStatus.UNAUTHORIZED.value:
|
|
logger.info("token expired, renew")
|
|
try:
|
|
json_data = response.json()
|
|
return json_data.get('moreInformation') == 'Token is invalid'
|
|
except ValueError:
|
|
return False
|
|
else:
|
|
return False
|
|
|
|
@property
|
|
def access_token(self):
|
|
return self._access_token
|
|
|
|
|
|
class Oauth2PSACCApiConfig(psac.Configuration):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.refresh_callback = None
|
|
|
|
def set_refresh_callback(self, callback):
|
|
self.refresh_callback = callback
|
|
|
|
|
|
class OauthAPIClient(ApiClient):
|
|
# pylint: disable=no-member,too-many-arguments
|
|
def call_api(self, resource_path, method,
|
|
path_params=None, query_params=None, header_params=None,
|
|
body=None, post_params=None, files=None,
|
|
response_type=None, auth_settings=None, async_req=None,
|
|
_return_http_data_only=None, collection_formats=None,
|
|
_preload_content=True, _request_timeout=None):
|
|
for _ in range(0, 2):
|
|
try:
|
|
if not async_req:
|
|
return self._ApiClient__call_api(resource_path, method,
|
|
path_params, query_params, header_params,
|
|
body, post_params, files,
|
|
response_type, auth_settings,
|
|
_return_http_data_only, collection_formats,
|
|
_preload_content, _request_timeout)
|
|
return self.pool.apply_async(self.__call_api, (resource_path,
|
|
method, path_params, query_params,
|
|
header_params, body,
|
|
post_params, files,
|
|
response_type, auth_settings,
|
|
_return_http_data_only,
|
|
collection_formats,
|
|
_preload_content, _request_timeout))
|
|
except ApiException as e:
|
|
if e.reason == 'Unauthorized':
|
|
self.configuration.refresh_callback()
|
|
else:
|
|
raise e
|
|
return None
|