| |
| try: |
| import Queue |
| except ImportError: |
| # pylint: disable=F0401 |
| # http is a Python3 module, replacing httplib. Ditto. |
| import queue as Queue |
| import threading |
| |
| try: |
| import httplib |
| except ImportError: |
| # pylint: disable=F0401 |
| from http import client as httplib |
| |
| try: |
| from urllib import urlencode |
| except ImportError: |
| # pylint: disable=F0401,E0611 |
| from urllib.parse import urlencode |
| |
| import datetime |
| import json |
| import logging |
| |
| # use generators for python2 and python3 |
| try: |
| xrange |
| except NameError: |
| xrange = range |
| |
| # some constants |
| MAX_RETRY = 1 # 0 means no retry |
| |
| |
| # logger |
| logger = None |
| DEBUG_LOG = False |
| |
| |
| def enable_log(filename=None): |
| global logger |
| global DEBUG_LOG |
| timestamp = datetime.datetime.today() |
| if not filename: |
| logfile = "./log/predictionio_%s.log" % timestamp.strftime( |
| "%Y-%m-%d_%H:%M:%S.%f") |
| else: |
| logfile = filename |
| logging.basicConfig(filename=logfile, |
| filemode='w', |
| level=logging.DEBUG, |
| format='[%(levelname)s] %(name)s (%(threadName)s) %(message)s') |
| logger = logging.getLogger(__name__) |
| DEBUG_LOG = True |
| |
| |
| class PredictionIOAPIError(Exception): |
| pass |
| |
| |
| class NotSupportMethodError(PredictionIOAPIError): |
| pass |
| |
| |
| class ProgramError(PredictionIOAPIError): |
| pass |
| |
| |
| class AsyncRequest(object): |
| |
| """AsyncRequest object |
| |
| """ |
| |
| def __init__(self, method, path, **params): |
| self.method = method # "GET" "POST" etc |
| # the sub path eg. POST /v1/users.json GET /v1/users/1.json |
| self.path = path |
| # dictionary format eg. {"appkey" : 123, "id" : 3} |
| self.params = params |
| # use queue to implement response, store AsyncResponse object |
| self.response_q = Queue.Queue(1) |
| self.qpath = "%s?%s" % (self.path, urlencode(self.params)) |
| self._response = None |
| # response function to be called to handle the response |
| self.rfunc = None |
| |
| def __str__(self): |
| return "%s %s %s %s" % (self.method, self.path, self.params, |
| self.qpath) |
| |
| def set_rfunc(self, func): |
| self.rfunc = func |
| |
| def set_response(self, response): |
| """ store the response |
| |
| NOTE: Must be only called once |
| """ |
| self.response_q.put(response) |
| |
| def get_response(self): |
| """Get the response. Blocking. |
| |
| :returns: self.rfunc's return type. |
| """ |
| if self._response is None: |
| tmp_response = self.response_q.get(True) # NOTE: blocking |
| if self.rfunc is None: |
| self._response = tmp_response |
| else: |
| self._response = self.rfunc(tmp_response) |
| |
| return self._response |
| |
| |
| class AsyncResponse(object): |
| """Store the response of asynchronous request |
| |
| When get the response, user should check if error is None (which means no |
| Exception happens). |
| If error is None, then should check if the status is expected. |
| """ |
| |
| def __init__(self): |
| #: exception object if any happens |
| self.error = None |
| |
| self.version = None |
| self.status = None |
| self.reason = None |
| #: Response header. str |
| self.headers = None |
| #: Response body. str |
| self.body = None |
| #: Jsonified response body. Remains None if conversion is unsuccessful. |
| self.json_body = None |
| #: Point back to the AsyncRequest object |
| self.request = None |
| |
| def __str__(self): |
| return "e:%s v:%s s:%s r:%s h:%s b:%s" % (self.error, self.version, |
| self.status, self.reason, |
| self.headers, self.body) |
| |
| def set_resp(self, version, status, reason, headers, body): |
| self.version = version |
| self.status = status |
| self.reason = reason |
| self.headers = headers |
| self.body = body |
| # Try to extract the json. |
| try: |
| self.json_body = json.loads(body.decode('utf8')) |
| except ValueError as ex: |
| self.json_body = None |
| |
| def set_error(self, error): |
| self.error = error |
| |
| def set_request(self, request): |
| self.request = request |
| |
| |
| class PredictionIOHttpConnection(object): |
| |
| def __init__(self, host, https=True, timeout=5): |
| if https: # https connection |
| self._connection = httplib.HTTPSConnection(host, timeout=timeout) |
| else: |
| self._connection = httplib.HTTPConnection(host, timeout=timeout) |
| |
| def connect(self): |
| self._connection.connect() |
| |
| def close(self): |
| self._connection.close() |
| |
| def request(self, method, url, body={}, headers={}): |
| """ |
| http request wrapper function, with retry capability in case of error. |
| catch error exception and store it in AsyncResponse object |
| return AsyncResponse object |
| |
| Args: |
| method: http method, type str |
| url: url path, type str |
| body: http request body content, type dict |
| header: http request header , type dict |
| """ |
| |
| response = AsyncResponse() |
| |
| try: |
| # number of retry in case of error (minimum 0 means no retry) |
| retry_limit = MAX_RETRY |
| mod_headers = dict(headers) # copy the headers |
| mod_headers["Connection"] = "keep-alive" |
| enc_body = None |
| if body: # if body is not empty |
| #enc_body = urlencode(body) |
| #mod_headers[ |
| # "Content-type"] = "application/x-www-form-urlencoded" |
| enc_body = json.dumps(body) |
| mod_headers[ |
| "Content-type"] = "application/json" |
| #mod_headers["Accept"] = "text/plain" |
| except Exception as e: |
| response.set_error(e) |
| return response |
| |
| if DEBUG_LOG: |
| logger.debug("Request m:%s u:%s h:%s b:%s", method, url, |
| mod_headers, enc_body) |
| # retry loop |
| for i in xrange(retry_limit + 1): |
| try: |
| if i != 0: |
| if DEBUG_LOG: |
| logger.debug("retry request %s times" % i) |
| if self._connection.sock is None: |
| self._connection.connect() |
| self._connection.request(method, url, enc_body, mod_headers) |
| except Exception as e: |
| self._connection.close() |
| if i == retry_limit: |
| # new copy of e created everytime?? |
| response.set_error(e) |
| else: # NOTE: this is try's else clause |
| # connect() and request() OK |
| try: |
| resp = self._connection.getresponse() |
| except Exception as e: |
| self._connection.close() |
| if i == retry_limit: |
| response.set_error(e) |
| else: # NOTE: this is try's else clause |
| # getresponse() OK |
| resp_version = resp.version # int |
| resp_status = resp.status # int |
| resp_reason = resp.reason # str |
| # resp.getheaders() returns list of tuples |
| # converted to dict format |
| resp_headers = dict(resp.getheaders()) |
| # NOTE: have to read the response before sending out next |
| # http request |
| resp_body = resp.read() # str |
| response.set_resp(version=resp_version, status=resp_status, |
| reason=resp_reason, headers=resp_headers, |
| body=resp_body) |
| break # exit retry loop |
| # end of retry loop |
| if DEBUG_LOG: |
| logger.debug("Response %s", response) |
| return response # AsyncResponse object |
| |
| |
| def connection_worker(host, request_queue, https=True, timeout=5, loop=True): |
| """worker function which establishes connection and wait for request jobs |
| from the request_queue |
| |
| Args: |
| request_queue: the request queue storing the AsyncRequest object |
| valid requests: |
| GET |
| POST |
| DELETE |
| KILL |
| https: HTTPS (True) or HTTP (False) |
| timeout: timeout for HTTP connection attempts and requests in seconds |
| loop: This worker function stays in a loop waiting for request |
| For testing purpose only. should always be set to True. |
| """ |
| |
| connect = PredictionIOHttpConnection(host, https, timeout) |
| |
| # loop waiting for job form request queue |
| killed = not loop |
| |
| while True: |
| # print "thread %s waiting for request" % thread.get_ident() |
| request = request_queue.get(True) # NOTE: blocking get |
| # print "get request %s" % request |
| method = request.method |
| if method == "GET": |
| path = request.qpath |
| d = connect.request("GET", path) |
| elif method == "POST": |
| path = request.path |
| body = request.params |
| d = connect.request("POST", path, body) |
| elif method == "DELETE": |
| path = request.qpath |
| d = connect.request("DELETE", path) |
| elif method == "KILL": |
| # tell the thread to kill the connection |
| killed = True |
| d = AsyncResponse() |
| else: |
| d = AsyncResponse() |
| d.set_error(NotSupportMethodError( |
| "Don't Support the method %s" % method)) |
| |
| d.set_request(request) |
| request.set_response(d) |
| request_queue.task_done() |
| if killed: |
| break |
| |
| # end of while loop |
| connect.close() |
| |
| |
| class Connection(object): |
| |
| """abstract object for connection with server |
| |
| spawn multiple connection_worker threads to handle jobs in the queue q |
| """ |
| |
| def __init__(self, host, threads=1, qsize=0, https=True, timeout=5): |
| """constructor |
| |
| Args: |
| host: host of the server. |
| threads: type int, number of threads to be spawn |
| qsize: size of the queue q |
| https: indicate it is httpS (True) or http connection (False) |
| timeout: timeout for HTTP connection attempts and requests in |
| seconds |
| """ |
| self.host = host |
| self.https = https |
| self.q = Queue.Queue(qsize) # if qsize=0, means infinite |
| self.threads = threads |
| self.timeout = timeout |
| # start thread based on threads number |
| self.tid = {} # dictionary of thread object |
| |
| for i in xrange(threads): |
| tname = "PredictionIOThread-%s" % i # thread name |
| self.tid[i] = threading.Thread( |
| target=connection_worker, name=tname, |
| kwargs={'host': self.host, 'request_queue': self.q, |
| 'https': self.https, 'timeout': self.timeout}) |
| self.tid[i].setDaemon(True) |
| self.tid[i].start() |
| |
| def make_request(self, request): |
| """put the request into the q |
| """ |
| self.q.put(request) |
| |
| def pending_requests(self): |
| """number of pending requests in the queue |
| """ |
| return self.q.qsize() |
| |
| def close(self): |
| """close this Connection. Call this when main program exits |
| """ |
| # set kill message to q |
| for i in xrange(self.threads): |
| self.make_request(AsyncRequest("KILL", "")) |
| |
| self.q.join() # wait for q empty |
| |
| for i in xrange(self.threads): # wait for all thread finish |
| self.tid[i].join() |
