import sys
import urllib.parse
import requests
class APIServer:
def __init__(self, server):
self.api_version = '0'
self.server = server
self.routes = {}
def route(self, method, path, handler):
if method not in self.routes:
self.routes[method] = {}
self.routes[method][path] = handler
def get_route(self, method, path=None):
if not path:
path = method
method = 'GET'
path = urllib.parse.urlparse(path).path
if method in self.routes:
if path in self.routes[method]:
print(method + ': ' + path)
return self.routes[method][path]
if 'error_404' in self.routes['__']:
print('__: error_404 (requested: ' + method + ': ' + path + ')')
return self.routes['__']['error_404']
return None
def init(self):
self.init_routes()
if '__' not in self.routes or 'error_404' not in self.routes['__']:
print('error: subclasses of "API" should define an "error_404" route with the "__" method')
sys.exit(1)
return self
def init_routes(self):
pass
@staticmethod
def get_url_params(request, response):
_ = response
data = urllib.parse.parse_qs(urllib.parse.urlparse(request.path).query)
return {x: data[x][0] for x in data}
@staticmethod
def get_post_data(request, response):
_ = response
content_length = request.headers.get('content-length')
if not content_length:
return {}
length = int(content_length)
data = str(request.rfile.read(length), 'utf8')
data = urllib.parse.parse_qs(urllib.parse.urlparse('?' + data).query)
return {x: data[x][0] for x in data}
class APIClient:
def __init__(self, host='localhost', port=44363, insecure=False):
self.api_version = '0'
self.host = host
self.port = port
self.insecure = insecure
self.session = requests.Session()
self.connected = False
def get_host_port(self):
return self.build_url(api_version=False)
def set_host_port(self, host, port):
if self.is_connected():
return False
self.host = host
self.port = port
return True
def set_insecure(self, insecure):
if self.is_connected():
return False
self.insecure = insecure
return True
def is_connected(self):
return self.connected
def build_url(self, path='', api_version=True):
scheme = 'https'
if self.insecure:
scheme = 'http'
if api_version:
path = '/v' + self.api_version + path
return urllib.parse.urljoin(scheme + '://' + self.host + ':' + str(self.port), path)