Address.delete_address() - Code Metrics - Inspection of "Merge branch 'master' of https://github.com/route4..." - route4me/route4me-python-sdk - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( e3a209...bb2a36 )

by Juan Jose

created 2017-03-02 05:37 UTC

Address.delete_address() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
dl	0
loc	3
rs	10
c	0
b	0
f	0

# codebeat:disable[ABC]
import xmltodict
import time
import random
import requests
import json

from route4me.base import Base
from route4me.exceptions import ParamValueException
from route4me.utils import json2obj
from route4me.api_endpoints import ADD_ROUTE_NOTES_HOST, BATCH_GEOCODER, ADDRESS_HOST


class Address(Base):
    """
    An Address is a destination in a route or optimization problem.
    Addresses can be depots, which means they are a departure points.
    Addresses can belong to only one route and one optimization problem,
    except for depots. One depot can be part of many routes if we have a
    VRP (multi-route) solution.
    """
    REQUIRED_FIELDS = ['address', 'lat', 'lng', ]

    def __init__(self, api, addresses=[]):
        """
        Address Instance
        :param api:
        :param addresses:
        :return:
        """
        self.addresses = addresses
        Base.__init__(self, api)

    def get_route_id(self):
        """
        Return Route ID
        :return:
        """
        return self.get_response()['route_id']

    def get_route_destination_id(self):
        """
        Return Destination ID
        :return:
        """
        return self.get_response()['route_destination_id']

    def get_addresses(self):
        """
        Return Addresses
        :return:
        """
        return self.addresses

    def add_address(self, **kwargs):
        """
        Add addresses to optimization
        :param kwargs:
        :return:
        """
        if self.check_required_params(kwargs, self.REQUIRED_FIELDS):
            self.addresses.append(kwargs)
            self.api.optimization.data['addresses'] = self.addresses
        else:
            raise ParamValueException('addresses', 'Params are not complete')

    def batch_fix_geocodes(self, addresses):
        geocoding_error = []
        param_address = 'addresses='
        for a in addresses:
            param_address = '{0}{1}||'.format(param_address, a.get('address'))
        params = {'format': 'xml', 'addresses': param_address}
        content = self.get_batch_geocodes(params)
        obj = xmltodict.parse(content)
        geocoded_addresses = []
        for i, d in enumerate(obj.get('destinations').get('destination')):
            try:
                address = dict([('lat', float(d.get('@lat'))),
                                ('lng', float(d.get('@lng'))),
                                ('time', addresses[i].get('time')),
                                ('alias', addresses[i].get('alias')),
                                ('address', d.get('@destination'))])
                if addresses[i].get('is_depot') == 1:
                    address.update(dict([('is_depot', 1)]))
                geocoded_addresses.append(address)
            except IndexError:
                geocoding_error.append(d)
        return geocoding_error, geocoded_addresses

    def get_geocode(self, params):
        """
        Get Geocodes from given address
        :param params:
        :return: response as a object
        """
        request_method = self._request_get
        self.response = self._make_request(self.single_geocoder_url(), params,
                                           [], request_method)
        return self.response.content

    def get_batch_geocodes(self, params):
        """
        Get Geocodes from given addresses
        :param params:
        :return: response as a object
        """
        self.response = self.api._make_request(self.batch_geocoder_url(),
                                               params,
                                               [],
                                               self.api._request_get)
        return self.response.content


    def fix_geocode(self, address):
        geocoding_error = None
        params = {'format': 'xml', 'address': address.get('address')}
        count = 0
        while True:
            try:
                content = self.get_geocode(params)
                obj = xmltodict.parse(content)
                obj = obj.get('result')
                address.update(dict([('lat', float(obj.get('@lat'))),
                                     ('lng', float(obj.get('@lng'))), ]))
                return geocoding_error, address
            except (AttributeError, requests.exceptions.ConnectionError):
                count += 1
                if count > 5:
                    geocoding_error = address
                    break
                time.sleep(random.randrange(1, 5) * 0.5)

        return geocoding_error, address

    def delete_address(self, params):
        params.update({'api_key': self.key})
        return self._make_request(ADDRESS_HOST, params, None, self._request_delete)

    def update_address(self, params, data):
        params.update({'api_key': self.key})
        data = json.dumps(data)
        return self._make_request(ADDRESS_HOST, params, data, self._request_put)

    def request_address(self, params):
        params.update({'api_key': self.key})
        return self._make_request(ADDRESS_HOST, params, None, self._request_get)

    def get_address(self, route_id, route_destination_id):
        params = {'route_id': route_id,
                  'route_destination_id': route_destination_id
                  }
        response = self.api.request_address(params)
        return json2obj(response.content)

    def get_address_notes(self, route_id, route_destination_id):
        params = {'route_id': route_id,
                  'route_destination_id': route_destination_id,
                  'notes': True,
                  }
        response = self.api.request_address(params)
        return json2obj(response.content)

    def update_address(self, data, route_id, route_destination_id):
        params = {'route_id': route_id,
                  'route_destination_id': route_destination_id
                  }
        response = self.api.update_address(params, data)
        return json2obj(response.content)

    def delete_address_from_route(self, route_id, route_destination_id):
        params = {'route_id': route_id,
                  'route_destination_id': route_destination_id
                  }
        response = self.api.delete_address(params)
        return json2obj(response.content)

    def add_address_notes(self, note, **kwargs):
        """
        Add Address  Note using POST request
        :return: API response
        :raise: ParamValueException if required params are not present.
        """
        if self.check_required_params(kwargs, ['address_id', 'route_id']):
            data = {'strUpdateType': kwargs.pop('activity_type'), 'strNoteContents': note}
            kwargs.update({'api_key': self.params['api_key'], })
            self.response = self.api._request_post(ADD_ROUTE_NOTES_HOST,
                                                   kwargs, data)
            response = json2obj(self.response.content)
            return response

        else:
            raise ParamValueException('params', 'Params are not complete')

    def geocode(self, **kwargs):
        """
        Bulk Geocoder using POST request
        :return: API response
        :raise: ParamValueException if required params are not present.
        """
        if 'format' not in kwargs:
            kwargs.update({'format': 'csv'})
        kwargs.update({'api_key': self.params['api_key'], })
        if self.check_required_params(kwargs, ['addresses', ]):
            response = self.api._request_post(BATCH_GEOCODER,
                                              kwargs)
            return response.content

        else:
            raise ParamValueException('params', 'Params are not complete')
# codebeat:enable[ABC]


1			# codebeat:disable[ABC]
2			import xmltodict
3			import time
4			import random
5			import requests
6			import json
7
8			from route4me.base import Base
9			from route4me.exceptions import ParamValueException
10			from route4me.utils import json2obj
11			from route4me.api_endpoints import ADD_ROUTE_NOTES_HOST, BATCH_GEOCODER, ADDRESS_HOST
12
13
14			class Address(Base):
15			"""
16			An Address is a destination in a route or optimization problem.
17			Addresses can be depots, which means they are a departure points.
18			Addresses can belong to only one route and one optimization problem,
19			except for depots. One depot can be part of many routes if we have a
20			VRP (multi-route) solution.
21			"""
22			REQUIRED_FIELDS = ['address', 'lat', 'lng', ]
23
24			def __init__(self, api, addresses=[]):
25			"""
26			Address Instance
27			:param api:
28			:param addresses:
29			:return:
30			"""
31			self.addresses = addresses
32			Base.__init__(self, api)
33
34			def get_route_id(self):
35			"""
36			Return Route ID
37			:return:
38			"""
39			return self.get_response()['route_id']
40
41			def get_route_destination_id(self):
42			"""
43			Return Destination ID
44			:return:
45			"""
46			return self.get_response()['route_destination_id']
47
48			def get_addresses(self):
49			"""
50			Return Addresses
51			:return:
52			"""
53			return self.addresses
54
55			def add_address(self, **kwargs):
56			"""
57			Add addresses to optimization
58			:param kwargs:
59			:return:
60			"""
61			if self.check_required_params(kwargs, self.REQUIRED_FIELDS):
62			self.addresses.append(kwargs)
63			self.api.optimization.data['addresses'] = self.addresses
64			else:
65			raise ParamValueException('addresses', 'Params are not complete')
66
67			def batch_fix_geocodes(self, addresses):
68			geocoding_error = []
69			param_address = 'addresses='
70			for a in addresses:
71			param_address = '{0}{1}\|\|'.format(param_address, a.get('address'))
72			params = {'format': 'xml', 'addresses': param_address}
73			content = self.get_batch_geocodes(params)
74			obj = xmltodict.parse(content)
75			geocoded_addresses = []
76			for i, d in enumerate(obj.get('destinations').get('destination')):
77			try:
78			address = dict([('lat', float(d.get('@lat'))),
79			('lng', float(d.get('@lng'))),
80			('time', addresses[i].get('time')),
81			('alias', addresses[i].get('alias')),
82			('address', d.get('@destination'))])
83			if addresses[i].get('is_depot') == 1:
84			address.update(dict([('is_depot', 1)]))
85			geocoded_addresses.append(address)
86			except IndexError:
87			geocoding_error.append(d)
88			return geocoding_error, geocoded_addresses
89
90			def get_geocode(self, params):
91			"""
92			Get Geocodes from given address
93			:param params:
94			:return: response as a object
95			"""
96			request_method = self._request_get
97			self.response = self._make_request(self.single_geocoder_url(), params,
98			[], request_method)
99			return self.response.content
100
101			def get_batch_geocodes(self, params):
102			"""
103			Get Geocodes from given addresses
104			:param params:
105			:return: response as a object
106			"""
107			self.response = self.api._make_request(self.batch_geocoder_url(),
108			params,
109			[],
110			self.api._request_get)
111			return self.response.content
112
113
114			def fix_geocode(self, address):
115			geocoding_error = None
116			params = {'format': 'xml', 'address': address.get('address')}
117			count = 0
118			while True:
119			try:
120			content = self.get_geocode(params)
121			obj = xmltodict.parse(content)
122			obj = obj.get('result')
123			address.update(dict([('lat', float(obj.get('@lat'))),
124			('lng', float(obj.get('@lng'))), ]))
125			return geocoding_error, address
126			except (AttributeError, requests.exceptions.ConnectionError):
127			count += 1
128			if count > 5:
129			geocoding_error = address
130			break
131			time.sleep(random.randrange(1, 5) * 0.5)
132
133			return geocoding_error, address
134
135			def delete_address(self, params):
136			params.update({'api_key': self.key})
137			return self._make_request(ADDRESS_HOST, params, None, self._request_delete)
138
139			def update_address(self, params, data):
140			params.update({'api_key': self.key})
141			data = json.dumps(data)
142			return self._make_request(ADDRESS_HOST, params, data, self._request_put)
143
144			def request_address(self, params):
145			params.update({'api_key': self.key})
146			return self._make_request(ADDRESS_HOST, params, None, self._request_get)
147
148			def get_address(self, route_id, route_destination_id):
149			params = {'route_id': route_id,
150			'route_destination_id': route_destination_id
151			}
152			response = self.api.request_address(params)
153			return json2obj(response.content)
154
155			def get_address_notes(self, route_id, route_destination_id):
156			params = {'route_id': route_id,
157			'route_destination_id': route_destination_id,
158			'notes': True,
159			}
160			response = self.api.request_address(params)
161			return json2obj(response.content)
162
163			def update_address(self, data, route_id, route_destination_id):
164			params = {'route_id': route_id,
165			'route_destination_id': route_destination_id
166			}
167			response = self.api.update_address(params, data)
168			return json2obj(response.content)
169
170			def delete_address_from_route(self, route_id, route_destination_id):
171			params = {'route_id': route_id,
172			'route_destination_id': route_destination_id
173			}
174			response = self.api.delete_address(params)
175			return json2obj(response.content)
176
177			def add_address_notes(self, note, **kwargs):
178			"""
179			Add Address Note using POST request
180			:return: API response
181			:raise: ParamValueException if required params are not present.
182			"""
183			if self.check_required_params(kwargs, ['address_id', 'route_id']):
184			data = {'strUpdateType': kwargs.pop('activity_type'), 'strNoteContents': note}
185			kwargs.update({'api_key': self.params['api_key'], })
186			self.response = self.api._request_post(ADD_ROUTE_NOTES_HOST,
187			kwargs, data)
188			response = json2obj(self.response.content)
189			return response
190
191			else:
192			raise ParamValueException('params', 'Params are not complete')
193
194			def geocode(self, **kwargs):
195			"""
196			Bulk Geocoder using POST request
197			:return: API response
198			:raise: ParamValueException if required params are not present.
199			"""
200			if 'format' not in kwargs:
201			kwargs.update({'format': 'csv'})
202			kwargs.update({'api_key': self.params['api_key'], })
203			if self.check_required_params(kwargs, ['addresses', ]):
204			response = self.api._request_post(BATCH_GEOCODER,
205			kwargs)
206			return response.content
207
208			else:
209			raise ParamValueException('params', 'Params are not complete')
210			# codebeat:enable[ABC]
211

route4me / route4me-python-sdk

GitHub Access Token became invalid

Push — master ( e3a209...bb2a36 )

Address.delete_address() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like