Skip to main content

flask decorators check with cerberus,flask return result auto convet

Project description

1. pip install flask_ext_ydf


2.自动返回code data message 三个字段格式的json。 兼容flask接口已经返回了完整json和返回 列表 数组

  1. 使用装饰器做参数校验,校验格式使用cerberus包的语法。
各种flask 扩展
import os
import traceback
import random
import base64
import functools
from bson import json_util
import json

import redis
from flask import current_app, request, Flask
from flask.globals import _request_ctx_stack
from cerberus import Validator
from nb_log import LogManager, LoggerMixin

flask_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')

class FlaskIpStatistics():

    def __init__(self, app=None): = app
        if app is not None:
            self.init_app(app, )

    def init_app(self, app: Flask):
        if 'REDSI_URL' not in app.config:
            raise LookupError('请在flask的config 配置中指明 REDSI_URL 的连接配置')
        if 'SHOW_IP_STATISTIC_PATH' not in app.config:
            raise LookupError('请在flask的config 配置中指明 SHOW_IP_STATISTIC_PATH 的配置,用来返回ip访问情况')
        if 'REDSI_KEY_PREFIX_FOR_STATISTICS' not in app.config:
            raise LookupError('请在flask的config 配置中指明 REDSI_KEY_PREFIX_FOR_STATISTICS 的配置')
        if 'SHOW_IP_STATISTIC_PATH' not in app.config:
            raise LookupError('请在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', ])

    def _get_user_ip():
        if request.headers.get('X-Forwarded-For'):
            user_ip = request.headers['X-Forwarded-For']
        elif request.headers.get('X-Real-IP'):
            user_ip = request.headers.get('X-Real-IP')
            user_ip = request.remote_addr
        return user_ip.split(',')[0]

    def _inrc_ip(self):
        ip_key_name = f'''{self._redsi_key_prefix_of_app}:{request.path}:{self._get_user_ip()}'''
        print(f'执行inrc    {ip_key_name}')
        if self._redis_db_for_ip_statistic.exists(ip_key_name):
            self._redis_db_for_ip_statistic.incr(ip_key_name, 1)
            with self._redis_db_for_ip_statistic.pipeline() as p:
                p.incr(ip_key_name, 1)
                p.expire(ip_key_name, 3600)

    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() for key in key_iters}
        return {'count': len(ip__count_map), 'ip__count_map': ip__count_map}

def api_return_deco(v):
    对flask的返回 加一个固定的状态码。在测试环境即使是非debug,直接在错误信息中返回错误堆栈。在生产环境使用随机四位字母 加 错误信息的base64作为错误信息。
    :param v:视图函数
    flask_request = request

    def _api_return_deco(*args, **kwargs):
        # noinspection PyBroadException
            data = v(*args, **kwargs)
            if isinstance(data, str):
                result = data
                if 'code' in data and 'data' in data:
                    result = json_util.dumps(data)
                    result = json_util.dumps({
                        "code": 200,
                        "data": data,
                        "message": "SUCCESS"}, ensure_ascii=False)
                f'请求路径:{flask_request.path}  请求参数:{json.dumps(flask_request.values.to_dict())},返回正常,结果长度是{len(result)}')
            return result
        except Exception as e:
            except_str0 = f'请求路径:{flask_request.path}  请求参数:{json.dumps(flask_request.values.to_dict())} ,出错了 {type(e)} {e} {traceback.format_exc()}'.replace(
                '\n', '<br>')
            exception_str_encode = base64.b64encode(except_str0.encode()).decode().replace('=', '').strip()
            message = except_str0 if os.environ.get(
            {"".join(random.sample("abcdefghijklmnopqrstABCDEFGHIJKLMNOPQRST", 4))}{exception_str_encode}'''
            return json.dumps({
                "code": 500,
                "data": None,
                "message": message}, ensure_ascii=False)

    return _api_return_deco

flask_api_result_deco = api_return_deco

def _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 =
    if req.routing_exception is not None:
    rule = req.url_rule
    # if we provide automatic options for this URL and the
    # request came with the OPTIONS method, reply automatically
    if getattr(rule, 'provide_automatic_options', False) \
            and req.method == 'OPTIONS':
        return self.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)
    return v2(**req.view_args)

class CustomFlaskApiConversion(LoggerMixin):
    自动转化每个接口的返回,自动将各种类型转成code data message格式的json

    def __init__(self, app=None): = app
        if app is not None:
            self.init_app(app, )

    def monkey_patch_dispatch_request(self):
        self.logger.warn('改变了flask的dispatch_request 方法')
        Flask.dispatch_request = _dispatch_request_with_flask_api_result_deco

    def init_app(self, app: Flask):
                'flask的config没有配置 IS_RETURN_PYTHON_TRACEBACK_PLAINTEXT_FROM_FLASK_INTERFACE,则默认为"0",使用密文')
        app.before_first_request_funcs.append(self.monkey_patch_dispatch_request)  # 直接把返回装饰器加到app上,免得每个接口加一次装饰器麻烦。

def flask_check_param_deco(schema, ):
    自动检查参数,返回400 code
    :param schema:

    def _check_param_deco(v):
        def ___check_param_deco(*ags, **kwargs):
            request_values = {}
            if request.values:
                request_values = request.values.to_dict()
            if request.json:
            # print(request_values)
            # print(schema)
            vd = Validator()
            vd.allow_unknown = True
            # document, schema=None
            is_ok = vd.validate(request_values, schema)
            check_errors = None
            if is_ok is False:
                check_errors = vd.errors
            if is_ok:
                return v(*ags, **kwargs)
                return {'code': 400,
                        'message': check_errors,
                        'data': None}

        return ___check_param_deco

    return _check_param_deco

if __name__ == '__main__':
    schemax = {"x": {'type': 'string', 'empty': False, 'nullable': False, 'required': True}}

    app = Flask(__name__)
    app.config['REDSI_URL'] = 'redis://'
    app.config['REDSI_KEY_PREFIX_FOR_STATISTICS'] = 'flask_test1'
    app.config['SHOW_IP_STATISTIC_PATH'] = '/proj/ip_st'

    @app.route('/', methods=['get'])
    def index():
        return 'hello'

    @app.route('/list', methods=['get', 'post'])
    @flask_check_param_deco(schemax, )
    def listx():
        {"code": 200, "data": ["dsd"], "message": "SUCCESS"}
        return ['dsd', 'lalala']  # 可以直接返回字典 和列表类型,不需要json dumps。

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Files for flask-ext-ydf, version 1.2
Filename, size File type Python version Upload date Hashes
Filename, size flask_ext_ydf-1.2.tar.gz (6.7 kB) File type Source Python version None Upload date Hashes View

Supported by

AWS AWS Cloud computing Datadog Datadog Monitoring DigiCert DigiCert EV certificate Facebook / Instagram Facebook / Instagram PSF Sponsor Fastly Fastly CDN Google Google Object Storage and Download Analytics Pingdom Pingdom Monitoring Salesforce Salesforce PSF Sponsor Sentry Sentry Error logging StatusPage StatusPage Status page