#!/usr/bin/env python # -*- encoding: utf-8 -*- """ @Version: 1.0 @Python Version:3.6.6 @Author: ludq1 @Email: ludq1@chinaunicom.cn @date: 2023/04/07 11:40:00 @Description: """ import logging from datetime import datetime from typing import Union, List from flask import request, Response from .globalconst import GlobalConst, UserInfoKeyConst from .globalutility import Utility from .my_anykeyvalue_utils import AnyKeyValueUtils from .my_baseexception import MyBaseException from .my_servicehandle import ServiceHandle from .my_stringutils import MyStringUtils from .my_utils import MyUtils class BaseHttpServlet: r''' 基础的HttpServlet类,集成常用变量和常用方法,加速开发 ''' _my_serviceurl: str = None _app_logger: logging.Logger = Utility.get_common_logger() def do_service(self) -> Response: r''' 接收请求的总的处理函数 Returns: ''' request_method: str = request.method if request_method.upper() == 'GET': return self.do_get() elif request_method.upper() == 'POST': return self.do_post() elif request_method.upper() == 'PUT': return self.do_put() elif request_method.upper() == 'DELETE': return self.do_delete() else: return self.common_do_service_with_unsupportmethod() def do_get(self) -> Response: r''' 处理 GET 请求 Returns: ''' # 默认 GET 和 POST 使用同一个处理方式 return self.do_post() def do_post(self) -> Response: r''' 处理 POST 请求 Returns: ''' return self.common_do_service() def do_put(self) -> Response: r''' 处理 PUT 请求 Returns: ''' # 默认不支持 PUT 方法 return self.common_do_service_with_unsupportmethod() def do_delete(self) -> Response: r''' 处理 DELETE 请求 Returns: ''' # 默认不支持 DELETE 方法 return self.common_do_service_with_unsupportmethod() def common_do_service_with_unsupportmethod(self) -> Response: r''' 通用的返回不支持的方法的处理方式 Returns: ''' return self.do_make_response(self.gen_err_dict('不支持的方法:{}'.format(request.method))) def common_do_service(self) -> Response: r''' 通用的处理请求的方法 Returns: ''' map_webarg: dict = self.get_mapwebarg() # 记录开始时间 start_datetime = datetime.now() try: self.loginfo('{}({}) 访问调用:{}, 参数:{}'.format('TODO', 'TODO', request.path, map_webarg)) response = self.my_do_get() # 记录访问日志 end_datetime = datetime.now() consume_seconds = (end_datetime - start_datetime).microseconds / 1000.0 self.loginfo('{}({}) 访问调用:{} 正常结束,耗时 {} 秒'.format('TODO', 'TODO', request.path, consume_seconds)) return response except Exception as exception: # 生成错误信息 ret_dict = MyBaseException.format_exception_to_standard_dict(exception) ret_str = MyUtils.dict2jsonstr(ret_dict) response = self.do_make_response(ret_str) end_datetime = datetime.now() consume_seconds = (end_datetime - start_datetime).microseconds / 1000.0 self.logexception( '{}({}) 访问调用:{} 异常结束,耗时 {} 秒,异常信息:{}'.format('TODO', 'TODO', request.path, consume_seconds, ret_str)) return response def my_do_get(self) -> Response: r''' common_do_service 中调用此函数 Returns: ''' return self.do_make_response(self.gen_err_dict('未实现')) @property def app_logger(self): r''' 获取应用的logger的便捷函数 Returns: ''' return self._app_logger @property def my_serviceurl(self): r''' 获取本服务的serviceurl,用于鉴权 Returns: ''' return self._my_serviceurl def logdebug(self, msg, *args, **kwargs) -> None: r''' 日志记录便捷函数 Returns: ''' self.app_logger.debug(msg, *args, **kwargs) def loginfo(self, msg, *args, **kwargs) -> None: r''' 日志记录便捷函数 Returns: ''' self.app_logger.info(msg, *args, **kwargs) def logwarning(self, msg, *args, **kwargs) -> None: r''' 日志记录便捷函数 Returns: ''' self.app_logger.warning(msg, *args, **kwargs) def logerror(self, msg, *args, **kwargs) -> None: r''' 日志记录便捷函数 Returns: ''' self.app_logger.error(msg, *args, **kwargs) def logexception(self, msg, *args, exc_info=True, **kwargs) -> None: r''' 日志记录便捷函数 Returns: ''' self.app_logger.exception(msg, *args, exc_info=exc_info, **kwargs) def get_mapwebarg(self) -> dict: r''' 获取request中的mapWebArg,用于在处理request的过程中更改请求参数, 获取原始的请求参数请使用 request.values.to_dict Returns: ''' return ServiceHandle.get_mapwebarg() def gen_ok_dict(self) -> dict: r''' 生成标准的 返回给前端的jsonObject,ok Returns: ''' return MyBaseException.gen_ok_dict() def gen_err_dict(self, ret_val: str, ret_code: str = GlobalConst.RETCODE_COMMON_ERROR, submitted_msg: str = None) -> dict: r''' 生成标准的 返回给前端的jsonObject,err Returns: ''' return MyBaseException.format_to_standard_dict(ret_val=ret_val, ret_code=ret_code, submitted_msg=submitted_msg) def gen_err_dict_for_exception(self, exception: Exception) -> dict: r''' Args: exception: Returns: ''' return MyBaseException.format_exception_to_standard_dict(exception) def get_othermsg_from_exception(self, exception: Exception) -> str: r''' 如果e是MyBaseException,则获取e.getSubmittedWebArg(),否则返回StringUtils.EMPTY Args: exception: Returns: ''' return ServiceHandle.get_othermsg_from_exception(exception=exception) def do_make_response(self, retinfo: Union[str, dict]) -> Response: r''' 将 retinfo转成字符串作为返回结果, 尤其是 retinfo为字典时会将其转换为 json字符串 Args: retinfo: Returns: ''' return ServiceHandle.do_make_response(retinfo=retinfo) def check_is_success(self, ret_dict: dict, expect_retcode: str = GlobalConst.RETCODE_SUCESS_CODE) -> bool: r''' 快速判断标准格式的json的retcode是否为1 Args: ret_dict: Returns: ''' # 当 ret_dict 不为 None 且其 RETCODE为期望值时返回True return ServiceHandle.check_is_success(ret_dict=ret_dict, expect_retcode=expect_retcode) def safe_get_first_dict_from_list(self, a_list: list) -> dict: r''' 尝试从 list 中获取第一个 dict 对象，获取失败则返回 None Args: a_list: Returns: ''' return ServiceHandle.safe_get_first_dict_from_list(a_list) def safe_get_datarows(self, a_dict: dict, key_name: str = GlobalConst.RETKEY_DATAROWS) -> list: r''' 尝试从 dict 获取指定 DataRows 表示的 list ，出错则返回一个空的 list, 这里的safe是指 key 不存在时不出错，但是key对应的对象或字符串不是 list 时仍然会出错 Args: a_dict: Returns: ''' return ServiceHandle.safe_get_list_from_dict(a_dict=a_dict, key_name=key_name) def querycount_db(self, service_url: str, query_str: str, dict_header: dict = None) -> int: r''' 查询数据库,执行count数据库查询,直接得到返回的整数 Args: service_url: query_str: Returns: ''' return ServiceHandle.querycount_db(service_url=service_url, query_str=query_str, dict_header=dict_header) def querydb(self, service_url: str, query_str: str, dict_header: dict = None) -> dict: r''' 查询数据库 Args: service_url: query_str: Returns: ''' return ServiceHandle.querydb(service_url=service_url, query_str=query_str, dict_header=dict_header) def operate_db(self, service_url: str, operate_sql_list: list, dict_header: dict = None) -> dict: r''' 操作数据库 Args: service_url: operate_sql_list: 要执行的sql语句列表 Returns: ''' return ServiceHandle.operate_db(service_url=service_url, operate_sql_list=operate_sql_list, dict_header=dict_header) def call_proc(self, service_url: str, proc_name: str, arg_dict: dict, dict_header: dict = None) -> dict: r''' 执行存储过程 Args: service_url: proc_name: Returns: ''' return ServiceHandle.call_proc(service_url=service_url, proc_name=proc_name, arg_dict=arg_dict, dict_header=dict_header) def get_userinfo(self, global_obj=None, key_userinfo: str = GlobalConst.G_KEY_USERINFO) -> dict: r''' 从全局缓存中获取用户信息 Args: global_obj: Returns: ''' return ServiceHandle.get_userinfo(global_obj=global_obj, key_userinfo=key_userinfo) def get_key_from_userinfo(self, global_obj=None, key_userinfo: str = GlobalConst.G_KEY_USERINFO, key: str = None, default: str = MyStringUtils.EMPTY) -> str: r''' 从全局缓存中获取用户信息 , 然后获取 key 对应的内容 , 不会返回None,只会返回 EMPTY Args: global_obj: key: default: Returns: ''' return ServiceHandle.get_key_from_userinfo(global_obj=global_obj, key_userinfo=key_userinfo, key=key, default=default) def get_usersystem(self, global_obj=None, key_userinfo: str = GlobalConst.G_KEY_USERINFO, key_usersystem: str = UserInfoKeyConst.USER_SYSTEM) -> str: r''' 从全局缓存中获取用户信息 , 然后获取 usersystem , 不会返回None,如果key不存在或dict不存在,返回 default Args: global_obj: key_usersystem: Returns: ''' return ServiceHandle.get_usersystem(global_obj=global_obj, key_userinfo=key_userinfo, key_usersystem=key_usersystem) def get_userid(self, global_obj=None, key_userinfo: str = GlobalConst.G_KEY_USERINFO, key_userid: str = UserInfoKeyConst.USER_ID) -> str: r''' 从全局缓存中获取用户信息 , 然后获取 userid , 不会返回None,只会返回 EMPTY Args: global_obj: key_userinfo: Returns: ''' return ServiceHandle.get_userid(global_obj=global_obj, key_userinfo=key_userinfo, key_userid=key_userid) def is_login(self, global_obj=None, key_userinfo: str = GlobalConst.G_KEY_USERINFO, key_userid: str = UserInfoKeyConst.USER_ID, key_usersystem: str = UserInfoKeyConst.USER_SYSTEM, default_usersystem: str = None, expect_usersystem: str = None) -> str: r''' 从全局缓存中获取用户信息 , 然后获取 userid 和 usersystem, 判断用户的 usersystem 是否和预期的 usersystem 相同,并且有 登陆的userid Args: global_obj: key_userinfo: key_userid: key_usersystem: default_usersystem: 默认为 default expect_usersystem: 默认为 default Returns: ''' return ServiceHandle.is_login(global_obj=global_obj, key_userinfo=key_userinfo, key_userid=key_userid, key_usersystem=key_usersystem, default_usersystem=default_usersystem, expect_usersystem=expect_usersystem) def get_usertenantid(self, global_obj=None, key_userinfo: str = GlobalConst.G_KEY_USERINFO, key_tenantid: str = UserInfoKeyConst.TENANT_ID) -> str: r''' 从全局缓存中获取用户信息 , 然后获取 userid , 不会返回None,只会返回 EMPTY Args: global_obj: key_userinfo: key_tenantid: Returns: ''' return ServiceHandle.get_usertenantid(global_obj=global_obj, key_userinfo=key_userinfo, key_tenantid=key_tenantid) def get_userdeptid(self, global_obj=None, key_userinfo: str = GlobalConst.G_KEY_USERINFO, key_deptid: str = UserInfoKeyConst.DEPT_ID) -> str: r''' 从全局缓存中获取用户信息 , 然后获取 , 不会返回None,只会返回 EMPTY Args: global_obj: key_userinfo: key_tenantid: Returns: ''' return ServiceHandle.get_userdeptid(global_obj=global_obj, key_userinfo=key_userinfo, key_deptid=key_deptid) def get_userdept_shortname(self, global_obj=None, key_userinfo: str = GlobalConst.G_KEY_USERINFO, key_deptshortname: str = UserInfoKeyConst.DEPT_NAME_SHORT) -> str: r''' 从全局缓存中获取用户信息 , 然后获取 , 不会返回None,只会返回 EMPTY Args: global_obj: key_userinfo: key_deptshortname: Returns: ''' return ServiceHandle.get_userdept_shortname(global_obj=global_obj, key_userinfo=key_userinfo, key_deptshortname=key_deptshortname) def get_userdept_fullname(self, global_obj=None, key_userinfo: str = GlobalConst.G_KEY_USERINFO, key_deptfullname: str = UserInfoKeyConst.DEPT_NAME_FULL) -> str: r''' 从全局缓存中获取用户信息 , 然后获取 , 不会返回None,只会返回 EMPTY Args: global_obj: key_userinfo: key_deptfullname: Returns: ''' return ServiceHandle.get_userdept_fullname(global_obj=global_obj, key_userinfo=key_userinfo, key_deptfullname=key_deptfullname) def get_user_ouid(self, global_obj=None, key_userinfo: str = GlobalConst.G_KEY_USERINFO, key_ouid: str = UserInfoKeyConst.OU_ID) -> str: r''' 从全局缓存中获取用户信息 , 然后获取 , 不会返回None,只会返回 EMPTY Args: global_obj: key_userinfo: key_ouid: Returns: ''' return ServiceHandle.get_user_ouid(global_obj=global_obj, key_userinfo=key_userinfo, key_ouid=key_ouid) def get_user_ouname(self, global_obj=None, key_userinfo: str = GlobalConst.G_KEY_USERINFO, key_ouname: str = UserInfoKeyConst.OU_NAME) -> str: r''' 从全局缓存中获取用户信息 , 然后获取 , 不会返回None,只会返回 EMPTY Args: global_obj: key_userinfo: key_ouname: Returns: ''' return ServiceHandle.get_user_ouname(global_obj=global_obj, key_userinfo=key_userinfo, key_ouname=key_ouname) def get_user_email(self, global_obj=None, key_userinfo: str = GlobalConst.G_KEY_USERINFO, key_email: str = UserInfoKeyConst.EMAIL) -> str: r''' 从全局缓存中获取用户信息 , 然后获取 , 不会返回None,只会返回 EMPTY Args: global_obj: key_userinfo: key_email: Returns: ''' return ServiceHandle.get_user_email(global_obj=global_obj, key_userinfo=key_userinfo, key_email=key_email) def put_userinfo_to_dict(self, arg_dict: dict, self_serviceurl: str = None, global_obj=None, key_userinfo: str = GlobalConst.G_KEY_USERINFO, key_userid: str = UserInfoKeyConst.USER_ID, key_username: str = UserInfoKeyConst.USER_NAME, key_tenantid: str = UserInfoKeyConst.TENANT_ID, key_deptid: str = UserInfoKeyConst.DEPT_ID, key_deptshortname: str = UserInfoKeyConst.DEPT_NAME_SHORT, key_deptfullname: str = UserInfoKeyConst.DEPT_NAME_FULL, key_ouid: str = UserInfoKeyConst.OU_ID, key_ouname: str = UserInfoKeyConst.OU_NAME, key_email: str = UserInfoKeyConst.EMAIL) -> dict: r''' 将 g.userinfo 中的用户信息加入到 map 中, 必须保证 g.userinfo 有值 map一般就是mapWebArg, 加入的信息为 arg_myuserid arg_myusername arg_mytenantid arg_mydeptid arg_mydeptname arg_myouid arg_req_userid arg_req_tenantid 然后就是将要要鉴权的服务url添加进去, arg_req_serviceurl 是否覆盖前端传递的内容，由环境变量 DontOverrideUserInfoFromFront 决定 Args: arg_dict: self_serviceurl: Returns: ''' return ServiceHandle.put_userinfo_to_dict(arg_dict=arg_dict, self_serviceurl=self_serviceurl, global_obj=global_obj, key_userinfo=key_userinfo, key_userid=key_userid, key_username=key_username, key_tenantid=key_tenantid, key_deptid=key_deptid, key_deptshortname=key_deptshortname, key_deptfullname=key_deptfullname, key_ouid=key_ouid, key_ouname=key_ouname, key_email=key_email) def get_info_with_fromenv_first(self, info_obj: Union[dict, str], info_key: str): r''' 如果 环境变量 LOCAL_RUN = 1 , 则优先获取 info_key 的环境变量对应的 内容 , 否则获取 info_obj 中该key对应的 内容 或 info_obj 如果是字符串,则直接获取该字符串 Args: info_obj: 要获取的内容或字典 info_key: 要获取的内容或字典中该内容对应的key,也是环境变量名称 Returns: ''' return ServiceHandle.get_info_with_fromenv_first(info_obj=info_obj, info_key=info_key) def query_any_key(self, key_id: Union[List[str], str], key_type: str = AnyKeyValueUtils.DEFAULT_KEY_TYPE, table_name: str = AnyKeyValueUtils.DEFAULT_TABLENAME, col_keyid: str = AnyKeyValueUtils.DEFAULT_COLUMNNAME_KEYID, col_keytype: str = AnyKeyValueUtils.DEFAULT_COLUMNNAME_KEYTYPE, col_keyinfo=AnyKeyValueUtils.DEFAULT_COLUMNNAME_KEYINFO, service_name: str = AnyKeyValueUtils.DEFAULT_SERVICE_NAME, api_path=AnyKeyValueUtils.DEFAULT_READ_API_PATH) -> dict: r''' 查询二维码库中的 keyId和keyType对应的info , Args: key_id: key_type: Returns: ''' return AnyKeyValueUtils.query_any_key(key_id=key_id, key_type=key_type, table_name=table_name, col_keyid=col_keyid, col_keytype=col_keytype, col_keyinfo=col_keyinfo, service_name=service_name, api_path=api_path)