Coverage for mindsdb / integrations / handlers / oilpriceapi_handler / oilpriceapi.py: 0%

1import requests 

2 

3 

4class OilPriceAPIClient: 

5 def __init__(self, api_key): 

6 self.api_key = api_key 

7 self.base_endpoint = "https://api.oilpriceapi.com/v1/prices" 

8 self.valid_values_by_type = ["spot_price", "daily_average_price"] 

9 self.valid_values_by_code = ["BRENT_CRUDE_USD", "WTI_USD"] 

10 

11 def make_request(self, url, params={}): 

12 headers = {'Content-type': 'application/json'} 

13 if self.api_key: 

14 headers['Authorization'] = 'Token ' + self.api_key 

15 resp = requests.get(url, headers=headers, params=params) 

16 content = {} 

17 if resp.status_code == 200: 

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

19 else: 

20 content = {'content': {}, 'code': resp.status_code, 'error': resp.text} 

21 return content 

22 

23 def _is_valid_by_type(self, val): 

24 return val in self.valid_values_by_type 

25 

26 def _is_valid_by_code(self, val): 

27 return val in self.valid_values_by_code 

28 

29 def create_params_dict(self, by_type, by_code): 

30 params = {} 

31 if by_type is not None: 

32 params["by_type"] = by_type 

33 if by_code is not None: 

34 params["by_code"] = by_code 

35 return params 

36 

37 def get_latest_price(self, by_type=None, by_code=None): 

38 url = f'{self.base_endpoint}/latest/' 

39 params = self.create_params_dict(by_type=by_type, by_code=by_code) 

40 return self.make_request(url, params=params) 

41 

42 def get_price_past_day(self, by_type=None, by_code=None): 

43 url = f'{self.base_endpoint}/past_day/' 

44 params = self.create_params_dict(by_type=by_type, by_code=by_code) 

45 return self.make_request(url, params=params)