6Python Cookbook(第3版)

数据编码与处理

数据处理

struct二进制数据XMLJSON编解码

🎯 AI 时代的学习建议:不要只读理论。理解这一章的设计动机(为什么需要这个概念?)和边界条件(什么情况下会失效?)。当 AI 生成的代码涉及这些概念时,你能判断它是否用对了场景。AI 能帮你总结内容,但无法替代你亲手阅读、思考、建立直觉的过程。

第6-12章:数据编码、函数、类与对象、元编程、模块、网络、并发

一、导读

本章合并了 Python Cookbook 剩余章节的核心内容,涵盖数据编码与处理、函数编程、类与对象设计、元编程技巧、模块与包组织、网络与 Web 编程、并发编程。

二、核心概念详解

2.1 数据编码(struct、二进制)

python
import struct

# struct: 二进制数据打包
data = struct.pack('>ihb', 1, 2, 3)  # big-endian: int, int, byte
print(len(data))  # 7 bytes

values = struct.unpack('>ihb', data)
print(values)  # (1, 2, 3)

# 读取二进制文件
with open('data.bin', 'rb') as f:
    header = f.read(32)
    magic, version, count = struct.unpack('<4sII', header[:12])

2.2 函数编程

python
from functools import partial, lru_cache

# 闭包
def make_multiplier(factor):
    def multiplier(x):
        return x * factor
    return multiplier

double = make_multiplier(2)
print(double(5))  # 10

# partial: 固定部分参数
def power(base, exponent):
    return base ** exponent

square = partial(power, exponent=2)
cube = partial(power, exponent=3)
print(square(5))  # 25

# 装饰器
def memoize(func):
    cache = {}
    def wrapper(*args):
        if args not in cache:
            cache[args] = func(*args)
        return cache[args]
    return wrapper

# lru_cache: 内置缓存装饰器
@lru_cache(maxsize=128)
def fibonacci(n):
    if n < 2: return n
    return fibonacci(n-1) + fibonacci(n-2)

2.3 类与对象

python
class LazyProperty:
    """延迟计算的属性描述符"""
    def __init__(self, func):
        self.func = func
        self.name = func.__name__
    
    def __get__(self, instance, owner):
        if instance is None:
            return self
        value = self.func(instance)
        instance.__dict__[self.name] = value
        return value

class DataProcessor:
    def __init__(self, data):
        self.data = data
    
    @LazyProperty
    def statistics(self):
        print("Computing statistics...")
        return {
            'mean': sum(self.data) / len(self.data),
            'min': min(self.data),
            'max': max(self.data),
        }

dp = DataProcessor([1, 2, 3, 4, 5])
print(dp.statistics)  # 第一次: 计算
print(dp.statistics)  # 第二次: 缓存

2.4 模块与包

mypackage/
    __init__.py
    module1.py
    module2.py
    subpackage/
        __init__.py
        module3.py
python
# __init__.py: 控制包的公共接口
from .module1 import func1
from .module2 import func2
__all__ = ['func1', 'func2']

# 延迟导入
def heavy_function():
    import expensive_module  # 只在调用时导入
    return expensive_module.compute()

2.5 网络编程

python
import socket
import urllib.request

# TCP 客户端
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
    s.connect(('example.com', 80))
    s.sendall(b'GET / HTTP/1.0\r\nHost: example.com\r\n\r\n')
    response = s.recv(4096)

# HTTP 请求
with urllib.request.urlopen('https://httpbin.org/get') as response:
    data = response.read()

# 简单 HTTP 服务器
from http.server import HTTPServer, SimpleHTTPRequestHandler
server = HTTPServer(('localhost', 8080), SimpleHTTPRequestHandler)
server.serve_forever()

2.6 并发编程

python
import threading
import multiprocessing
import asyncio

# 线程
def worker(n):
    print(f"Thread {n}: working")

threads = [threading.Thread(target=worker, args=(i,)) for i in range(5)]
for t in threads: t.start()
for t in threads: t.join()

# 进程池
with multiprocessing.Pool(4) as pool:
    results = pool.map(lambda x: x*x, range(10))

# asyncio
async def fetch(url):
    await asyncio.sleep(1)
    return f"Response from {url}"

async def main():
    tasks = [fetch(f"url{i}") for i in range(5)]
    results = await asyncio.gather(*tasks)

asyncio.run(main())

三、本章小结

struct 处理二进制数据

闭包和装饰器是函数编程的核心

描述符控制属性访问

包结构组织大型项目

socket 实现网络通信

threading/multiprocessing/asyncio 提供多种并发模型


四、思考与练习

何时选择 threading vs multiprocessing vs asyncio?

如何设计一个线程安全的缓存?


6/12数据编码
0%