Route4Me._make_request() - Code Metrics - Inspection of "Added Optimized Callback URL feature and example" - route4me/route4me-python-sdk - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 71f83d...bc7c18 )

by Juan Jose

created 2017-08-03 22:59 UTC

Route4Me._make_request() A

↳ Parent: Route4Me

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	6
CRAP Score	2.0116

Importance

Changes	2
Bugs	0	Features	0

Metric	Value
cc	2
c	2
b	0
f	0
dl	0
loc	17
rs	9.4285
ccs	6
cts	7
cp	0.8571
crap	2.0116

import json

import requests

# codebeat:disable[TOO_MANY_FUNCTIONS, LOC, ABC, ARITY, TOTAL_LOC]
try:
    from urllib import urlencode
except ImportError:
    from urllib.parse import urlencode
from .activity_feed import ActivityFeed
from .address import Address
from .address_book import AddressBook
from .avoidance_zones import AvoindanceZones
from .exceptions import APIException
from .gps import GPS
from .optimization import Optimization
from .orders import Order
from .rapid_address import RapidAddress
from .route import Route
from .territory import Territory
from .utils import json2obj
from .file_uploading import FileUploading
from .members import Members
from .vehicles import Vehicle
from .api_endpoints import API_HOST


HEADERS = {
    'User-Agent': 'python-sdk',
    'Accept-Encoding': 'identity, deflate, compress, gzip',
    'Accept': '*/*',
}


class Route4Me(object):
    """
    Route4Me Python SDK
    """

    def __init__(self,
                 key,
                 headers=HEADERS,
                 redirects=True,
                 verify_ssl=True,
                 proxies={}):
        self.key = key
        self.response = None
        self.activity_feed = ActivityFeed(self)
        self.address = Address(self)
        self.address_book = AddressBook(self)
        self.avoidance_zones = AvoindanceZones(self)
        self.file_uploading = FileUploading(self)
        self.optimization = Optimization(self)
        self.members = Members(self)
        self.vehicles = Vehicle(self)
        self.order = Order(self)
        self.gps = GPS(self)
        self.route = Route(self)
        self.rapid_address = RapidAddress(self)
        self.territory = Territory(self)
        self.headers = headers
        self.redirects = redirects
        self.verify_ssl = verify_ssl
        self.proxies = proxies

    def _make_request(self, url, params, data, request_method):
        """
        Make request to API
        :param url:
        :param params:
        :param data:
        :param request_method:
        :return: response
        :raise: APIException
        """
        params['api_key'] = self.key
        request_params = self._transform_params(params)
        response = request_method(url, request_params, data)
        if not 200 <= response.status_code < 400:
            raise APIException(response.status_code, response.text,
                               response.url)
        return response

    def _transform_params(self, params):
        """
        Convert params dict to url params
        :param params:
        :return:
        """
        return urlencode(params)

    def get(self, request_method):
        """
        Execute optimization
        :param request_method:
        :return: JSON
        """
        params = self.optimization.get_params()
        return self._make_request(API_HOST, params,
                                  json.dumps(self.optimization.data),
                                  request_method)

    def _request_post(self, url, request_params, data=None, files=None):
        """
        POST request
        :param url:
        :param request_params:
        :param data:
        :param files:
        :return:
        """
        return requests.post(url, params=request_params,
                             allow_redirects=self.redirects,
                             proxies=self.proxies, files=files,
                             data=data, headers=self.headers,
                             verify=self.verify_ssl)

    def _request_get(self, url, request_params, data=None):
        """
        GET request
        :param url:
        :param request_params:
        :param data:
        :return:
        """
        return requests.get(url, params=request_params,
                            allow_redirects=self.redirects,
                            proxies=self.proxies,
                            data=data,
                            headers=self.headers,
                            verify=self.verify_ssl)

    def _request_put(self, url, request_params, data=None):
        """
        PUT request
        :param url:
        :param request_params:
        :param data:
        :return:
        """
        return requests.request('PUT', url, params=request_params,
                                proxies=self.proxies,
                                data=data,
                                headers=self.headers,
                                verify=self.verify_ssl)

    def _request_delete(self, url, request_params, data=None):
        """
        DELETE request
        :param url:
        :param request_params:
        :param data:
        :return:
        """
        return requests.request('DELETE', url, params=request_params,
                                data=data,
                                headers=self.headers,
                                verify=self.verify_ssl)

    def run_optimization(self):
        """
        Run optimization and return response as an object.
        :return: response as an object
        """
        self.response = self.get(self._request_post)
        try:
            response = self.response.json()
            return response
        except AttributeError:
            raise
        except ValueError:
            raise
        except Exception:
            raise

    def re_optimization(self, optimization_id, data={}):
        """
        Execute reoptimization
        :param optimization_id:
        :return: response as a object
        """
        self.optimization.optimization_problem_id(optimization_id)
        self.optimization.reoptimize(1)
        data = {'parameters': data}
        self.response = self._make_request(API_HOST,
                                           self.optimization.get_params(),
                                           json.dumps(data),
                                           self._request_put)
        try:
            response = json2obj(self.response.content)
            return response
        except ValueError:
            raise
        except Exception:
            raise

    def get_optimization(self, optimization_problem_id):
        """
        Get optimization given optimization_problem_id
        :param optimization_problem_id:
        :return:
        """
        self.optimization.optimization_problem_id(optimization_problem_id)
        self.response = self.get(self._request_get)
        self.parse_response()

    def parse_response(self):
        """
        Parse response and set it to Route4me instance
        :return:
        """
        response = json.loads(self.response.content)
        if 'addresses' in response:
            self.address.addresses = self.response['addresses']

    def export_result_to_json(self, file_name):
        """
        Export response to JSON File
        :param file_name:
        :return:
        """
        if self.response:
            try:
                f = open(file_name, 'w')
                f.write(json.dumps(self.response.content,
                                   ensure_ascii=False,
                                   sort_keys=True,
                                   indent=4))
                f.close()
            except Exception:
                raise

    def export_request_to_json(self, file_name):
        """
        Export resquest to JSON File
        :param file_name:
        :return:
        """
        if self.optimization.data:
            try:
                f = open(file_name, 'w')
                f.write(json.dumps(self.optimization.data,
                                   ensure_ascii=False,
                                   sort_keys=True,
                                   indent=4))
                f.close()
            except Exception:
                raise

# codebeat:enable[TOO_MANY_FUNCTIONS, LOC, ABC, ARITY, TOTAL_LOC]


1	1	import json
2
3	1	import requests
4
5		# codebeat:disable[TOO_MANY_FUNCTIONS, LOC, ABC, ARITY, TOTAL_LOC]
6	1	try:
7	1	from urllib import urlencode
8	1	except ImportError:
9	1	from urllib.parse import urlencode
10	1	from .activity_feed import ActivityFeed
11	1	from .address import Address
12	1	from .address_book import AddressBook
13	1	from .avoidance_zones import AvoindanceZones
14	1	from .exceptions import APIException
15	1	from .gps import GPS
16	1	from .optimization import Optimization
17	1	from .orders import Order
18	1	from .rapid_address import RapidAddress
19	1	from .route import Route
20	1	from .territory import Territory
21	1	from .utils import json2obj
22	1	from .file_uploading import FileUploading
23	1	from .members import Members
24	1	from .vehicles import Vehicle
25	1	from .api_endpoints import API_HOST
26
27
28	1	HEADERS = {
29		'User-Agent': 'python-sdk',
30		'Accept-Encoding': 'identity, deflate, compress, gzip',
31		'Accept': '/',
32		}
33
34
35	1	class Route4Me(object):
36		"""
37		Route4Me Python SDK
38		"""
39
40	1	def __init__(self,
41		key,
42		headers=HEADERS,
43		redirects=True,
44		verify_ssl=True,
45		proxies={}):
46	1	self.key = key
47	1	self.response = None
48	1	self.activity_feed = ActivityFeed(self)
49	1	self.address = Address(self)
50	1	self.address_book = AddressBook(self)
51	1	self.avoidance_zones = AvoindanceZones(self)
52	1	self.file_uploading = FileUploading(self)
53	1	self.optimization = Optimization(self)
54	1	self.members = Members(self)
55	1	self.vehicles = Vehicle(self)
56	1	self.order = Order(self)
57	1	self.gps = GPS(self)
58	1	self.route = Route(self)
59	1	self.rapid_address = RapidAddress(self)
60	1	self.territory = Territory(self)
61	1	self.headers = headers
62	1	self.redirects = redirects
63	1	self.verify_ssl = verify_ssl
64	1	self.proxies = proxies
65
66	1	def _make_request(self, url, params, data, request_method):
67		"""
68		Make request to API
69		:param url:
70		:param params:
71		:param data:
72		:param request_method:
73		:return: response
74		:raise: APIException
75		"""
76	1	params['api_key'] = self.key
77	1	request_params = self._transform_params(params)
78	1	response = request_method(url, request_params, data)
79	1	if not 200 <= response.status_code < 400:
80		raise APIException(response.status_code, response.text,
81		response.url)
82	1	return response
83
84	1	def _transform_params(self, params):
85		"""
86		Convert params dict to url params
87		:param params:
88		:return:
89		"""
90	1	return urlencode(params)
91
92	1	def get(self, request_method):
93		"""
94		Execute optimization
95		:param request_method:
96		:return: JSON
97		"""
98	1	params = self.optimization.get_params()
99	1	return self._make_request(API_HOST, params,
100		json.dumps(self.optimization.data),
101		request_method)
102
103	1	def _request_post(self, url, request_params, data=None, files=None):
104		"""
105		POST request
106		:param url:
107		:param request_params:
108		:param data:
109		:param files:
110		:return:
111		"""
112	1	return requests.post(url, params=request_params,
113		allow_redirects=self.redirects,
114		proxies=self.proxies, files=files,
115		data=data, headers=self.headers,
116		verify=self.verify_ssl)
117
118	1	def _request_get(self, url, request_params, data=None):
119		"""
120		GET request
121		:param url:
122		:param request_params:
123		:param data:
124		:return:
125		"""
126	1	return requests.get(url, params=request_params,
127		allow_redirects=self.redirects,
128		proxies=self.proxies,
129		data=data,
130		headers=self.headers,
131		verify=self.verify_ssl)
132
133	1	def _request_put(self, url, request_params, data=None):
134		"""
135		PUT request
136		:param url:
137		:param request_params:
138		:param data:
139		:return:
140		"""
141		return requests.request('PUT', url, params=request_params,
142		proxies=self.proxies,
143		data=data,
144		headers=self.headers,
145		verify=self.verify_ssl)
146
147	1	def _request_delete(self, url, request_params, data=None):
148		"""
149		DELETE request
150		:param url:
151		:param request_params:
152		:param data:
153		:return:
154		"""
155		return requests.request('DELETE', url, params=request_params,
156		data=data,
157		headers=self.headers,
158		verify=self.verify_ssl)
159
160	1	def run_optimization(self):
161		"""
162		Run optimization and return response as an object.
163		:return: response as an object
164		"""
165	1	self.response = self.get(self._request_post)
166	1	try:
167	1	response = self.response.json()
168	1	return response
169		except AttributeError:
170		raise
171		except ValueError:
172		raise
173		except Exception:
174		raise
175
176	1	def re_optimization(self, optimization_id, data={}):
177		"""
178		Execute reoptimization
179		:param optimization_id:
180		:return: response as a object
181		"""
182		self.optimization.optimization_problem_id(optimization_id)
183		self.optimization.reoptimize(1)
184		data = {'parameters': data}
185		self.response = self._make_request(API_HOST,
186		self.optimization.get_params(),
187		json.dumps(data),
188		self._request_put)
189		try:
190		response = json2obj(self.response.content)
191		return response
192		except ValueError:
193		raise
194		except Exception:
195		raise
196
197	1	def get_optimization(self, optimization_problem_id):
198		"""
199		Get optimization given optimization_problem_id
200		:param optimization_problem_id:
201		:return:
202		"""
203		self.optimization.optimization_problem_id(optimization_problem_id)
204		self.response = self.get(self._request_get)
205		self.parse_response()
206
207	1	def parse_response(self):
208		"""
209		Parse response and set it to Route4me instance
210		:return:
211		"""
212		response = json.loads(self.response.content)
213		if 'addresses' in response:
214		self.address.addresses = self.response['addresses']
215
216	1	def export_result_to_json(self, file_name):
217		"""
218		Export response to JSON File
219		:param file_name:
220		:return:
221		"""
222		if self.response:
223		try:
224		f = open(file_name, 'w')
225		f.write(json.dumps(self.response.content,
226		ensure_ascii=False,
227		sort_keys=True,
228		indent=4))
229		f.close()
230		except Exception:
231		raise
232
233	1	def export_request_to_json(self, file_name):
234		"""
235		Export resquest to JSON File
236		:param file_name:
237		:return:
238		"""
239		if self.optimization.data:
240		try:
241		f = open(file_name, 'w')
242		f.write(json.dumps(self.optimization.data,
243		ensure_ascii=False,
244		sort_keys=True,
245		indent=4))
246		f.close()
247		except Exception:
248		raise
249
250		# codebeat:enable[TOO_MANY_FUNCTIONS, LOC, ABC, ARITY, TOTAL_LOC]
251

route4me / route4me-python-sdk

GitHub Access Token became invalid

Push — master ( 71f83d...bc7c18 )

Route4Me._make_request() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like