python中常用的公共函数封装

函数

计算函数的执行时间

 1 import time
 2  
 3 def timer(func):
 4     """
 5     用于计时的装饰器函数
 6  
 7     :param func: 被装饰函数
 8     :return: 闭包函数，封装了自定义行为与被装饰函数的调用
 9     """
10  
11     def wrapper(*args, **kwargs):
12         """
13         闭包函数
14  
15         :param args: 被装饰函数的位置参数
16         :param kwargs: 被装饰函数的关键字参数
17         :return: int,被装饰函数的计算结果
18         """
19         t1 = time.time()  # 开始时间
20  
21         r = func(*args, **kwargs)  # 被装饰函数调用
22  
23         t2 = time.time()  # 结束时间
24  
25         cost = t2 - t1  # 计算时长
26  
27         print('function cost{} second'.format(cost))
28  
29         return r
30  
31     return wrapper

装饰器

使用装饰器的方式进行参数校验

from functools import wraps
 
import json
 
from django.http import HttpResponse
 
 
def request_verify(request_method: str, need_params=None):
    """
        在views方法上加装饰器 例如：@request_verify('get', ['id'])
        :param request_method:
        :param need_params:
        :return:
    """
    def decorator(func):
 
        @wraps(func)
 
        def inner(a,request, *args, **kwargs):
 
            method = str(request.method).lower()
 
            # 先判断类型，判断请求方式是否符合，类型不符合，直接return
            if request_method and not method == request_method.lower():
 
                response = "method {0} not allowed for: {1}".format(request.method, request.path)
 
                return response_failure(response)
 
            request.params = {}
 
            # 暂时不用get请求传参，get的情况可以先忽略
            if method == 'get':
 
                if not request.GET:
 
                    if need_params:
 
                        response = "缺少参数"
 
                        return response_failure(response)
 
                else:
                    params = {}
 
                    request_params = request.GET
 
                    for item in request_params:
 
                        params.update({item: request_params.get(item)})
 
                    # get 必填参数校验
                    if need_params:
 
                        for need_param_name in need_params:
 
                            if not params.get(need_param_name):
 
                                response = "参数 '%s' 不能为空" % (need_param_name)
 
                                return response_failure(response)
 
                    request.params = params
 
            else:  # method == post
 
                if not request.body or request.body == {}:  # 参数为空的情况下
 
                    if need_params:  # 验证必传
 
                        response = "缺失参数"
 
                        return response_failure(response)
 
                else:  # 非空的时候，尝试去获取参数
 
                    try:
 
                        real_request_params = json.loads(request.body)  # 这边要try一下，如果前端传参不是json，json.loads会异常
 
                    except Exception as e:
 
                        response = "参数格式不合法,不是json类型"
                        return response_failure(response)
 
                    # 取出来以后再去判断下必填项是否每一项都有值
                    if need_params:
 
                        for need_param_name in need_params:
 
                            if not real_request_params.get(need_param_name):
 
                                # 如果必传参数取出来是'' (PS: 没传和传了''通过get取出来都是'')，就抛出
                                response = "参数 {0} 不能为空".format(need_param_name)
 
                                return response_failure(response)
 
                    # 一直到这里都无异常那么就可以把参数塞进request对象了，避免view里还要再去json.loads
                    request.params = real_request_params
 
                    print(request.params,'DDDD')
 
            return func(a,request, *args, **kwargs)
 
        return inner
 
    return decorator
 
 
# 校验结果封装,失败的封装结果
def response_failure(message):
 
    return HttpResponse(json.dumps({'code': 400,'message': message}, ensure_ascii=False), 'application/json')

中间件

使用中间件进行ip限流

# 导入时间模块
import time
 
from django.http import HttpResponse
 
from django.utils.deprecation import  MiddlewareMixin
 
 
# 限制用户访问次数，60秒内只能同一个ip只能访问5次
 
# 建立访问者ip池
interviewer_ip = {}
 
# 创建类并继承MiddlewareMixin，用于构建中间件
class interviewerMiddwaremixin(MiddlewareMixin):
 
    # 使用process_request可以接受request请求之后的函数
    def process_request(self,request):
 
        # 获取用户的访问ip
        ip = request.META.get('REMOTE_ADDR')
 
        # 获取访问时间
        visit_time = time.time()
 
        # 判断ip是否存在interviewer_ip中
        if ip not in interviewer_ip:
 
            # 维护字典,将新的ip地址加入字典
            interviewer_ip[ip] = [visit_time]
 
        else:
 
            # 已经存在，则将ip对应值的插入列表开始位置
            interviewer_ip[ip].insert(0, visit_time)
 
        # 获取ip_list列表
        ip_list = interviewer_ip[ip]
 
        # 计算访问时间差
        lead_time = ip_list[0] - ip_list[-1]
 
        print('地址:', ip, '访问次数:', len(ip_list), '时间差', lead_time)
 
        # 两个条件同时成立则，间隔时间在60s内
        while ip_list and lead_time > 60:
             
            # 默认移除列表中的最后一个元素
            ip_list.pop()
 
        # 判断访问的次数是否大于5次
        if len(ip_list) > 5:
            return HttpResponse("不好意思，您的访问过于频繁，请稍后重试")
 
        print('地址:', ip, '访问次数:', len(ip_list), '时间差', lead_time)