百度360必应搜狗淘宝本站头条
python判断字典是否为空crontab每周一执行aes和des区别安装指定版本npmadb pull
当前位置：网站首页 > IT知识 > 正文

Python开发必备:自定义JSON编码器完全指南

liuian 2025-06-12 14:11 4 浏览

在现代软件开发中，数据序列化是一个至关重要的技术环节，它负责将复杂的程序对象转换为可传输和存储的格式。JSON作为最广泛使用的数据交换格式，在Web服务、API接口和数据持久化中发挥着核心作用。然而，Python标准库中的JSON模块仅支持基本数据类型的序列化，面对复杂的自定义对象时往往力不从心。

基本原理与挑战

JSON序列化本质上是一个将内存中的对象表示转换为字符串格式的过程。Python的标准json模块基于递归下降的方式处理数据结构，它能够自动识别并序列化字典、列表、字符串、数字、布尔值和None等基本类型。这种机制的核心在于类型检测和格式转换，通过遍历对象的内部结构来生成对应的JSON表示。

当面对自定义类实例、日期时间对象、集合类型或其他复杂数据结构时，标准JSON模块会抛出TypeError异常。这是因为JSON规范本身只定义了有限的数据类型，无法直接表示Python中丰富的对象类型。解决这一挑战的关键在于建立对象到JSON表示的映射关系，将复杂对象的内部状态提取出来，转换为JSON支持的基本类型。

自定义编码器实现

实现自定义JSON编码器的核心方法是继承json.JSONEncoder类并重写其default方法。这个方法在遇到无法序列化的对象时被调用，可以提供自定义的序列化逻辑。

下面的实现展示了一个完整的自定义编码器，能够处理日期时间对象、集合类型、自定义类实例等多种复杂情况。

import json
import datetime
from decimal import Decimal
from dataclasses import dataclass

class CustomJSONEncoder(json.JSONEncoder):
    """
    自定义JSON编码器，支持多种复杂数据类型的序列化
    处理日期时间、集合、自定义对象等类型
    """
    
    def default(self, obj):
        # 处理日期时间对象
        if isinstance(obj, datetime.datetime):
            return {
                '__type__': 'datetime',
                'value': obj.isoformat()
            }
        
        if isinstance(obj, datetime.date):
            return {
                '__type__': 'date',
                'value': obj.isoformat()
            }
        
        # 处理集合类型
        if isinstance(obj, set):
            return {
                '__type__': 'set',
                'value': list(obj)
            }
        
        if isinstance(obj, tuple):
            return {
                '__type__': 'tuple',
                'value': list(obj)
            }
        
        # 处理Decimal类型
        if isinstance(obj, Decimal):
            return {
                '__type__': 'decimal',
                'value': str(obj)
            }
        
        # 处理自定义对象
        if hasattr(obj, '__dict__'):
            return {
                '__type__': 'custom_object',
                '__class__': obj.__class__.__name__,
                'attributes': obj.__dict__
            }
        
        # 处理dataclass对象
        if hasattr(obj, '__dataclass_fields__'):
            return {
                '__type__': 'dataclass',
                '__class__': obj.__class__.__name__,
                'fields': {field.name: getattr(obj, field.name) 
                          for field in obj.__dataclass_fields__.values()}
            }
        
        return super().default(obj)

# 定义测试类
@dataclass
class Person:
    name: str
    age: int
    email: str

class Product:
    def __init__(self, name, price, tags):
        self.name = name
        self.price = price
        self.tags = tags
        self.created_at = datetime.datetime.now()

# 创建测试数据
test_data = {
    'person': Person('张三', 30, 'zhangsan@example.com'),
    'product': Product('智能手机', Decimal('2999.99'), {'电子产品', '通讯设备'}),
    'timestamp': datetime.datetime.now(),
    'numbers': (1, 2, 3, 4, 5)
}

# 使用自定义编码器进行序列化
json_string = json.dumps(test_data, cls=CustomJSONEncoder, indent=2, ensure_ascii=False)
print("序列化结果:")
print(json_string)

运行结果：

序列化结果:
{
  "person": {
    "__type__": "custom_object",
    "__class__": "Person",
    "attributes": {
      "name": "张三",
      "age": 30,
      "email": "zhangsan@example.com"
    }
  },
  "product": {
    "__type__": "custom_object",
    "__class__": "Product",
    "attributes": {
      "name": "智能手机",
      "price": {
        "__type__": "decimal",
        "value": "2999.99"
      },
      "tags": {
        "__type__": "set",
        "value": [
          "电子产品",
          "通讯设备"
        ]
      },
      "created_at": {
        "__type__": "datetime",
        "value": "2025-06-08T12:59:16.355264"
      }
    }
  },
  "timestamp": {
    "__type__": "datetime",
    "value": "2025-06-08T12:59:16.355270"
  },
  "numbers": [
    1,
    2,
    3,
    4,
    5
  ]
}

高级编码器

为了构建更加强大的序列化系统，需要实现循环引用检测、深度限制和选择性序列化等高级功能。

下面的实现展示了一个功能完整的高级编码器，提供了生产环境所需的各种特性。

import datetime
import json


class AdvancedJSONEncoder(json.JSONEncoder):
    """
    高级JSON编码器，支持循环引用检测、深度限制等功能
    """

    def __init__(self, *args, **kwargs):
        self.max_depth = kwargs.pop('max_depth', 10)
        self.skip_private = kwargs.pop('skip_private', True)
        super().__init__(*args, **kwargs)
        self._obj_tracker = set()
        self._current_depth = 0

    def encode(self, obj):
        self._obj_tracker.clear()
        self._current_depth = 0
        return super().encode(obj)

    def default(self, obj):
        # 深度检查
        if self._current_depth > self.max_depth:
            return f"<深度超限>"

        # 循环引用检查
        obj_id = id(obj)
        if obj_id in self._obj_tracker:
            return f"<循环引用: {type(obj).__name__}>"

        self._obj_tracker.add(obj_id)
        self._current_depth += 1

        try:
            # 处理日期时间
            if isinstance(obj, datetime.datetime):
                return {'__type__': 'datetime', 'value': obj.isoformat()}

            # 处理自定义对象
            if hasattr(obj, '__dict__'):
                attributes = {}
                for key, value in obj.__dict__.items():
                    if self.skip_private and key.startswith('_'):
                        continue
                    if not callable(value):
                        attributes[key] = value
                return {
                    '__type__': 'custom_object',
                    '__class__': obj.__class__.__name__,
                    'attributes': attributes
                }

            return str(obj)

        finally:
            self._obj_tracker.discard(obj_id)
            self._current_depth -= 1


# 测试高级编码器
class Person:
    def __init__(self, name, age, email):
        self.name = name
        self.age = age
        self.email = email
        self._internal_id = "person_001"


class Department:
    def __init__(self, name):
        self.name = name
        self.employees = []
        self._internal_id = "dept_001"

    def add_employee(self, employee):
        self.employees.append(employee)


dept = Department("技术部")
person = Person("李四", 25, "lisi@example.com")
dept.add_employee(person)

encoder = AdvancedJSONEncoder(indent=2, ensure_ascii=False, max_depth=5, skip_private=True)
result = encoder.encode(dept)
print("高级编码器结果:")
print(result)

运行结果：

高级编码器结果:
{
  "__type__": "custom_object",
  "__class__": "Department",
  "attributes": {
    "name": "技术部",
    "employees": [
      {
        "__type__": "custom_object",
        "__class__": "Person",
        "attributes": {
          "name": "李四",
          "age": 25,
          "email": "lisi@example.com"
        }
      }
    ]
  }
}

反序列化机制实现

完整的序列化解决方案还需要支持从JSON到对象的反向转换。通过实现自定义的object_hook函数，可以在JSON解析过程中识别特殊的类型标记，并执行相应的对象重构逻辑。

import datetime
import json
from decimal import Decimal


class CustomJSONEncoder(json.JSONEncoder):
    """
    自定义JSON编码器，支持多种复杂数据类型的序列化
    处理日期时间、集合、自定义对象等类型
    """

    def default(self, obj):
        # 处理日期时间对象
        if isinstance(obj, datetime.datetime):
            return {
                '__type__': 'datetime',
                'value': obj.isoformat()
            }

        if isinstance(obj, datetime.date):
            return {
                '__type__': 'date',
                'value': obj.isoformat()
            }

        # 处理集合类型
        if isinstance(obj, set):
            return {
                '__type__': 'set',
                'value': list(obj)
            }

        if isinstance(obj, tuple):
            return {
                '__type__': 'tuple',
                'value': list(obj)
            }

        # 处理Decimal类型
        if isinstance(obj, Decimal):
            return {
                '__type__': 'decimal',
                'value': str(obj)
            }

        # 处理自定义对象
        if hasattr(obj, '__dict__'):
            return {
                '__type__': 'custom_object',
                '__class__': obj.__class__.__name__,
                'attributes': obj.__dict__
            }

        # 处理dataclass对象
        if hasattr(obj, '__dataclass_fields__'):
            return {
                '__type__': 'dataclass',
                '__class__': obj.__class__.__name__,
                'fields': {field.name: getattr(obj, field.name)
                           for field in obj.__dataclass_fields__.values()}
            }

        return super().default(obj)

class JSONDecoder:
    """
    自定义JSON解码器，支持对象反序列化
    """

    def __init__(self):
        self.type_handlers = {
            'datetime': self._decode_datetime,
            'date': self._decode_date,
            'set': self._decode_set,
            'tuple': self._decode_tuple,
            'decimal': self._decode_decimal
        }

    def decode(self, json_string):
        return json.loads(json_string, object_hook=self._object_hook)

    def _object_hook(self, obj):
        if '__type__' in obj:
            type_name = obj['__type__']
            if type_name in self.type_handlers:
                return self.type_handlers[type_name](obj)
        return obj

    def _decode_datetime(self, obj):
        return datetime.datetime.fromisoformat(obj['value'])

    def _decode_date(self, obj):
        return datetime.date.fromisoformat(obj['value'])

    def _decode_set(self, obj):
        return set(obj['value'])

    def _decode_tuple(self, obj):
        return tuple(obj['value'])

    def _decode_decimal(self, obj):
        return Decimal(obj['value'])


# 测试完整的序列化和反序列化
original_data = {
    'timestamp': datetime.datetime.now(),
    'price': Decimal('99.99'),
    'tags': {'python', 'json', 'serialization'},
    'coordinates': (10, 20, 30)
}

# 序列化
json_data = json.dumps(original_data, cls=CustomJSONEncoder)
print("序列化:", json_data)

# 反序列化
decoder = JSONDecoder()
restored_data = decoder.decode(json_data)
print("反序列化成功，时间类型:", type(restored_data['timestamp']))

运行结果：

序列化: {"timestamp": {"__type__": "datetime", "value": "2025-06-08T13:03:42.075846"}, "price": {"__type__": "decimal", "value": "99.99"}, "tags": {"__type__": "set", "value": ["json", "serialization", "python"]}, "coordinates": [10, 20, 30]}
反序列化成功，时间类型: <class 'datetime.datetime'>

总结

自定义JSON编码器为Python应用程序提供了强大的数据序列化能力。通过扩展标准库的功能，我们能够处理复杂的对象结构，实现完整的数据持久化和传输方案。在实际应用中，需要注意安全性考虑，建立白名单机制来限制可重建的类型。同时要考虑性能优化，避免过度复杂的序列化逻辑影响系统效率。合理使用自定义JSON编码器，能够显著提升系统的数据处理能力，为构建可扩展的现代应用奠定坚实基础。通过掌握这些技术，开发者可以更好地应对复杂的数据序列化需求，构建高质量的Python应用程序。

相关推荐

面试怕被问Hashmap,多看看这个文章

o数据结构otable数组长度永远为2的幂次方o那么为什么要把数组长度设计为2的幂次方呢？o扩容o链表树化o红黑树拆分o查找o插入o删除o遍历oequasl和hashcode总结HashMap是面试中...

非常简洁地重试Retry组件，使用起来杠杠的

前言小伙伴是不是经常遇到接口调用异常，超时的场景？尤其网络抖动导致timeout超时的场景，我们一般产品就会叫我们要重试几次。很多小伙伴的实现方式是写个循环调用for（inti=1;i<=3;...

Kafka消息可靠传输之幂等、事务机制

一般而言，消息中间件的消息传输保障有3个层级，分别如下。atmostonce：至多一次。消息可能会丢失，但绝对不会重复传输。atleastonce：最少一次。消息绝不会丢失，但可能会重复传输。...

Seata源码—9.Seata XA模式的事务处理

大纲1.SeataXA分布式事务案例及AT与XA的区别2.SeataXA分布式事务案例的各模块运行流程3.Seata使用SpringBoot自动装配简化复杂配置4.全局事务注解扫描组件的自动装配...

Disruptor—3.核心源码实现分析一

大纲1.Disruptor的生产者源码分析2.Disruptor的消费者源码分析3.Disruptor的WaitStrategy等待策略分析4.Disruptor的高性能原因5.Disruptor高性...

Spring Boot 进阶-详解SpringBoot中条件注解使用

作为使用SpringBoot框架的开发者来讲，如果你连如下的这些注解你都没有听说过，没有用过，那我劝你还是放弃吧？在SpringBoot中我们最常见到的注解应该是条件注解了吧！也就是@Condit...

如何自定义编解码器（如何自定义编解码器的程序）

1.前言上一节我们一节了解了什么是编码解码、序列化和反序列化了，并且留有一道思考题，本节内容主要是深入解析该思考题。思考题：能否把我们的编码和解码封装成独立的Handler呢？那么应该如何去封装...

Disruptor—3.核心源码实现分析二

大纲1.Disruptor的生产者源码分析2.Disruptor的消费者源码分析3.Disruptor的WaitStrategy等待策略分析4.Disruptor的高性能原因5.Disruptor高性...

线程的状态有哪些?它是如何工作的?

线程的状态有哪些？它是如何工作的？线程（Thread）是并发编程的基础，也是程序执行的最小单元，它依托进程而存在。一个进程中可以包含多个线程，多线程可以共享一块内存空间和一组系统资源，因此线程之间的切...

有图解有案例，我终于把Condition的原理讲透彻了

平时加解锁都是直接使用Synchronized关键字来实现的，简单好用，为啥还要引用ReentrantLock呢？为了解决小伙伴的疑问，我们来对两者做个简单的比较吧：相同点两者都是“可重入锁”，即当前...

白话DUBBO原理，通俗易记，再也不怕面试时讲不清楚了

现在的各种面试免不了要问些中间件，尤其是互联网公司，更注重获选人对中间件的掌握情况。在中间件中，有一大类是关于RPC框架的，Dubbo即是阿里出品的一款很著名的RPC中间件，很多互联网公司都在用，面试...

Java 最细的集合类总结（java常用的集合类有哪些）

数据结构作为每一个开发者不可回避的问题，而Java对于不同的数据结构提供了非常成熟的实现，这一个又一个实现既是面试中的难点，也是工作中必不可少的工具，在此，笔者经历漫长的剖析，将其抽丝剥茧的呈现出...

详解Java异常(Exception)处理及常见异常

很多事件并非总是按照人们自己设计意愿顺利发展的，经常出现这样那样的异常情况。例如：你计划周末郊游，计划从家里出发→到达目的→游泳→烧烤→回家。但天有不测风云，当你准备烧烤时候突然天降大雨，只能终止郊...

为什么阿里强制要求不要在foreach循环里进行元素remove和add操作

在阅读《阿里巴巴Java开发手册》时，发现有一条关于在foreach循环里进行元素的remove/add操作的规约，具体内容如下：错误演示我们首先在IDEA中编写一个在foreach循...

SpringBoot条件化配置(@Conditional)全面解析与实战指南

一、条件化配置基础概念1.1什么是条件化配置条件化配置是Spring框架提供的一种基于特定条件来决定是否注册Bean或加载配置的机制。在SpringBoot中，这一机制通过@Conditional...

一周热门
最近发表
标签列表