Coverage for mindsdb / integrations / handlers / aqicn_handler / aqicn.py: 0%

1import requests 

2 

3 

4class AQIClient: 

5 def __init__(self, api_key): 

6 self.api_key = api_key 

7 self.params = {"token": api_key} 

8 self.base_endpoint = "https://api.waqi.info/feed" 

9 

10 def make_request(self, url, additionalParams={}): 

11 newParams = {**self.params, **additionalParams} 

12 resp = requests.get(url, params=newParams) 

13 res = resp.json() 

14 content = {} 

15 if res["status"] == "ok": 

16 content = {'content': resp.json(), 'code': 200} 

17 else: 

18 content = {'content': resp.json(), 'code': 404} 

19 return content 

20 

21 def air_quality_city(self, city): 

22 url = f'{self.base_endpoint}/{city}/' 

23 return self.make_request(url) 

24 

25 def air_quality_lat_lng(self, lat, lng): 

26 url = f'{self.base_endpoint}/geo:{lat};{lng}/' 

27 return self.make_request(url) 

28 

29 def air_quality_user_location(self): 

30 url = f'{self.base_endpoint}/here/' 

31 return self.make_request(url) 

32 

33 def air_quality_station_by_name(self, name): 

34 url = 'https://api.waqi.info/search/' 

35 return self.make_request(url, {"keyword": name})