Address.delete_address() - Code Metrics - Inspection of "[FIX] bugs and issues reported by flake8" - route4me/route4me-python-sdk - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( 022c91...926271 )

by Juan Jose

created 2017-03-02 07:32 UTC

Address.delete_address() A

↳ Parent: Address.fix_geocode()

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,
    SINGLE_GEOCODER,
)


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
        """
        self.response = self.api._make_request(SINGLE_GEOCODER,
                                               params,
                                               [],
                                               self.api._request_get)
        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(BATCH_GEOCODER,
                                               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 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
                  }
        params.update({'api_key': self.key})
        data = json.dumps(data)
        response = self.api._make_request(ADDRESS_HOST,
                                          params,
                                          data,
                                          self.api._request_put)
        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
                  }
        params.update({'api_key': self.key})
        response = self.api._make_request(ADDRESS_HOST,
                                          params,
                                          None,
                                          self.api._request_delete)

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

route4me / route4me-python-sdk

GitHub Access Token became invalid

Push — master ( 022c91...926271 )

Address.delete_address() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like