import requests from nullptr.models.system import System from nullptr.models.waypoint import Waypoint from nullptr.models.marketplace import Marketplace from nullptr.models.jumpgate import Jumpgate from .util import * from time import sleep class ApiError(Exception): def __init__(self, msg, code): super().__init__(msg) self.code = code class ApiLimitError(Exception): pass class Api: def __init__(self, store, agent): self.agent = agent self.store = store self.meta = None self.root = 'https://api.spacetraders.io/v2/' def token(self): if self.agent.token is None: raise ApiError('no token. Please register', 1337) return self.agent.token def request(self, method, path, data=None, need_token=True, params={}): try: return self.request_once(method, path, data, need_token, params) except ApiLimitError: print('oops, hit the limit. take a break') sleep(10) return self.request_once(method, path, data, need_token, params) def request_once(self, method, path, data=None, need_token=True, params={}): headers = {} if need_token: headers['Authorization'] = 'Bearer ' + self.token() if method == 'get': params['limit'] = 20 r = requests.request(method, self.root+path, json=data, headers=headers, params=params) if r.status_code == 429: raise ApiLimitError() result = r.json() self.last_result = result if result is None: raise ApiError('http call failed', r.status_code) if 'meta' in result: self.last_meta = result['meta'] if 'data' not in result: error_code = sg(result, 'error.code', -1) self.last_error = error_code error_message = sg(result, 'error.message') raise ApiError(error_message, error_code) else: self.last_error = 0 return result['data'] def register(self, faction): callsign = self.agent.symbol data = { 'symbol': callsign, 'faction': faction } result = self.request('post', 'register', data, need_token=False) token = mg(result, 'token') self.agent.token = token def info(self): data = self.request('get', 'my/agent') self.agent.update(data) return self.agent def list_systems(self, page=1): data = self.request('get', 'systems', params={'page': page}) #pprint(self.last_meta) return self.store.update_list(System, data) def list_waypoints(self, system): data = self.request('get', f'systems/{system}/waypoints/') # pprint(data) return self.store.update_list(Waypoint, data) def marketplace(self, waypoint): system = waypoint.system() symbol = str(waypoint) data = self.request('get', f'systems/{system}/waypoints/{waypoint}/market') return self.store.update(Marketplace, symbol, data) def jumps(self, waypoint): data = self.request('get', f'systems/{waypoint.system()}/waypoints/{waypoint}/jump-gate') symbol = str(waypoint) return self.store.update(Jumpgate, symbol, data)