src.api.ApiData.get_all_customers() - Code Metrics - Inspection of "Created start method" - Rahn20/spark-program - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( ce71f4...8ca48e )

by Rahn20

created 2023-01-07 04:19 UTC

src.api.ApiData.get_all_customers() A

↳ Parent: src.api

Complexity

Conditions

Size

Total Lines	10
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	7
CRAP Score	2

Importance

Changes

Metric	Value
cc	2
eloc	7
nop	1
dl	0
loc	10
ccs	7
cts	7
cp	1
crap	2
rs	10
c	0
b	0
f	0

#!/usr/bin/python3
# pylint: disable=broad-except

"""
URL: http://localhost:1337/api/v1/
"""

from datetime import datetime
import requests

from src.scooter import Scooter

class ApiData(Scooter):
    """ Api class """

    ## API endpoint URL
    _URL = "http://localhost:1337/api/v1/graphql"

    ## Set the headers
    _HEADERS = {'Content-Type': 'application/json'}


    def __init__(self, user_id: int) -> None:
        """ Initialize class """
        super().__init__()
        self.user_id = user_id
        self.log_id = ""


    def get_scooter_data(self, scooter_id: int) -> dict:
        """
        Get scooter data from API.
        """
        ## Create the GraphQL query
        query = ''' query getScooterById($id: String!) {
            getScooterById(id: $id) {
                id
                latitude
                longitude
                speed
                battery
                status {
                    id
                    status
                }
                station {
                    id
                    station_name
                }
            }
        } '''

        payload = {
            'query': query,
            'variables': {
                'id': str(scooter_id)
            }
        }

        try:
            ## Send the POST request
            response = requests.post(self._URL, json = payload, headers = self._HEADERS)

            return response.json()["data"]["getScooterById"][0]
        except (Exception, ConnectionError):
            return -1


    def update_scooter(self) -> None:
        """ Update api with scooter's new position, speed, status and battery level. """
        mutation = ''' mutation updateScooterById(
            $id: String!,
            $battery: String!,
            $status_id: String!,
            $longitude: String!,
            $latitude: String!,
            $price_id: String!,
            $speed: String!,
            $station_id: String!) {
                updateScooterById(
                    id: $id,
                    battery: $battery,
                    status_id: $status_id,
                    longitude: $longitude,
                    latitude: $latitude,
                    price_id: $price_id,
                    speed: $speed,
                    station_id: $station_id) { id }
        } '''

        payload = {
            'query': mutation,
            'variables': {
                'id': self.data["id"],
                'status_id': str(self.data["status"]),
                'latitude': str(self.data["lat"]),
                'longitude': str(self.data["lon"]),
                'speed': str(self.data["speed"]),
                'battery': str(self.data["battery"]),
                'station_id': str(self.data["station"]),
                'price_id': "1"
            }
        }

        try:
            requests.post(self._URL, json = payload, headers = self._HEADERS)
        except (Exception, ConnectionError) as error:
            print(error)



    def create_log(self) -> None:
        """
        Create log. Data to be added is scooter's position, start date/time and scooter/user id.
        """
        mutation = ''' mutation createLog(
            $scooter_id: String!,
            $customer_id: String!,
            $start_longitude: String!,
            $start_latitude: String!,
            $price_id: String!) {
                createLog(
                    scooter_id: $scooter_id,
                    customer_id: $customer_id,
                    start_longitude: $start_longitude,
                    start_latitude: $start_latitude,
                    price_id: $price_id) { id }
        } '''

        payload = {
            'query': mutation,
            'variables': {
                'scooter_id': self.data["id"],
                'customer_id': str(self.user_id),
                'start_longitude': str(self.data["lon"]),
                'start_latitude': str(self.data["lat"]),
                'price_id': "1"
            }
        }

        try:
            response = requests.post(self._URL, json = payload, headers = self._HEADERS)

            ## save the log id
            self.log_id = response.json()["data"]["createLog"]["id"]
        except (Exception, ConnectionError) as error:
            print(error)


    def update_log(self) -> None:
        """ Update log. Data to be updated is scooter's position and end date/time. """
        mutation = ''' mutation updateLogByLogId(
            $id: String!,
            $end_time: String!,
            $end_longitude: String!,
            $end_latitude: String!) {
                updateLogByLogId(
                    id: $id,
                    end_time: $end_time,
                    end_longitude: $end_longitude,
                    end_latitude: $end_latitude) { id }
        } '''

        payload = {
            'query': mutation,
            'variables': {
                'id': self.log_id,
                'end_time': datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
                'end_longitude': str(self.data["lon"]),
                'end_latitude': str(self.data["lat"]),
            }
        }

        try:
            requests.post(self._URL, json = payload, headers = self._HEADERS)
        except (Exception, ConnectionError) as error:
            print(error)


    def get_city_data(self) -> dict:
        """
        Get city's center position, id and area where the scooter is located.
        And adds it to city dictionary.
        """
        query = ''' query getCityByScooterId($id: String!) {
            getCityByScooterId(id: $id) {
                id
                latitude
                longitude
                area
            }
        } '''

        payload = {
            'query': query,
            'variables': {
                'id': self.data["id"]
            }
        }

        try:
            response = requests.post(self._URL, json = payload, headers = self._HEADERS)

            return response.json()["data"]["getCityByScooterId"][0]
        except (Exception, ConnectionError):
            return -1


    def get_station(self, zone_id: str) -> dict:
        """
        return random charging/maintenance station data in the city where the scooter is located.
        Zone id: 1- Charging Station, 2- Parking Station, 3- Bike Statione, 4- Maintenance Station.
        """
        query = ''' query getStationByCityIdAndZoneId($cityId: String!, $zoneId: String!) {
            getStationByCityIdAndZoneId(cityId: $cityId, zoneId: $zoneId) {
                id
                latitude
                longitude
            }
        } '''

        payload = {
            'query': query,
            'variables': {
                'cityId': self.city["id"],
                'zoneId': zone_id
            }
        }

        try:
            response = requests.post(self._URL, json = payload, headers = self._HEADERS)

            return response.json()["data"]["getStationByCityIdAndZoneId"][0]
        except (Exception, ConnectionError):
            return -1


    def get_all_customers(self) -> list:
        """ Get all customers data from API. """
        query = ''' query { getAllCustomers { id } }'''

        try:
            response = requests.post(self._URL, json = {'query': query}, headers = self._HEADERS)

            return response.json()["data"]["getAllCustomers"]
        except (Exception, ConnectionError):
            return -1


1		#!/usr/bin/python3
2		# pylint: disable=broad-except
3
4	1	"""
5		URL: http://localhost:1337/api/v1/
6		"""
7
8	1	from datetime import datetime
9	1	import requests
10
11	1	from src.scooter import Scooter
12
13	1	class ApiData(Scooter):
14		""" Api class """
15
16		## API endpoint URL
17	1	_URL = "http://localhost:1337/api/v1/graphql"
18
19		## Set the headers
20	1	_HEADERS = {'Content-Type': 'application/json'}
21
22
23	1	def __init__(self, user_id: int) -> None:
24		""" Initialize class """
25	1	super().__init__()
26	1	self.user_id = user_id
27	1	self.log_id = ""
28
29
30	1	def get_scooter_data(self, scooter_id: int) -> dict:
31		"""
32		Get scooter data from API.
33		"""
34		## Create the GraphQL query
35	1	query = ''' query getScooterById($id: String!) {
36		getScooterById(id: $id) {
37		id
38		latitude
39		longitude
40		speed
41		battery
42		status {
43		id
44		status
45		}
46		station {
47		id
48		station_name
49		}
50		}
51		} '''
52
53	1	payload = {
54		'query': query,
55		'variables': {
56		'id': str(scooter_id)
57		}
58		}
59
60	1	try:
61		## Send the POST request
62	1	response = requests.post(self._URL, json = payload, headers = self._HEADERS)
63
64	1	return response.json()["data"]["getScooterById"][0]
65	1	except (Exception, ConnectionError):
66	1	return -1
67
68
69	1	def update_scooter(self) -> None:
70		""" Update api with scooter's new position, speed, status and battery level. """
71	1	mutation = ''' mutation updateScooterById(
72		$id: String!,
73		$battery: String!,
74		$status_id: String!,
75		$longitude: String!,
76		$latitude: String!,
77		$price_id: String!,
78		$speed: String!,
79		$station_id: String!) {
80		updateScooterById(
81		id: $id,
82		battery: $battery,
83		status_id: $status_id,
84		longitude: $longitude,
85		latitude: $latitude,
86		price_id: $price_id,
87		speed: $speed,
88		station_id: $station_id) { id }
89		} '''
90
91	1	payload = {
92		'query': mutation,
93		'variables': {
94		'id': self.data["id"],
95		'status_id': str(self.data["status"]),
96		'latitude': str(self.data["lat"]),
97		'longitude': str(self.data["lon"]),
98		'speed': str(self.data["speed"]),
99		'battery': str(self.data["battery"]),
100		'station_id': str(self.data["station"]),
101		'price_id': "1"
102		}
103		}
104
105	1	try:
106	1	requests.post(self._URL, json = payload, headers = self._HEADERS)
107	1	except (Exception, ConnectionError) as error:
108	1	print(error)
109
110
111
112	1	def create_log(self) -> None:
113		"""
114		Create log. Data to be added is scooter's position, start date/time and scooter/user id.
115		"""
116	1	mutation = ''' mutation createLog(
117		$scooter_id: String!,
118		$customer_id: String!,
119		$start_longitude: String!,
120		$start_latitude: String!,
121		$price_id: String!) {
122		createLog(
123		scooter_id: $scooter_id,
124		customer_id: $customer_id,
125		start_longitude: $start_longitude,
126		start_latitude: $start_latitude,
127		price_id: $price_id) { id }
128		} '''
129
130	1	payload = {
131		'query': mutation,
132		'variables': {
133		'scooter_id': self.data["id"],
134		'customer_id': str(self.user_id),
135		'start_longitude': str(self.data["lon"]),
136		'start_latitude': str(self.data["lat"]),
137		'price_id': "1"
138		}
139		}
140
141	1	try:
142	1	response = requests.post(self._URL, json = payload, headers = self._HEADERS)
143
144		## save the log id
145	1	self.log_id = response.json()["data"]["createLog"]["id"]
146	1	except (Exception, ConnectionError) as error:
147	1	print(error)
148
149
150	1	def update_log(self) -> None:
151		""" Update log. Data to be updated is scooter's position and end date/time. """
152	1	mutation = ''' mutation updateLogByLogId(
153		$id: String!,
154		$end_time: String!,
155		$end_longitude: String!,
156		$end_latitude: String!) {
157		updateLogByLogId(
158		id: $id,
159		end_time: $end_time,
160		end_longitude: $end_longitude,
161		end_latitude: $end_latitude) { id }
162		} '''
163
164	1	payload = {
165		'query': mutation,
166		'variables': {
167		'id': self.log_id,
168		'end_time': datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
169		'end_longitude': str(self.data["lon"]),
170		'end_latitude': str(self.data["lat"]),
171		}
172		}
173
174	1	try:
175	1	requests.post(self._URL, json = payload, headers = self._HEADERS)
176	1	except (Exception, ConnectionError) as error:
177	1	print(error)
178
179
180	1	def get_city_data(self) -> dict:
181		"""
182		Get city's center position, id and area where the scooter is located.
183		And adds it to city dictionary.
184		"""
185	1	query = ''' query getCityByScooterId($id: String!) {
186		getCityByScooterId(id: $id) {
187		id
188		latitude
189		longitude
190		area
191		}
192		} '''
193
194	1	payload = {
195		'query': query,
196		'variables': {
197		'id': self.data["id"]
198		}
199		}
200
201	1	try:
202	1	response = requests.post(self._URL, json = payload, headers = self._HEADERS)
203
204	1	return response.json()["data"]["getCityByScooterId"][0]
205	1	except (Exception, ConnectionError):
206	1	return -1
207
208
209	1	def get_station(self, zone_id: str) -> dict:
210		"""
211		return random charging/maintenance station data in the city where the scooter is located.
212		Zone id: 1- Charging Station, 2- Parking Station, 3- Bike Statione, 4- Maintenance Station.
213		"""
214	1	query = ''' query getStationByCityIdAndZoneId($cityId: String!, $zoneId: String!) {
215		getStationByCityIdAndZoneId(cityId: $cityId, zoneId: $zoneId) {
216		id
217		latitude
218		longitude
219		}
220		} '''
221
222	1	payload = {
223		'query': query,
224		'variables': {
225		'cityId': self.city["id"],
226		'zoneId': zone_id
227		}
228		}
229
230	1	try:
231	1	response = requests.post(self._URL, json = payload, headers = self._HEADERS)
232
233	1	return response.json()["data"]["getStationByCityIdAndZoneId"][0]
234	1	except (Exception, ConnectionError):
235	1	return -1
236
237
238	1	def get_all_customers(self) -> list:
239		""" Get all customers data from API. """
240	1	query = ''' query { getAllCustomers { id } }'''
241
242	1	try:
243	1	response = requests.post(self._URL, json = {'query': query}, headers = self._HEADERS)
244
245	1	return response.json()["data"]["getAllCustomers"]
246	1	except (Exception, ConnectionError):
247		return -1
248

Rahn20 / spark-program

Push — main ( ce71f4...8ca48e )

src.api.ApiData.get_all_customers() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like