烧瓶装潢师用cerberus检查烧瓶，烧瓶返回结果自动转换
flask-ext-ydf的Python项目详细描述
1. pip install flask_ext_ydf
1.自动使用redis统计每个ip和每个接口的访问次数
2.自动返回code data message 三个字段格式的json。 兼容flask接口已经返回了完整json和返回 列表 数组
- 使用装饰器做参数校验，校验格式使用cerberus包的语法。
"""各种flask 扩展"""importosimporttracebackimportrandomimportbase64importfunctoolsfrombsonimportjson_utilimportjsonimportredisfromflaskimportcurrent_app,request,Flaskfromflask.globalsimport_request_ctx_stackfromcerberusimportValidatorfromnb_logimportLogManager,LoggerMixinflask_error_logger=LogManager('flask_error').get_logger_and_add_handlers(log_filename='flask_error.log')flask_record_logger=LogManager('flask_record').get_logger_and_add_handlers(log_filename='flask_record.log')classFlaskIpStatistics():""" 自动统计每个接口的访问情况 """def__init__(self,app=None):self.app=appifappisnotNone:self.init_app(app,)definit_app(self,app:Flask):if'REDSI_URL'notinapp.config:raiseLookupError('请在flask的config 配置中指明 REDSI_URL 的连接配置')if'SHOW_IP_STATISTIC_PATH'notinapp.config:raiseLookupError('请在flask的config 配置中指明 SHOW_IP_STATISTIC_PATH 的配置,用来返回ip访问情况')if'REDSI_KEY_PREFIX_FOR_STATISTICS'notinapp.config:raiseLookupError('请在flask的config 配置中指明 REDSI_KEY_PREFIX_FOR_STATISTICS 的配置')if'SHOW_IP_STATISTIC_PATH'notinapp.config:raiseLookupError('请在flask的config 配置中指明 SHOW_IP_STATISTIC_PATH 的配置,用来返回ip访问情况')self._redsi_key_prefix_of_app=app.config['REDSI_KEY_PREFIX_FOR_STATISTICS']self._redis_db_for_ip_statistic=redis.Redis.from_url(app.config['REDSI_URL'])app.before_request_funcs.setdefault(None,[]).append(self._inrc_ip)app.add_url_rule(app.config['SHOW_IP_STATISTIC_PATH'],'',self._show_count,methods=['GET',])@staticmethoddef_get_user_ip():ifrequest.headers.get('X-Forwarded-For'):user_ip=request.headers['X-Forwarded-For']elifrequest.headers.get('X-Real-IP'):user_ip=request.headers.get('X-Real-IP')else:user_ip=request.remote_addrreturnuser_ip.split(',')[0]def_inrc_ip(self):print(request.path)ip_key_name=f'''{self._redsi_key_prefix_of_app}:{request.path}:{self._get_user_ip()}'''print(f'执行inrc {ip_key_name}')ifself._redis_db_for_ip_statistic.exists(ip_key_name):self._redis_db_for_ip_statistic.incr(ip_key_name,1)else:withself._redis_db_for_ip_statistic.pipeline()asp:p.incr(ip_key_name,1)p.expire(ip_key_name,3600)p.execute()def_show_count(self):# 显示每个ip的访问次数# return 'aaaaa'key_iters=self._redis_db_for_ip_statistic.scan_iter(f'{self._redsi_key_prefix_of_app}:*')ip__count_map={key.decode():self._redis_db_for_ip_statistic.get(key).decode()forkeyinkey_iters}return{'count':len(ip__count_map),'ip__count_map':ip__count_map}defapi_return_deco(v):""" 对flask的返回 加一个固定的状态码。在测试环境即使是非debug，直接在错误信息中返回错误堆栈。在生产环境使用随机四位字母 加 错误信息的base64作为错误信息。 :param v:视图函数 :return: """flask_request=request@functools.wraps(v)def_api_return_deco(*args,**kwargs):# noinspection PyBroadExceptiontry:data=v(*args,**kwargs)ifisinstance(data,str):result=dataelse:if'code'indataand'data'indata:result=json_util.dumps(data)else:result=json_util.dumps({"code":200,"data":data,"message":"SUCCESS"},ensure_ascii=False)flask_record_logger.debug(f'请求路径：{flask_request.path} 请求参数：{json.dumps(flask_request.values.to_dict())},返回正常,结果长度是{len(result)}')returnresultexceptExceptionase:except_str0=f'请求路径：{flask_request.path} 请求参数：{json.dumps(flask_request.values.to_dict())} ,出错了 {type(e)}{e}{traceback.format_exc()}'.replace('\n','<br>')flask_error_logger.exception(except_str0)exception_str_encode=base64.b64encode(except_str0.encode()).decode().replace('=','').strip()message=except_str0ifos.environ.get('IS_RETURN_PYTHON_TRACEBACK_PLAINTEXT_FROM_FLASK_INTERFACE')=='1'elsef'''{"".join(random.sample("abcdefghijklmnopqrstABCDEFGHIJKLMNOPQRST",4))}{exception_str_encode}'''returnjson.dumps({"code":500,"data":None,"message":message},ensure_ascii=False)return_api_return_decoflask_api_result_deco=api_return_decodef_dispatch_request_with_flask_api_result_deco(self):"""Does the request dispatching. Matches the URL and returns the return value of the view or error handler. This does not have to be a response object. In order to convert the return value to a proper response object, call :func:`make_response`. .. versionchanged:: 0.7 This no longer does the exception handling, this code was moved to the new :meth:`full_dispatch_request`. """req=_request_ctx_stack.top.requestifreq.routing_exceptionisnotNone:self.raise_routing_exception(req)rule=req.url_rule# if we provide automatic options for this URL and the# request came with the OPTIONS method, reply automaticallyifgetattr(rule,'provide_automatic_options',False) \ andreq.method=='OPTIONS':returnself.make_default_options_response()# otherwise dispatch to the handler for that endpoint# return self.view_functions[rule.endpoint](**req.view_args)v=self.view_functions[rule.endpoint]v2=flask_api_result_deco(v)returnv2(**req.view_args)classCustomFlaskApiConversion(LoggerMixin):""" 自动转化每个接口的返回，自动将各种类型转成code data message格式的json """def__init__(self,app=None):self.app=appifappisnotNone:self.init_app(app,)defmonkey_patch_dispatch_request(self):self.logger.warn('改变了flask的dispatch_request 方法')Flask.dispatch_request=_dispatch_request_with_flask_api_result_decodefinit_app(self,app:Flask):if'IS_RETURN_PYTHON_TRACEBACK_PLAINTEXT_FROM_FLASK_INTERFACE'notinapp.config:self.logger.warning('flask的config没有配置 IS_RETURN_PYTHON_TRACEBACK_PLAINTEXT_FROM_FLASK_INTERFACE，则默认为"0"，使用密文')os.environ.setdefault('IS_RETURN_PYTHON_TRACEBACK_PLAINTEXT_FROM_FLASK_INTERFACE','0')app.before_first_request_funcs.append(self.monkey_patch_dispatch_request)# 直接把返回装饰器加到app上，免得每个接口加一次装饰器麻烦。defflask_check_param_deco(schema,):""" 自动检查参数，返回400 code :param schema: :return: """def_check_param_deco(v):@functools.wraps(v)def___check_param_deco(*ags,**kwargs):request_values={}print(request.values)ifrequest.values:request_values=request.values.to_dict()ifrequest.json:request_values.update(request.json)# print(request_values)# print(schema)vd=Validator()vd.allow_unknown=True# document, schema=Noneis_ok=vd.validate(request_values,schema)check_errors=Noneifis_okisFalse:check_errors=vd.errorsifis_ok:returnv(*ags,**kwargs)else:return{'code':400,'message':check_errors,'data':None}return___check_param_decoreturn_check_param_decoif__name__=='__main__':schemax={"x":{'type':'string','empty':False,'nullable':False,'required':True}}app=Flask(__name__)app.config['REDSI_URL']='redis://127.0.0.1/0'app.config['REDSI_KEY_PREFIX_FOR_STATISTICS']='flask_test1'app.config['SHOW_IP_STATISTIC_PATH']='/proj/ip_st'FlaskIpStatistics(app)CustomFlaskApiConversion(app)@app.route('/',methods=['get'])defindex():return'hello'@app.route('/list',methods=['get','post'])@flask_check_param_deco(schemax,)deflistx():""" {"code": 200, "data": ["dsd"], "message": "SUCCESS"} :return: """return['dsd','lalala']# 可以直接返回字典 和列表类型，不需要json dumps。app.run()
- 项目
标签：
欢迎加入QQ群-->： 979659372
