Skip to content

Base Powertrain API Client

Bases: BaseAPIClient

Base client for Powertrain APIs.

It is used to communicate with Powertrain APIs that provides the data useful for Report Generation process.

Source code in reportconnectors/api_client/powertrain/base.py 
 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
195
196
197
198
199
200
201
202
203
class BasePowertrainAPIClient(BaseAPIClient):
    """
    Base client for Powertrain APIs.

    It is used to communicate with Powertrain APIs that provides the data useful for Report Generation process.
    """

    __version__ = "0.1.0"
    _datetime_format: str = "%Y-%m-%dT%H:%M:%SZ"
    _api_version: str = "1.0"
    _status_prefix = ""

    class KeyNames(BaseAPIClient.KeyNames):
        JWT_PAYLOAD = "jwt_payload"

    def __init__(
        self, url: str, ciam_url: Optional[str] = None, ciam_api_client: Optional[CiamAPIClient] = None, **kwargs
    ):
        """
        Initializes the Powertrain API client.
        Because the Powertrain service uses CIAM as an authentication provider,
        you need to provide the existing instance of CIAM API Client or the URL of the CIAM API.

        Args:
            url: Powertrain API URL
            ciam_api_client: CIAM API client. If not provided, a new one will be initialized
                with the provided `ciam_url`.
            ciam_url: CIAM API URL. If provided it will be used to initialize the CIAM API client.

        Keyword Args:
            timeout (float): Timeout for the API requests. Default: `60`
            proxies (Dict[str, str]): Proxy settings. Default: `None`
            escape_html (bool): If set to `True`, the HTML content will be escaped. Default: `True`
            cert_path (Union[str, bool]): Path to the certificate file or `False` to disable certificate verification.
                Default: `False`
        """
        super().__init__(url=url, **kwargs)
        if isinstance(ciam_api_client, CiamAPIClient):
            self.ciam_api_client = ciam_api_client
        elif ciam_url is not None:
            self.ciam_api_client = CiamAPIClient(url=ciam_url, **kwargs)
        else:
            raise ValueError("CIAM API client or URL must be provided.")

    def authenticate(self, client_id: str, client_secret: str, password: str, username: str, **kwargs) -> bool:
        """
        Authenticates the client.

        Because the Powertrain microservices use CIAM as an authentication provider,
        the authentication process is done through the CIAM API client.

        Args:
            client_id: Client ID
            client_secret: Client secret
            password: User password
            username: Username

        Keyword Args:
            token_url (str): Authentication token URL. If not provided, it is taken from the CIAM API client.
            timeout (float): Timeout for the authentication process. Default: `60`

        Returns:
            True if client is successfully authenticated and access token in obtained and set. Otherwise, False.
        """
        self.ciam_api_client.authenticate(
            client_id=client_id, client_secret=client_secret, password=password, username=username, **kwargs
        )
        self.auth_data = dict(self.ciam_api_client.auth_data)
        return self.is_logged

    @property
    def is_logged(self):
        """
        Checks if the client is logged in.
        """
        time_now = datetime.datetime.now(datetime.timezone.utc)
        _is_logged = (
            bool(self.auth_token)
            and isinstance(self.token_expiration_date, datetime.datetime)
            and (self.token_expiration_date > time_now)
        )
        return _is_logged

    def set_token(self, access_token: str, validate: bool = False, **kwargs) -> bool:
        """
        Sets the authorization tokens without the authentication process. It might be used when the
        tokens are obtained from external source.

        Args:
            access_token: Access token to be set.
            validate: If set to True, the token will be validated. Default: False

        Returns:
            True if provided token is set and valid. Otherwise, False.
        """

        status = self.ciam_api_client.set_token(access_token=access_token, validate=validate, **kwargs)
        if status:
            self.auth_data = dict(self.ciam_api_client.auth_data)
        return status

    def check_api_status(self) -> bool:
        """
        Checks the api state by calling the dedicated status endpoints under the `/Status/IsRunning` path.

        Returns:
            True if the status is OK, False otherwise.
        """

        endpoint = f"{self._status_prefix}/Status/IsRunning"
        response = self._make_request(method="GET", endpoint=endpoint)
        status_code = self._get_status_code(response)
        is_running = status_code == HTTPStatus.OK
        return is_running

    def _make_access_token_refresh(self, force: bool = False) -> bool:
        """
        Refreshes the access token using the existing refresh token.

        Because it overrides the base class method, it will be called by `_process_request` method just before calling
        `_send_request` method.

        To refresh the token it uses the CiamAPIClient's method .refresh_authentication().
        If the process ends with success, the auth data is updated.

        Args:
            force: If True, the token will be refreshed even if it is not expired. Default: False

        Returns:
            True, if the token refresh process is successful. False, otherwise.
        """

        status = self.ciam_api_client.refresh_authentication(force=force)
        if status:
            self.auth_data = dict(self.ciam_api_client.auth_data)
        return status

    @classmethod
    def from_access_token(cls, url: str, ciam_url: str, access_token: str, **kwargs) -> "BasePowertrainAPIClient":
        """
        Initialize the client with the provided access token.

        Args:
            url: URL of the API.
            ciam_url: URL of the CIAM API.
            access_token: Valid access token to the API.

        Keyword Args:
            refresh_token (str): Valid refresh token that will be used to automatically refresh the access token
            id_token (str): Valid ID token
            validate (bool): If True, the token will be validated. Default: `False`
            audience (str): Audience for the token validation. Default: `None`

        Returns:
            Initialized and authenticated Powertrain API client.
        """
        validate: bool = kwargs.pop("validate", False)
        audience: Optional[str] = kwargs.pop("audience", None)
        refresh_token: Optional[str] = kwargs.pop("refresh_token", None)
        id_token: Optional[str] = kwargs.pop("id_token", None)

        client = cls(url=url, ciam_url=ciam_url, **kwargs)
        client.set_token(
            access_token=access_token,
            refresh_token=refresh_token,
            id_token=id_token,
            validate=validate,
            audience=audience,
        )
        return client

is_logged property

Checks if the client is logged in.

__init__(url, ciam_url=None, ciam_api_client=None, **kwargs)

Initializes the Powertrain API client. Because the Powertrain service uses CIAM as an authentication provider, you need to provide the existing instance of CIAM API Client or the URL of the CIAM API.

Parameters:

Name Type Description Default
url str

Powertrain API URL

 required
ciam_api_client Optional[CiamAPIClient]

CIAM API client. If not provided, a new one will be initialized with the provided ciam_url.

 None
ciam_url Optional[str]

CIAM API URL. If provided it will be used to initialize the CIAM API client.

 None

Other Parameters:

Name Type Description
timeout float

Timeout for the API requests. Default: 60
proxies Dict[str, str]

Proxy settings. Default: None
escape_html bool

If set to True, the HTML content will be escaped. Default: True
cert_path Union[str, bool]

Path to the certificate file or False to disable certificate verification. Default: False
Source code in reportconnectors/api_client/powertrain/base.py 
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
def __init__(
    self, url: str, ciam_url: Optional[str] = None, ciam_api_client: Optional[CiamAPIClient] = None, **kwargs
):
    """
    Initializes the Powertrain API client.
    Because the Powertrain service uses CIAM as an authentication provider,
    you need to provide the existing instance of CIAM API Client or the URL of the CIAM API.

    Args:
        url: Powertrain API URL
        ciam_api_client: CIAM API client. If not provided, a new one will be initialized
            with the provided `ciam_url`.
        ciam_url: CIAM API URL. If provided it will be used to initialize the CIAM API client.

    Keyword Args:
        timeout (float): Timeout for the API requests. Default: `60`
        proxies (Dict[str, str]): Proxy settings. Default: `None`
        escape_html (bool): If set to `True`, the HTML content will be escaped. Default: `True`
        cert_path (Union[str, bool]): Path to the certificate file or `False` to disable certificate verification.
            Default: `False`
    """
    super().__init__(url=url, **kwargs)
    if isinstance(ciam_api_client, CiamAPIClient):
        self.ciam_api_client = ciam_api_client
    elif ciam_url is not None:
        self.ciam_api_client = CiamAPIClient(url=ciam_url, **kwargs)
    else:
        raise ValueError("CIAM API client or URL must be provided.")

authenticate(client_id, client_secret, password, username, **kwargs)

Authenticates the client.

Because the Powertrain microservices use CIAM as an authentication provider, the authentication process is done through the CIAM API client.

Parameters:

Name Type Description Default
client_id str

Client ID

 required
client_secret str

Client secret

 required
password str

User password

 required
username str

Username

 required

Other Parameters:

Name Type Description
token_url str

Authentication token URL. If not provided, it is taken from the CIAM API client.
timeout float

Timeout for the authentication process. Default: 60

Returns:

Type Description
bool

True if client is successfully authenticated and access token in obtained and set. Otherwise, False.
Source code in reportconnectors/api_client/powertrain/base.py 
 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
def authenticate(self, client_id: str, client_secret: str, password: str, username: str, **kwargs) -> bool:
    """
    Authenticates the client.

    Because the Powertrain microservices use CIAM as an authentication provider,
    the authentication process is done through the CIAM API client.

    Args:
        client_id: Client ID
        client_secret: Client secret
        password: User password
        username: Username

    Keyword Args:
        token_url (str): Authentication token URL. If not provided, it is taken from the CIAM API client.
        timeout (float): Timeout for the authentication process. Default: `60`

    Returns:
        True if client is successfully authenticated and access token in obtained and set. Otherwise, False.
    """
    self.ciam_api_client.authenticate(
        client_id=client_id, client_secret=client_secret, password=password, username=username, **kwargs
    )
    self.auth_data = dict(self.ciam_api_client.auth_data)
    return self.is_logged

check_api_status()

Checks the api state by calling the dedicated status endpoints under the /Status/IsRunning path.

Returns:

Type Description
bool

True if the status is OK, False otherwise.
Source code in reportconnectors/api_client/powertrain/base.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
def check_api_status(self) -> bool:
    """
    Checks the api state by calling the dedicated status endpoints under the `/Status/IsRunning` path.

    Returns:
        True if the status is OK, False otherwise.
    """

    endpoint = f"{self._status_prefix}/Status/IsRunning"
    response = self._make_request(method="GET", endpoint=endpoint)
    status_code = self._get_status_code(response)
    is_running = status_code == HTTPStatus.OK
    return is_running

from_access_token(url, ciam_url, access_token, **kwargs) classmethod

Initialize the client with the provided access token.

Parameters:

Name Type Description Default
url str

URL of the API.

 required
ciam_url str

URL of the CIAM API.

 required
access_token str

Valid access token to the API.

 required

Other Parameters:

Name Type Description
refresh_token str

Valid refresh token that will be used to automatically refresh the access token
id_token str

Valid ID token
validate bool

If True, the token will be validated. Default: False
audience str

Audience for the token validation. Default: None

Returns:

Type Description
BasePowertrainAPIClient

Initialized and authenticated Powertrain API client.
Source code in reportconnectors/api_client/powertrain/base.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
@classmethod
def from_access_token(cls, url: str, ciam_url: str, access_token: str, **kwargs) -> "BasePowertrainAPIClient":
    """
    Initialize the client with the provided access token.

    Args:
        url: URL of the API.
        ciam_url: URL of the CIAM API.
        access_token: Valid access token to the API.

    Keyword Args:
        refresh_token (str): Valid refresh token that will be used to automatically refresh the access token
        id_token (str): Valid ID token
        validate (bool): If True, the token will be validated. Default: `False`
        audience (str): Audience for the token validation. Default: `None`

    Returns:
        Initialized and authenticated Powertrain API client.
    """
    validate: bool = kwargs.pop("validate", False)
    audience: Optional[str] = kwargs.pop("audience", None)
    refresh_token: Optional[str] = kwargs.pop("refresh_token", None)
    id_token: Optional[str] = kwargs.pop("id_token", None)

    client = cls(url=url, ciam_url=ciam_url, **kwargs)
    client.set_token(
        access_token=access_token,
        refresh_token=refresh_token,
        id_token=id_token,
        validate=validate,
        audience=audience,
    )
    return client

set_token(access_token, validate=False, **kwargs)

Sets the authorization tokens without the authentication process. It might be used when the tokens are obtained from external source.

Parameters:

Name Type Description Default
access_token str

Access token to be set.

 required
validate bool

If set to True, the token will be validated. Default: False

 False

Returns:

Type Description
bool

True if provided token is set and valid. Otherwise, False.
Source code in reportconnectors/api_client/powertrain/base.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def set_token(self, access_token: str, validate: bool = False, **kwargs) -> bool:
    """
    Sets the authorization tokens without the authentication process. It might be used when the
    tokens are obtained from external source.

    Args:
        access_token: Access token to be set.
        validate: If set to True, the token will be validated. Default: False

    Returns:
        True if provided token is set and valid. Otherwise, False.
    """

    status = self.ciam_api_client.set_token(access_token=access_token, validate=validate, **kwargs)
    if status:
        self.auth_data = dict(self.ciam_api_client.auth_data)
    return status