6786892849

Committed 07 Nov 2023 03:44PM UTC coverage: 99.032%. First build

Build # 6786892849

Build Type

push

github

Committed by

tdetry

Commit Message

[0.1.23] fix accounting/invoices params

Run Details

14 of 16 new or added lines in 3 files covered. (87.5%)

2864 of 2892 relevant lines covered (99.03%)

0.99 hits per line

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

87.1

/chift/api/client.py

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

import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry

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 = False

    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.max_retries = kwargs.get("max_retries")
        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,
            )

            if self.max_retries:
                retries = Retry(
                    total=int(self.max_retries),
                    backoff_factor=0.1,
                    status_forcelist=[502],
                )
                self.session.mount("http://", HTTPAdapter(max_retries=retries))
                self.session.mount("https://", HTTPAdapter(max_retries=retries))

    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

        try:
            req = self.process_request(
                request_type, url, headers=headers, params=params, json=data
            )
        except requests.exceptions.RetryError as e:
            raise exceptions.ChiftException(
                f"After {self.max_retries} retries, the request failed."
            )

        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"),
                detail=result.get("detail"),
            )
        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, 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.delete(url_path, params=params)

    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, params=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, params=params)

    def update_one(
        self, chift_vertical, chift_model, chift_id, data, extra_path=None, params=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, params=params)

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

chift-oneapi / chift-python-sdk / 6786892849

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