Django

Django 自定义中间件及视图

Django | View | Middleware | 视图 | 中间件 | 自定义 | 定制 | 快速开发

Posted by luoruiqing on December 6, 2020

在进行接口开发时, 对请求和响应都需要做一些统一 数据结构 的约定, 状态码 约定等等, 这种约定大多要根据自己的业务情况来进行. 这里提供封装思路及模板, 简单实现的功能如下:

  • 自定义错误
    •   错误码参数
    •   错误码唯一性校验
    •   HTTP 响应状态码存储
    •   支持字符串格式化
  • 中间件
    •   JSON 请求的序列化
    •   统一的 JSON 序列化失败提示
    •   固定响应数据的格式
    •   已知错误的捕获(包含断言抛出的)
    •   支持 Django QuerySet / Model 的 JSON 序列化
    •   支持自定义序列化器(中间件定制)
  • 扩展视图(与中间件功能相同)

开始

要满足以上几个功能点, 大体上需要几个基础模块

错误定制

Error.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

class BaseError(BaseException):
    CODES = {}  # 所有错误


class Error(BaseError):

    def __new__(cls, code, *args, **kwargs):
        check = kwargs.pop('check', True)
        instance = super().__new__(cls, code, *args, **kwargs)
        if check:
            assert code not in instance.CODES, f"Error code duplication: {code}."
            instance.CODES[code] = instance  # 记录错误实例
        return instance

    def __init__(self, code, message, status_code=400, *args, **kwargs):
        ''' 带参数进入格式化到消息体 '''
        self.code = code
        self.message = message
        self.status_code = status_code

    def __call__(self, *args, **kwargs):
        ''' 回调始终生成新的错误对象, 且code不允许变化 '''
        message = (self.message.format(*args, **kwargs)) if args or kwargs else self.message

        return self.__class__(
            code=self.code,
            message=message,
            status_code=self.status_code,
            check=False,
        )

    def __str__(self):
        return f'Error {self.code}({self.status_code}): {self.message}.'


REQUEST_JSON_ERROR = Error(10000, 'JSON请求不合法')

错误码唯一性校验

1
2

REQUEST_JSON_ERROR = Error(10000, 'JSON请求不合法')
REQUEST_JSON_ERROR = Error(10000, 'JSON请求不合法1')

此时在项目启动时即可报错

支持字符串格式化

1
2
3
4
5

REQUEST_JSON_ERROR = Error(1000, '路径 {} 下 {file} 文件不存在.')

raise REQUEST_JSON_ERROR('~/user1/text', file='a.text')
# or
assert False, REQUEST_JSON_ERROR('~/user1/text', file='a.text')

结果:

XXXX.Error: Error 1000(400): 路径 ~/user1/text 下 a.text 文件不存在..
AssertionError: Error 1000(400): 路径 ~/user1/text 下 a.text 文件不存在..

序列化器

支持范围

  •   Python基础变量类型
  •   Django QuerySet
  •   Django Model
  •   Iterable(生成器/迭代器/元祖等)
  •   float NaN 的兼容

encode.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

import math
from json import JSONEncoder
from collections import Iterable
from django.db.models import Model, QuerySet
from django.forms.models import model_to_dict as django_mtd


def model_to_dict(instance, fields=None, exclude=None):
    ''' 模型转字典 '''
    result = {}  # 原类型
    if isinstance(instance, dict):  # 字典类型
        for key, value in instance.items():
            if fields and key not in fields:  # 保留指定字段
                del instance[key]  # 清除
                continue
            if exclude and key in exclude:  # 排除指定字段
                del instance[key]
                continue
            result[key] = model_to_dict(value, fields=fields, exclude=exclude)  # 回调更新
    elif isinstance(instance, (list, Iterable, QuerySet)) and not isinstance(instance, str):  # 除字符类型的任何可迭代对象包含生成器
        result = [model_to_dict(row, fields=fields, exclude=exclude) for row in instance]  # 回调更新
    elif isinstance(instance, Model):
        result = django_mtd(instance, fields=fields, exclude=exclude)
    elif isinstance(result, float) and math.isnan(result):  # 增加nan的兼容逻辑
        result = ''
    return result or instance


mtd = model_to_dict


class DjangoJSONEncoder(JSONEncoder):

    def default(self, o):
        return model_to_dict(o) or super().default(o)

流程定制

因为 Django 中间件流程相对请求流程是足够合理的, 分为 处理请求前/处理请求/处理请求后, 所以这里模仿 Django 中间件的写法进行流程定制.

handler.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

import json
import logging
from django.http.response import HttpResponseBase, HttpResponse
from . import error


class APIHandler:
    ''' 自定义的处理模式, 用于请求前JSON校验, 响应后格式统一, 已知错误后格式统一 '''
    DEFAULT_RESPONSE_JSON = {"code": 200, 'message': "success"}

    JSONEncoder = json.JSONEncoder  # json转换器

    def process_request(self, request):
        request_json = 'application/json' in request.META.get('HTTP_ACCEPT', '')
        try:  # 增加JSON
            request.json = json.loads(request.body) if request_json and request.body else {}
        except Exception as e:  # 提交json错误
            logging.error(f'{error.REQUEST_JSON_ERROR} -> {request.body}')
            raise error.REQUEST_JSON_ERROR

    def process_response(self, request, response):
        if isinstance(response, HttpResponseBase):  # 正常响应对象
            return response

        data = json.dumps({**self.DEFAULT_RESPONSE_JSON.copy(), **{'data': response}}, cls=self.JSONEncoder)
        return HttpResponse(data, content_type='application/json')

    def process_exception(self, request, e):
        e = (isinstance(e, AssertionError) and isinstance(e.args[0], error.BaseError) and e.args[0]) or e
        # 处理已知错误
        if isinstance(e, error.BaseError):
            return HttpResponse(json.dumps({"code": e.code, 'message': e.message}), status=e.status_code, content_type='application/json')

中间件

middlewares.py (增加上面的序列化器)

1
2
3
4
5
6
7
8
9
10
11
12
13

from django.utils.deprecation import MiddlewareMixin

from .handler import APIHandler
from .encode import DjangoJSONEncoder


class APIMiddlewareMixinBase(MiddlewareMixin, APIHandler):
    pass


class APIMiddleware(APIMiddlewareMixinBase):
    JSONEncoder = DjangoJSONEncoder

视图级别

如果中间件影响的范围过大, 可以考虑增加视图级别的流程处理, 可以通过继承 django.views.generic 实现, 这里考虑到复用情况, 使用 Python 类的内置方法 __init_subclass__ 来实现子类方法的自动装饰.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

import functools
from django.views.generic import View

from .handler import APIHandler
from .encode import DjangoJSONEncoder


class ViewBase(View):
    '''  自定义视图基类 '''

    @staticmethod
    def wrapper_process(func):
        ''' 这是装饰器方法, 用于实现响应 '''
        raise NotImplementedError()

    def __init_subclass__(cls, **kwargs):
        ''' 利用 __init_subclass__ 方法, 给请求方法进行装饰 '''
        if next(filter(lambda c: issubclass(c, ViewBase), cls.__mro__), None):  # 正确的继承
            for method in cls.http_method_names:  # 针对Django View的装饰
                if hasattr(cls, method):
                    setattr(cls, method, cls.wrapper_process(getattr(cls, method)))
        return super(cls).__init_subclass__(**kwargs)


class APIViewBase(ViewBase):
    def wrapper_process(func):
        @functools.wraps(func)
        def wrapper(self, request, *args, **kwargs):
            handler = APIHandler()
            handler.JSONEncoder = getattr(self, 'JSONEncoder', handler.JSONEncoder)
            try:
                response = handler.process_request(request)
                if not response:
                    response = func(self, request, *args, **kwargs)
                response = handler.process_response(request, response)
            except Exception as e:
                response = handler.process_exception(request, e)
                if not response:
                    raise e
            return response
        return wrapper


class APIView(APIViewBase):
    JSONEncoder = DjangoJSONEncoder  # 切换序列化器

使用

中间件形式

注册

1
2
3
4

MIDDLEWARE = [
    # ...
    'utils.middlewares.APIMiddleware',
]

方法视图

1
2
3
4
5
6
7
8
9

from django.contrib.contenttypes.models import ContentType
from utils import error

def test_view(request):
    assert False, error.FILE_NOT_FOUND_ERROR('错误 1')
    return {
        'ContentType': ContentType.objects.all(),
        'data1': {'test': 1},
    }

类视图

1
2
3
4
5
6

from .utils.views import APIView
from django.contrib.contenttypes.models import ContentType

class TestView(APIView):
    def get(self, request):
        return ContentType.objects.all()
CATALOG
    FEATURED TAGS
    CSS Vue Django JavaScript Python Tools Docker Mysql VSCode Link Linux 爬虫 Chrome Webpack
    FRIENDS