src.api.ApiData.get_city_data() - Code Metrics - Inspection of "creates change location/check zone and maintenance..." - Rahn20/spark-program - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( 20f858...b2f07a )

by Rahn20

created 2022-12-27 00:23 UTC

src.api.ApiData.get_city_data() A

↳ Parent: src.api

Complexity

Conditions

Size

Total Lines	30
Code Lines	8

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	2

Importance

Changes

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

#!/usr/bin/python3

"""
url 1: https://spark.isal.live/api/v1/
url 2: 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/"

    ## 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 = int


    def check_scooter_status(self, scooter_id: int) -> bool:
        """
        Check scooter status. Returns true if the scooter is available
        and adds the scooter's data to data dictionary.
        """
        ## Create the GraphQL query
        #query = ''' query getScooterById($id: Int) {
        #    getScooterById(id: $id) {
        #        latitude
        #        longitude
        #        speed
        #        battery
        #        status
        #    }
        #} '''

        #payload = {
        #    'query': query,
        #    'variables': {
        #        'id': scooter_id
        #    }
        #}

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

        # if status == "available":
        # self.add_scooter_data(response.json())
        #  return True
        # otherwise False

        # returns true to test
        self.data["id"] = scooter_id
        return True


    def connect_user(self) -> None:
        """ Connect user to scooter. """
        mutation = ''' mutation updateScooter($id: Int!, $user_id: Int!) {
            updateScooter(id: $id, user_id: $user_id) {
                data
            }
        } '''

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

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


    def remove_user_connection(self) -> None:
        """ Remove connection between user and scooter. """
        mutation = ''' mutation updateScooter($id: Int!, $user_id: Int!) {
            updateScooter(id: $id, user_id: $user_id) {
                data
            }
        } '''

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

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


    def update_scooter(self) -> None:
        """ Update api with scooter's new position, speed, status and battery level. """
        mutation = ''' mutation updateScooter(
            $id: Int!,
            $user_id: Int!,
            $latitude: Float!,
            $longitude: Float!,
            $speed: Int!,
            $battery: Int!,
            $status: String!) {
                updateScooter(
                    id: $id,
                    user_id: $user_id,
                    latitude: $latitude,
                    longitude: $longitude,
                    speed: $speed,
                    battery: $battery
                    status: $status) { data }
        } '''

        payload = {
            'query': mutation,
            'variables': {
                'id': self.data["id"],
                'user_id': self._user_id,
                'latitude': self.data["lat"],
                'longitude': self.data["lon"],
                'speed': self.data["speed"],
                'battery': self.data["battery"],
                'status': self.data["status"]
            }
        }

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


    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: Int!,
            $user_id: Int!,
            $start_time: Datetime,
            $start_longitude: Float!,
            $start_latitude: Float!) {
                createLog(
                    scooter_id: $scooter_id,
                    user_id: $user_id,
                    start_time: $start_time,
                    start_longitude: $start_longitude,
                    start_latitude: $start_latitude) { id }
        } '''

        payload = {
            'query': mutation,
            'variables': {
                'scooter_id': self.data["id"],
                'user_id': self._user_id,
                'start_time': datetime.now(),
                'start_longitude': self.data["lon"],
                'start_latitude': self.data["lat"],
            }
        }

        response = requests.post(self._URL, json = payload, headers = self._HEADERS)
        # save the log id
        # self._log_id =
        print(response.json())


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

        payload = {
            'query': mutation,
            'variables': {
                'id': self._log_id,
                'end_time': datetime.now(),
                'end_longitude': self.data["lon"],
                'end_latitude': self.data["lat"]
            }
        }

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


    def get_city_data(self) -> None:
        """
        Get city's center position, id and area where the scooter is running.
        And adds it to city dictionary.
        """
        query = ''' query getCityData($scooter_id: Int) {
            getCityData(scooter_id: $scooter_id) {
                id
                latitude
                longitude
                area
            }
        } '''

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

        response = requests.post(self._URL, json = payload, headers = self._HEADERS)
        #
        #self.city = {
        #    "id": ,
        #    "area":,      # km²
        #    "lat": ,    # coordinates
        #    "lon":      # coordinates
        #}
        print(response.json())


    def get_station(self, station_type: str) -> dict:
        """
        return random charging/maintenance station data in the city where the scooter is running.
        """
        query = ''' query getStation($city_id: Int, $type: String) {
            getStation(city_id: $city_id, type: $type) {
                id
                latitude
                longitude
            }
        } '''

        payload = {
            'query': query,
            'variables': {
                'city_id': self.city["id"],
                'type': station_type
            }
        }

        response = requests.post(self._URL, json = payload, headers = self._HEADERS)
        print(response.json())
        # return the response.join() data


    def create_scooter(self) -> None:
        """ Create a new scooter."""

    def create_user(self) -> None:
        """ Create a new user."""


1			#!/usr/bin/python3
2
3			"""
4			url 1: https://spark.isal.live/api/v1/
5			url 2: http://localhost:1337/api/v1/
6			"""
7
8			from datetime import datetime
9			import requests
10
11			from src.scooter import Scooter
12
13			class ApiData(Scooter):
14			""" Api class """
15
16			## API endpoint URL
17			_URL = "http://localhost:1337/api/v1/"
18
19			## Set the headers
20			_HEADERS = {'Content-Type': 'application/json'}
21
22
23			def __init__(self, user_id: int) -> None:
24			""" Initialize class """
25			super().__init__()
26			self._user_id = user_id
27			self._log_id = int
28
29
30			def check_scooter_status(self, scooter_id: int) -> bool:
31			"""
32			Check scooter status. Returns true if the scooter is available
33			and adds the scooter's data to data dictionary.
34			"""
35			## Create the GraphQL query
36			#query = ''' query getScooterById($id: Int) {
37			# getScooterById(id: $id) {
38			# latitude
39			# longitude
40			# speed
41			# battery
42			# status
43			# }
44			#} '''
45
46			#payload = {
47			# 'query': query,
48			# 'variables': {
49			# 'id': scooter_id
50			# }
51			#}
52
53			## Send the POST request
54			#response = requests.post(self._URL, json = payload, headers = self._HEADERS)
55			#print(response.json())
56			# print(query)
57
58			# if status == "available":
59			# self.add_scooter_data(response.json())
60			# return True
61			# otherwise False
62
63			# returns true to test
64			self.data["id"] = scooter_id
65			return True
66
67
68			def connect_user(self) -> None:
69			""" Connect user to scooter. """
70			mutation = ''' mutation updateScooter($id: Int!, $user_id: Int!) {
71			updateScooter(id: $id, user_id: $user_id) {
72			data
73			}
74			} '''
75
76			payload = {
77			'query': mutation,
78			'variables': {
79			'id': self.data["id"],
80			'user_id': self._user_id
81			}
82			}
83
84			response = requests.post(self._URL, json = payload, headers = self._HEADERS)
85			print(response.json())
86
87
88			def remove_user_connection(self) -> None:
89			""" Remove connection between user and scooter. """
90			mutation = ''' mutation updateScooter($id: Int!, $user_id: Int!) {
91			updateScooter(id: $id, user_id: $user_id) {
92			data
93			}
94			} '''
95
96			payload = {
97			'query': mutation,
98			'variables': {
99			'id': self.data["id"],
100			'user_id': None
101			}
102			}
103
104			response = requests.post(self._URL, json = payload, headers = self._HEADERS)
105			print(response.json())
106
107
108			def update_scooter(self) -> None:
109			""" Update api with scooter's new position, speed, status and battery level. """
110			mutation = ''' mutation updateScooter(
111			$id: Int!,
112			$user_id: Int!,
113			$latitude: Float!,
114			$longitude: Float!,
115			$speed: Int!,
116			$battery: Int!,
117			$status: String!) {
118			updateScooter(
119			id: $id,
120			user_id: $user_id,
121			latitude: $latitude,
122			longitude: $longitude,
123			speed: $speed,
124			battery: $battery
125			status: $status) { data }
126			} '''
127
128			payload = {
129			'query': mutation,
130			'variables': {
131			'id': self.data["id"],
132			'user_id': self._user_id,
133			'latitude': self.data["lat"],
134			'longitude': self.data["lon"],
135			'speed': self.data["speed"],
136			'battery': self.data["battery"],
137			'status': self.data["status"]
138			}
139			}
140
141			response = requests.post(self._URL, json = payload, headers = self._HEADERS)
142			print(response.json())
143
144
145			def create_log(self) -> None:
146			"""
147			Create log. Data to be added is scooter's position, start date/time and scooter/user id.
148			"""
149			mutation = ''' mutation createLog(
150			$scooter_id: Int!,
151			$user_id: Int!,
152			$start_time: Datetime,
153			$start_longitude: Float!,
154			$start_latitude: Float!) {
155			createLog(
156			scooter_id: $scooter_id,
157			user_id: $user_id,
158			start_time: $start_time,
159			start_longitude: $start_longitude,
160			start_latitude: $start_latitude) { id }
161			} '''
162
163			payload = {
164			'query': mutation,
165			'variables': {
166			'scooter_id': self.data["id"],
167			'user_id': self._user_id,
168			'start_time': datetime.now(),
169			'start_longitude': self.data["lon"],
170			'start_latitude': self.data["lat"],
171			}
172			}
173
174			response = requests.post(self._URL, json = payload, headers = self._HEADERS)
175			# save the log id
176			# self._log_id =
177			print(response.json())
178
179
180			def update_log(self) -> None:
181			""" Update log. Data to be updated is scooter's position and end date/time. """
182			mutation = ''' mutation updateLog(
183			$id: Int!,
184			$end_time: Datetime,
185			$end_longitude: Float!,
186			$end_latitude: Float!) {
187			updateLog(
188			id: $id,
189			end_time: $end_time,
190			end_longitude: $end_longitude
191			end_latitude: $end_latitude)
192			{ data }
193			} '''
194
195			payload = {
196			'query': mutation,
197			'variables': {
198			'id': self._log_id,
199			'end_time': datetime.now(),
200			'end_longitude': self.data["lon"],
201			'end_latitude': self.data["lat"]
202			}
203			}
204
205			response = requests.post(self._URL, json = payload, headers = self._HEADERS)
206			print(response.json())
207
208
209			def get_city_data(self) -> None:
210			"""
211			Get city's center position, id and area where the scooter is running.
212			And adds it to city dictionary.
213			"""
214			query = ''' query getCityData($scooter_id: Int) {
215			getCityData(scooter_id: $scooter_id) {
216			id
217			latitude
218			longitude
219			area
220			}
221			} '''
222
223			payload = {
224			'query': query,
225			'variables': {
226			'scooter_id': self.data["id"]
227			}
228			}
229
230			response = requests.post(self._URL, json = payload, headers = self._HEADERS)
231			#
232			#self.city = {
233			# "id": ,
234			# "area":, # km²
235			# "lat": , # coordinates
236			# "lon": # coordinates
237			#}
238			print(response.json())
239
240
241			def get_station(self, station_type: str) -> dict:
242			"""
243			return random charging/maintenance station data in the city where the scooter is running.
244			"""
245			query = ''' query getStation($city_id: Int, $type: String) {
246			getStation(city_id: $city_id, type: $type) {
247			id
248			latitude
249			longitude
250			}
251			} '''
252
253			payload = {
254			'query': query,
255			'variables': {
256			'city_id': self.city["id"],
257			'type': station_type
258			}
259			}
260
261			response = requests.post(self._URL, json = payload, headers = self._HEADERS)
262			print(response.json())
263			# return the response.join() data
264
265
266			def create_scooter(self) -> None:
267			""" Create a new scooter."""
268
269			def create_user(self) -> None:
270			""" Create a new user."""
271

Rahn20 / spark-program

Push — main ( 20f858...b2f07a )

src.api.ApiData.get_city_data() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like