5961464362

Committed 24 Aug 2023 08:40AM UTC coverage: 99.31%. First build

Build # 5961464362

Build Type

Pull #3

github-actions

Committed by

tdetry

Commit Message

[FIX] run CI on push to master for badges

Pull Request Pull Request #3: [FIX] run CI on push to master for badges

Run Details

2592 of 2610 relevant lines covered (99.31%)

0.99 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

92.11

/chift/api/client.py

import http.client as httplib
import json
from datetime import datetime

import requests

from chift.api import exceptions


class ChiftAuth(requests.auth.AuthBase):
    def __init__(self, client_id, client_secret, account_id, url_base, env_id):
        self.client_id = client_id
        self.client_secret = client_secret
        self.account_id = account_id
        self.url_base = url_base
        self.env_id = env_id

        self.access_token = None
        self.exires_at = None

    def __call__(self, request):
        request.headers.update(self.get_auth_header())
        return request

    def get_auth_header(self):
        return {"Authorization": f"Bearer {self.get_access_token()}"}

    def _parse_token(self, token):
        self.access_token = token.get("access_token")
        self.exires_at = datetime.fromtimestamp(token.get("expires_on"))

    def get_access_token(self):
        if self.access_token:
            if datetime.now() < self.exires_at:
                return self.access_token

        payload = {
            "clientId": self.client_id,
            "clientSecret": self.client_secret,
            "accountId": self.account_id,
        }
        if self.env_id:
            payload["envId"] = self.env_id
        response = requests.post(self.url_base + "/token", json=payload)

        if not response.status_code == httplib.OK:
            raise exceptions.ChiftException(
                f"Error while authenticating '{response.status_code}': {response.text}"
            )

        self._parse_token(response.json())

        return self.access_token


class ChiftClient:
    session = None
    access_token = None
    auth = None
    consumer_id = None
    connection_id = None
    related_chain_execution_id = None

    __instance = None
    __use_global = True

    def __new__(cls, **kwargs):
        """
        __use_global=True for singleton usage
        """
        if ChiftClient.__use_global:
            if ChiftClient.__instance is None:
                ChiftClient.__instance = object.__new__(cls)
            instance = ChiftClient.__instance
        else:
            instance = object.__new__(cls)

        return instance

    def __init__(self, consumer_id=None, **kwargs):
        from chift import (
            account_id,
            client_id,
            client_secret,
            related_chain_execution_id,
            url_base,
        )

        self.consumer_id = consumer_id
        self.client_id = kwargs.get("client_id") or client_id
        self.client_secret = kwargs.get("client_secret") or client_secret
        self.account_id = kwargs.get("account_id") or account_id
        self.url_base = kwargs.get("url_base") or url_base
        self.env_id = kwargs.get("env_id")
        self.related_chain_execution_id = (
            kwargs.get("related_chain_execution_id") or related_chain_execution_id
        )
        self._start_session()

    def _start_session(self):
        if not self.session:
            self.session = requests.Session()
            self.session.auth = ChiftAuth(
                self.client_id,
                self.client_secret,
                self.account_id,
                self.url_base,
                self.env_id,
            )

    def make_request(
        self, request_type, url, data=None, content_type="application/json", params=None
    ):
        if not params:
            params = {}

        if isinstance(data, str):
            data = json.loads(data)

        headers = {
            "Content-Type": content_type,
            "Accept": "application/json",
            "User-Agent": "chift-python-sdk library",
        }

        if self.connection_id:
            headers["x-chift-connectionid"] = self.connection_id

        if self.related_chain_execution_id:
            headers["x-chift-relatedchainexecutionid"] = self.related_chain_execution_id

        req = self.process_request(
            request_type, url, headers=headers, params=params, json=data
        )

        if req.status_code == httplib.UNAUTHORIZED:
            raise exceptions.ChiftException(
                "Application authentication failed",
                error_code=req.status_code,
                detail=req.text,
            )

        try:
            result = req.json()
        except:
            raise exceptions.ChiftException(f"Error parsing json response: {req.text}")

        if isinstance(result, dict) and result.get("status") == "error":
            raise exceptions.ChiftException(
                result.get("message"), error_code=result.get("error_code")
            )
        elif not req.status_code == httplib.OK:
            raise exceptions.ChiftException(
                f"Error returned with status code '{req.status_code}': {req.text}"
            )
        else:
            return result

    def process_request(
        self, request_type, url_path, headers=None, params=None, json=None
    ):
        return self.session.request(
            request_type,
            self.url_base + url_path,
            headers=headers,
            params=params,
            json=json,
        )

    def get(self, *args, **kwargs):
        return self.make_request("GET", *args, **kwargs)

    def post(self, *args, **kwargs):
        return self.make_request("POST", *args, **kwargs)

    def patch(self, *args, **kwargs):
        return self.make_request("PATCH", *args, **kwargs)

    def delete(self, *args, **kwargs):
        return self.make_request("DELETE", *args, **kwargs)

    def path_builder(self, ordered_paths):
        if ordered_paths:
            return "/" + "/".join(
                [str(path).strip("/") for path in ordered_paths if path]
            )

    def delete_one(self, chift_vertical, chift_model, chift_id, extra_path=None):
        url_path = self.path_builder(
            [
                "consumers" if self.consumer_id else None,
                self.consumer_id,
                chift_vertical,
                chift_model,
                chift_id,
                extra_path,
            ]
        )

        return self.delete(url_path)

    def get_one(
        self, chift_vertical, chift_model, chift_id, params=None, extra_path=None
    ):
        url_path = self.path_builder(
            [
                "consumers" if self.consumer_id else None,
                self.consumer_id,
                chift_vertical,
                chift_model,
                chift_id,
                extra_path,
            ]
        )

        return self.get(url_path, params=params)

    def get_all(self, chift_vertical, chift_model, params=None, extra_path=None):
        url_path = self.path_builder(
            [
                "consumers" if self.consumer_id else None,
                self.consumer_id,
                chift_vertical,
                chift_model,
                extra_path,
            ]
        )

        return self.get(url_path, params=params)

    def post_one(self, chift_vertical, chift_model, data, extra_path=None):
        url_path = self.path_builder(
            [
                "consumers" if self.consumer_id else None,
                self.consumer_id,
                chift_vertical,
                chift_model,
                extra_path,
            ]
        )

        return self.post(url_path, data=data)

    def update_one(self, chift_vertical, chift_model, chift_id, data, extra_path=None):
        url_path = self.path_builder(
            [
                "consumers" if self.consumer_id else None,
                self.consumer_id,
                chift_vertical,
                chift_model,
                chift_id,
                extra_path,
            ]
        )

        return self.patch(url_path, data=data)

1	import http.client as httplib	1✔
2	import json	1✔
3	from datetime import datetime	1✔
4
5	import requests	1✔
6
7	from chift.api import exceptions	1✔
8
9
10	class ChiftAuth(requests.auth.AuthBase):	1✔
11	def __init__(self, client_id, client_secret, account_id, url_base, env_id):	1✔
12	self.client_id = client_id	1✔
13	self.client_secret = client_secret	1✔
14	self.account_id = account_id	1✔
15	self.url_base = url_base	1✔
16	self.env_id = env_id	1✔
17
18	self.access_token = None	1✔
19	self.exires_at = None	1✔
20
21	def __call__(self, request):	1✔
22	request.headers.update(self.get_auth_header())	1✔
23	return request	1✔
24
25	def get_auth_header(self):	1✔
26	return {"Authorization": f"Bearer {self.get_access_token()}"}	1✔
27
28	def _parse_token(self, token):	1✔
29	self.access_token = token.get("access_token")	1✔
30	self.exires_at = datetime.fromtimestamp(token.get("expires_on"))	1✔
31
32	def get_access_token(self):	1✔
33	if self.access_token:	1✔
34	if datetime.now() < self.exires_at:	1✔
35	return self.access_token	1✔
36
37	payload = {	1✔
38	"clientId": self.client_id,
39	"clientSecret": self.client_secret,
40	"accountId": self.account_id,
41	}
42	if self.env_id:	1✔
43	payload["envId"] = self.env_id	×
44	response = requests.post(self.url_base + "/token", json=payload)	1✔
45
46	if not response.status_code == httplib.OK:	1✔
47	raise exceptions.ChiftException(	×
48	f"Error while authenticating '{response.status_code}': {response.text}"
49	)
50
51	self._parse_token(response.json())	1✔
52
53	return self.access_token	1✔
54
55
56	class ChiftClient:	1✔
57	session = None	1✔
58	access_token = None	1✔
59	auth = None	1✔
60	consumer_id = None	1✔
61	connection_id = None	1✔
62	related_chain_execution_id = None	1✔
63
64	__instance = None	1✔
65	__use_global = True	1✔
66
67	def __new__(cls, **kwargs):	1✔
68	"""
69	__use_global=True for singleton usage
70	"""
71	if ChiftClient.__use_global:	1✔
72	if ChiftClient.__instance is None:	1✔
73	ChiftClient.__instance = object.__new__(cls)	1✔
74	instance = ChiftClient.__instance	1✔
75	else:
76	instance = object.__new__(cls)	×
77
78	return instance	1✔
79
80	def __init__(self, consumer_id=None, **kwargs):	1✔
81	from chift import (	1✔
82	account_id,
83	client_id,
84	client_secret,
85	related_chain_execution_id,
86	url_base,
87	)
88
89	self.consumer_id = consumer_id	1✔
90	self.client_id = kwargs.get("client_id") or client_id	1✔
91	self.client_secret = kwargs.get("client_secret") or client_secret	1✔
92	self.account_id = kwargs.get("account_id") or account_id	1✔
93	self.url_base = kwargs.get("url_base") or url_base	1✔
94	self.env_id = kwargs.get("env_id")	1✔
95	self.related_chain_execution_id = (	1✔
96	kwargs.get("related_chain_execution_id") or related_chain_execution_id
97	)
98	self._start_session()	1✔
99
100	def _start_session(self):	1✔
101	if not self.session:	1✔
102	self.session = requests.Session()	1✔
103	self.session.auth = ChiftAuth(	1✔
104	self.client_id,
105	self.client_secret,
106	self.account_id,
107	self.url_base,
108	self.env_id,
109	)
110
111	def make_request(	1✔
112	self, request_type, url, data=None, content_type="application/json", params=None
113	):
114	if not params:	1✔
115	params = {}	1✔
116
117	if isinstance(data, str):	1✔
118	data = json.loads(data)	×
119
120	headers = {	1✔
121	"Content-Type": content_type,
122	"Accept": "application/json",
123	"User-Agent": "chift-python-sdk library",
124	}
125
126	if self.connection_id:	1✔
127	headers["x-chift-connectionid"] = self.connection_id	1✔
128
129	if self.related_chain_execution_id:	1✔
130	headers["x-chift-relatedchainexecutionid"] = self.related_chain_execution_id	×
131
132	req = self.process_request(	1✔
133	request_type, url, headers=headers, params=params, json=data
134	)
135
136	if req.status_code == httplib.UNAUTHORIZED:	1✔
137	raise exceptions.ChiftException(	×
138	"Application authentication failed",
139	error_code=req.status_code,
140	detail=req.text,
141	)
142
143	try:	1✔
144	result = req.json()	1✔
145	except:	×
146	raise exceptions.ChiftException(f"Error parsing json response: {req.text}")	×
147
148	if isinstance(result, dict) and result.get("status") == "error":	1✔
149	raise exceptions.ChiftException(	1✔
150	result.get("message"), error_code=result.get("error_code")
151	)
152	elif not req.status_code == httplib.OK:	1✔
153	raise exceptions.ChiftException(	×
154	f"Error returned with status code '{req.status_code}': {req.text}"
155	)
156	else:
157	return result	1✔
158
159	def process_request(	1✔
160	self, request_type, url_path, headers=None, params=None, json=None
161	):
162	return self.session.request(	1✔
163	request_type,
164	self.url_base + url_path,
165	headers=headers,
166	params=params,
167	json=json,
168	)
169
170	def get(self, args, *kwargs):	1✔
171	return self.make_request("GET", args, *kwargs)	1✔
172
173	def post(self, args, *kwargs):	1✔
174	return self.make_request("POST", args, *kwargs)	1✔
175
176	def patch(self, args, *kwargs):	1✔
177	return self.make_request("PATCH", args, *kwargs)	1✔
178
179	def delete(self, args, *kwargs):	1✔
180	return self.make_request("DELETE", args, *kwargs)	1✔
181
182	def path_builder(self, ordered_paths):	1✔
183	if ordered_paths:	1✔
184	return "/" + "/".join(	1✔
185	[str(path).strip("/") for path in ordered_paths if path]
186	)
187
188	def delete_one(self, chift_vertical, chift_model, chift_id, extra_path=None):	1✔
189	url_path = self.path_builder(	1✔
190	[
191	"consumers" if self.consumer_id else None,
192	self.consumer_id,
193	chift_vertical,
194	chift_model,
195	chift_id,
196	extra_path,
197	]
198	)
199
200	return self.delete(url_path)	1✔
201
202	def get_one(	1✔
203	self, chift_vertical, chift_model, chift_id, params=None, extra_path=None
204	):
205	url_path = self.path_builder(	1✔
206	[
207	"consumers" if self.consumer_id else None,
208	self.consumer_id,
209	chift_vertical,
210	chift_model,
211	chift_id,
212	extra_path,
213	]
214	)
215
216	return self.get(url_path, params=params)	1✔
217
218	def get_all(self, chift_vertical, chift_model, params=None, extra_path=None):	1✔
219	url_path = self.path_builder(	1✔
220	[
221	"consumers" if self.consumer_id else None,
222	self.consumer_id,
223	chift_vertical,
224	chift_model,
225	extra_path,
226	]
227	)
228
229	return self.get(url_path, params=params)	1✔
230
231	def post_one(self, chift_vertical, chift_model, data, extra_path=None):	1✔
232	url_path = self.path_builder(	1✔
233	[
234	"consumers" if self.consumer_id else None,
235	self.consumer_id,
236	chift_vertical,
237	chift_model,
238	extra_path,
239	]
240	)
241
242	return self.post(url_path, data=data)	1✔
243
244	def update_one(self, chift_vertical, chift_model, chift_id, data, extra_path=None):	1✔
245	url_path = self.path_builder(	1✔
246	[
247	"consumers" if self.consumer_id else None,
248	self.consumer_id,
249	chift_vertical,
250	chift_model,
251	chift_id,
252	extra_path,
253	]
254	)
255
256	return self.patch(url_path, data=data)	1✔

chift-oneapi / chift-python-sdk / 5961464362

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous