DRF-Serializers序列化器组件源码分析及改编

1. 源码分析

注意：以下代码片段为方便明确已举行简化，只保留了与序列化功能相关的代码
序列化的源码中涉及到了元类的概念，我在这里简单阐明一下：元类（metaclass）是一个高级概念，用于定义类的创建行为。简单来说，元类是创建类的类，它决定了类的创建方式和行为。
在 Python 中一切皆为对象，包括类。每个类都有一个元类，它定义了如何创建这个类。通常情况下 Python 会使用默认的元类 type 来创建类。但是，当我们需要对类的创建过程举行自定义时，就可以使用元类，举例：

1. class Mytype(type)
2.         def __new__(cls,name,bases,attrs):   # 类名，继承的父类 ，成员
3.         # 此处可对要创建的类进行操作
4.         del attrs["v1"]
5.         attrs["name"] = "harry"
6.         
7.         xx = super().__new__(cls,name,bases,attrs)  # 调用type类创建对象（这个对象就是Bar类）
8.         retyrn xx
10.         
11. class Bar(object, metaclass=Mytype)  # metaclass指定自定义元类
12.         v1 = 123
13.    
14.     def func(self):
15.         pass
16.    
17. 由于元类中删除了v1属性，且增加了name属性，因此此时Bar中无v1属性，且多了name属性

复制代码

另：父类如果指定了元类metaclass，则其子类都默认是用该元类来创建类
补充：实例化Bar类时，相当于是 type对象（），因此会触发type类的__call__方法，其中就调用了Bar的__new__和__init__，因此在实例化类时才会自动触发类的__new__和__init__方法。本质上是由于 对象（） 而调用了type元类的call方法；



Serializers组件主要有两个功能：序列化和数据校验

• 序列化部分：
  起首定义一个序列化类
  

1. class DepartSerializer(serializers.Serializer):
2.     '''Serializer校验'''
3.     # 内置校验
4.     title = serializers.CharField(required=True, max_length=20, min_length=6)
5.     order = serializers.IntegerField(required=False, max_value=100, min_value=10)
6.     count = serializers.ChoiceField(choices=[(1, "高级"), (2, "中级")])

复制代码

查看Serializer的父类，可知其是通过SerializerMetaclass元类创建的

1. Serializer(BaseSerializer, metaclass=SerializerMetaclass)

复制代码

SerializerMetaclass元类源码：

1. class SerializerMetaclass(type):
2.     @classmethod
3.     def _get_declared_fields(cls, bases, attrs):
4.         fields = [(field_name, attrs.pop(field_name))  # 通过循环获取field字段对象
5.                   for field_name, obj in list(attrs.items())
6.                   if isinstance(obj, Field)]
7.         fields.sort(key=lambda x: x[1]._creation_counter)
9.         known = set(attrs)
10.         def visit(name):
11.             known.add(name)
12.             return name
14.         base_fields = [
15.             (visit(name), f)
16.             for base in bases if hasattr(base, '_declared_fields')
17.             for name, f in base._declared_fields.items() if name not in known
18.         ]
20.         return OrderedDict(base_fields + fields)
22.     def __new__(cls, name, bases, attrs):
23.         attrs['_declared_fields'] = cls._get_declared_fields(bases, attrs)    # 为类中增加了_declared_fields属性，其中封装了所有的Field字段名及对应的对象
24.         return super().__new__(cls, name, bases, attrs)

复制代码

通过serializer.data触发序列化流程：

1.     @property
2.     def data(self):
3.         ret = super().data   # 寻找其父类BaseSerializer的data方法
4.         return ReturnDict(ret, serializer=self)

复制代码

BaseSerializer的data方法源码：

1.     @property
2.     def data(self):
3.         if hasattr(self, 'initial_data') and not hasattr(self, '_validated_data'):
4.             msg = (
5.                 'When a serializer is passed a `data` keyword argument you '
6.                 'must call `.is_valid()` before attempting to access the '
7.                 'serialized `.data` representation.\n'
8.                 'You should either call `.is_valid()` first, '
9.                 'or access `.initial_data` instead.'
10.             )
11.             raise AssertionError(msg)
13.         if not hasattr(self, '_data'):
14.             if self.instance is not None and not getattr(self, '_errors', None):
15.                 self._data = self.to_representation(self.instance)    # 执行to_representation方法获取序列化数据
16.             elif hasattr(self, '_validated_data') and not getattr(self, '_errors', None):
17.                 self._data = self.to_representation(self.validated_data)
18.             else:
19.                 self._data = self.get_initial()
20.         return self._data

复制代码

to_representation方法源码（核心）：

1.     def to_representation(self, instance):
2.         ret = OrderedDict()
3.         fields = self._readable_fields  # 筛选出可读的字段对象（其内部对_declared_fields字段进行了深拷贝）
5.         for field in fields:
6.             try:
7.                 attribute = field.get_attribute(instance)  # 循环字段对象列表，并执行get_attribute方法获取对应的值
8.             except SkipField:
9.                 continue
10.             check_for_none = attribute.pk if isinstance(attribute, PKOnlyObject) else attribute
11.             if check_for_none is None:
12.                 ret[field.field_name] = None
13.             else:
14.                 ret[field.field_name] = field.to_representation(attribute)  # 执行to_representation转换格式，并将所有数据封装到ret字典中
16.         return ret

复制代码

get_attribute方法源码：

1. def get_attribute(self, instance):
2.     return get_attribute(instance, self.source_attrs)

复制代码

1. def get_attribute(instance, attrs): # attrs为source字段值  instance为模型对象
2.     for attr in attrs:
3.         try:
4.             if isinstance(instance, Mapping):
5.                 instance = instance[attr]
6.             else:
7.                 instance = getattr(instance, attr)  # 循环获取模型对象最终的attr的值
8.         except ObjectDoesNotExist:
9.             return None
10.     return instance  # 返回该字段值

复制代码

2. 数据校验部分
使用is_valid方法校验数据，获取_errors数据，_errors存在则is_valid返回False。在执行该函数的过程中，触发了run_validation方法：

1.     def is_valid(self, raise_exception=False):
2.         if not hasattr(self, '_validated_data'):
3.             try: # 触发了run_validation方法
4.                 self._validated_data = self.run_validation(self.initial_data)
5.             except ValidationError as exc:
6.                 self._validated_data = {}
7.                 self._errors = exc.detail
8.             else:
9.                 self._errors = {}
11.         if self._errors and raise_exception:
12.             raise ValidationError(self.errors)
14.         return not bool(self._errors)****

复制代码

run_validation方法，注意该方法是Serializer类下的方法，不是Field类的方法。在to_internal_value方法中调用字段内置校验，并执行钩子函数。

1.     def run_validation(self, data=empty):
3.         (is_empty_value, data) = self.validate_empty_values(data)
4.         if is_empty_value:
5.             return data
7.         value = self.to_internal_value(data)  # 调用字段内置校验，并执行钩子函数
8.         try:
9.             self.run_validators(value)
10.             value = self.validate(value)
11.             assert value is not None, '.validate() should return the validated data'
12.         except (ValidationError, DjangoValidationError) as exc:
13.             raise ValidationError(detail=as_serializer_error(exc))
15.         return value

复制代码

to_internal_value方法，fileds从_declared_fields中深拷贝而得到，且只包罗了只写的字段对象

1.     def to_internal_value(self, data):
2.         if not isinstance(data, Mapping):
3.             message = self.error_messages['invalid'].format(
4.                 datatype=type(data).__name__
5.             )
6.             raise ValidationError({
7.                 api_settings.NON_FIELD_ERRORS_KEY: [message]
8.             }, code='invalid')
10.         ret = OrderedDict()
11.         errors = OrderedDict()
12.         fields = self._writable_fields  # 筛选只写的字段对象
14.         for field in fields:
15.             validate_method = getattr(self, 'validate_' + field.field_name, None)
16.             primitive_value = field.get_value(data)
17.             try:
18.                 validated_value = field.run_validation(primitive_value) # 执行内置校验
19.                 if validate_method is not None:
20.                     validated_value = validate_method(validated_value)  # 执行钩子函数进行校验
21.             except ValidationError as exc:
22.                 errors[field.field_name] = exc.detail
23.             except DjangoValidationError as exc:
24.                 errors[field.field_name] = get_error_detail(exc)
25.             except SkipField:
26.                 pass
27.             else:
28.                 set_value(ret, field.source_attrs, validated_value)
29.         if errors:
30.             raise ValidationError(errors)
31.         return ret

复制代码

run_validation内置校验：

1.     def run_validation(self, data=empty):
2.         if data == '' or (self.trim_whitespace and str(data).strip() == ''):
3.             if not self.allow_blank:
4.                 self.fail('blank')
5.             return ''
6.         return super().run_validation(data)
8.     # 父类的run_validation方法
9.     def run_validation(self, data=empty):
11.         (is_empty_value, data) = self.validate_empty_values(data)
12.         if is_empty_value:
13.             return data
14.         value = self.to_internal_value(data)
15.         self.run_validators(value)  # 调用字段定义的run_validators进行校验
16.         return value

复制代码

2、源码改编：

• 自定义钩子：让某字段既能支持前端传入，又能自定义序列化返回的值；（SerializerMethodField默认是只可读的，用户无法输入，而普通field又无法自定义复杂逻辑返回值）
  

思绪：在调用ser.data开始序列化后的to_representation方法中判断有无自定义格式的钩子，如果有则替换掉该字段对象的值

1.     def to_representation(self, instance):
2.         ret = OrderedDict()
3.         fields = self._readable_fields
5.         for field in fields:
6.             if hasattr(self, 'get_%s' % field.field_name):  # 判断是否有"get_xxx"形式的函数，如则执行该方法并将instance传入
7.                 value = getattr(self, 'get_%s' % field.field_name)(instance)
8.                 ret[field.field_name] = value
9.             else:
10.                 try:
11.                     attribute = field.get_attribute(instance)
12.                 except SkipField:
13.                     continue
15.                 check_for_none = attribute.pk if isinstance(attribute, PKOnlyObject) else attribute
16.                 if check_for_none is None:
17.                     ret[field.field_name] = None
18.                 else:
19.                     ret[field.field_name] = field.to_representation(attribute)
21.         return ret

复制代码

如果其他类中也需要使用该重写方法，可将该重新方法封装成类，其他类中继续该类后，即可不消每次都重写to_representation方法

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。