Last active
May 23, 2022 06:14
-
-
Save chenen3/9eeae46124f6f5553f22a9dec545797c to your computer and use it in GitHub Desktop.
tornado helper
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| class BaseHandler(tornado.web.RequestHandler): | |
| def __init__(self, application, request, **kwargs): | |
| super(BaseHandler, self).__init__(application, request, **kwargs) | |
| @property | |
| def db(self): | |
| return self.application.db | |
| @staticmethod | |
| def current_timestamp(): | |
| return int(time.time()) | |
| def get_current_user(self): | |
| user_id_cookie = self.get_secure_cookie('user_id') | |
| if not user_id_cookie: | |
| return None | |
| return self.db.get('select id, username, create_time from user ' | |
| 'where id=%s', user_id_cookie) | |
| @property | |
| def reqargs(self): | |
| """parse and return request arguments""" | |
| return parse_request_arguments(self.request) | |
| def write_error(self, status_code, **kwargs): | |
| if 'exc_info' in kwargs: | |
| e = kwargs['exc_info'][1] | |
| if isinstance(e, Error): | |
| error_code = e.message | |
| self.set_status(400) | |
| res = make_fail_response_data(error_code) | |
| else: | |
| error_code = 500 | |
| error_msg = e.message or e.__class__.__name__ | |
| logger.error(error_msg) | |
| res = make_fail_response_data(error_code, error_msg) | |
| return self.write(res) | |
| super(BaseHandler, self).write_error(status_code, **kwargs) | |
| def response(self, data=None): | |
| return self.write(make_succ_response_data(data)) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import tornado | |
| def _parse_request_arguments(raw_request_arguments): | |
| ret = {} | |
| for name, value in raw_request_arguments.items(): | |
| ret[name] = value[0] if len(value) == 1 else value | |
| return ret | |
| def parse_request_arguments(request): | |
| """Parse Tornado HTTPRequest arguments | |
| @param request: RequestHandler.request object | |
| @return: dictionary contains request arguments | |
| @rtype: dict | |
| """ | |
| ret = {} | |
| if request.method == 'GET': | |
| ret = _parse_request_arguments(request.arguments) | |
| else: | |
| try: | |
| ret = tornado.escape.json_decode(request.body) | |
| except ValueError: | |
| ret = _parse_request_arguments(request.arguments) | |
| return ret |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment