标签 Python 下的文章

项目里到处都是 dict,读取数据时 data['name'],写错了 key 又不报错。上线后排查了半天,发现是 key 拼写错了。后来用 dataclass 替换 dict,IDE 能正确补全,写错属性名直接报错。

为什么不用 dict

# 典型的 dict 用法
user = {
    "name": "张三",
    "age": 25,
    "email": "zhangsan@example.com"
}

# 读取数据
name = user["name"]  # 正常
name = user["namee"]  # KeyError!运行时才发现

# 传参时
def send_email(user):
    print(f"发送邮件到 {user['emial']}")  # 拼写错误,运行时才发现

问题:

  1. 没有类型提示:IDE 不知道 user 有哪些字段
  2. 拼写错误不报错user['emial'] 运行时才报 KeyError
  3. 没有默认值:每个字段都要手动赋值
  4. 没有比较方法:两个 dict 比较需要手动实现

dataclass 基础用法

最简配置

from dataclasses import dataclass

@dataclass
class User:
    name: str
    age: int
    email: str

user = User(name="张三", age=25, email="zhangsan@example.com")
print(user)  # User(name='张三', age=25, email='zhangsan@example.com')

带默认值

@dataclass
class User:
    name: str
    age: int = 0
    email: str = ""
    is_active: bool = True

user = User(name="张三")
print(user)  # User(name='张三', age=0, email='', is_active=True)

坑在于:默认值必须放在没有默认值的字段后面。

从 dict 创建

data = {"name": "张三", "age": 25, "email": "zhangsan@example.com"}
user = User(**data)
print(user.name)  # 张三

进阶用法

字段选项

from dataclasses import dataclass, field

@dataclass
class User:
    name: str
    age: int = field(default=0, metadata={"min": 0, "max": 150})
    email: str = field(default="", repr=False)  # repr=False 不在打印时显示
    tags: list = field(default_factory=list)  # 可变默认值要用 default_factory

user = User(name="张三")
print(user)  # User(name='张三', age=0, tags=[])

坑在于:可变对象(list、dict)不能直接作为默认值,要用 default_factory

frozen dataclass

@dataclass(frozen=True)
class Point:
    x: float
    y: float

p1 = Point(1.0, 2.0)
# p1.x = 3.0  # 报错:FrozenInstanceError

frozen dataclass 不可变,可以用作 dict 的 key。

继承

@dataclass
class Animal:
    name: str
    age: int

@dataclass
class Dog(Animal):
    breed: str = "unknown"

dog = Dog(name="旺财", age=3, breed="柴犬")
print(dog)  # Dog(name='旺财', age=3, breed='柴犬')

与 Pydantic 的对比

from dataclasses import dataclass
from pydantic import BaseModel

# dataclass
@dataclass
class UserDC:
    name: str
    age: int

# Pydantic
class UserPydantic(BaseModel):
    name: str
    age: int

# dataclass 不做类型验证
user_dc = UserDC(name="张三", age="二十五")  # 不报错

# Pydantic 做类型验证
user_pydantic = UserPydantic(name="张三", age="二十五")  # 报错

选择建议:

  • dataclass:标准库,轻量,适合内部数据传递
  • Pydantic:第三方库,有类型验证,适合 API 参数校验

实战:配置管理

from dataclasses import dataclass, field
from typing import List

@dataclass
class DatabaseConfig:
    host: str = "localhost"
    port: int = 5432
    user: str = "postgres"
    password: str = ""
    db_name: str = "app"

@dataclass
class AppConfig:
    debug: bool = False
    database: DatabaseConfig = field(default_factory=DatabaseConfig)
    allowed_hosts: List[str] = field(default_factory=list)

config = AppConfig(debug=True, database=DatabaseConfig(host="db.example.com"))
print(config.database.host)  # db.example.com

实战:API 响应模型

from dataclasses import dataclass, asdict
from typing import Optional

@dataclass
class ApiResponse:
    code: int
    message: str
    data: Optional[dict] = None

def success(data=None):
    return ApiResponse(code=200, message="success", data=data)

def error(message):
    return ApiResponse(code=500, message=message)

# 转换为 dict
response = success({"user": "张三"})
print(asdict(response))  # {'code': 200, 'message': 'success', 'data': {'user': '张三'}}

踩坑总结

  1. 可变默认值(list、dict)要用 field(default_factory=...)
  2. frozen dataclass 不可变,可以用作 dict 的 key
  3. dataclass 不做类型验证,需要验证用 Pydantic
  4. asdict() 可以把 dataclass 转换为 dict
  5. dataclass 是标准库,Python 3.7+ 可用

从 dict 到 dataclass,只需要加一个 @dataclass 装饰器,但能获得类型提示、IDE 补全、自动 __repr____eq__。坑在于:很多人不知道 default_factory 这个坑。

项目里到处都是重复的日志代码,每个函数开头写 logger.info("开始..."),结尾写 logger.info("结束")。后来用装饰器统一处理,代码简洁了很多。这篇文章记录一下装饰器的实战用法。

什么是闭包

装饰器的基础是闭包。简单说,闭包就是函数里定义的函数,能记住外层函数的变量。

def outer(x):
    def inner(y):
        return x + y  # inner 记住了 x 的值
    return inner

add5 = outer(5)
print(add5(3))  # 输出 8
print(add5(10))  # 输出 15

坑在于:outer(5) 执行完后,按理说 x 应该被销毁了。但 inner 函数记住了 x 的值,所以还能用。

最简单的装饰器

def log_decorator(func):
    def wrapper(*args, **kwargs):
        print(f"调用 {func.__name__}")
        result = func(*args, **kwargs)
        print(f"{func.__name__} 执行完成")
        return result
    return wrapper

@log_decorator
def add(a, b):
    return a + b

add(1, 2)
# 输出:
# 调用 add
# add 执行完成

@log_decorator 等价于 add = log_decorator(add)

带参数的装饰器

如果装饰器本身需要参数,要再包一层:

import time
from functools import wraps

def timer(threshold=None):
    def decorator(func):
        @wraps(func)  # 保留原函数的 __name__ 和 __doc__
        def wrapper(*args, **kwargs):
            start = time.time()
            result = func(*args, **kwargs)
            elapsed = time.time() - start
            if threshold and elapsed > threshold:
                print(f"警告:{func.__name__} 耗时 {elapsed:.2f}s,超过阈值 {threshold}s")
            else:
                print(f"{func.__name__} 耗时 {elapsed:.2f}s")
            return result
        return wrapper
    return decorator

@timer(threshold=1)
def slow_function():
    time.sleep(2)

slow_function()
# 输出:警告:slow_function 耗时 2.00s,超过阈值 1s

坑在于:不加 @wraps(func),装饰后的函数 __name__ 会变成 wrapper,调试时很困惑。

实战:日志装饰器

import logging
import time
from functools import wraps

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

def log_execution(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        logger.info(f"开始执行 {func.__name__},参数: args={args}, kwargs={kwargs}")
        start = time.time()
        try:
            result = func(*args, **kwargs)
            elapsed = time.time() - start
            logger.info(f"{func.__name__} 执行成功,耗时: {elapsed:.3f}s")
            return result
        except Exception as e:
            elapsed = time.time() - start
            logger.error(f"{func.__name__} 执行失败,耗时: {elapsed:.3f}s,错误: {e}")
            raise
    return wrapper

@log_execution
def process_data(data):
    time.sleep(0.5)
    return [x * 2 for x in data]

result = process_data([1, 2, 3])

输出:

2026-07-20 14:30:00 [INFO] 开始执行 process_data,参数: args=([1, 2, 3],), kwargs={}
2026-07-20 14:30:01 [INFO] process_data 执行成功,耗时: 0.501s

实战:重试装饰器

import time
from functools import wraps

def retry(max_retries=3, delay=1):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise
                    print(f"第 {attempt + 1} 次失败:{e},{delay}s 后重试...")
                    time.sleep(delay)
        return wrapper
    return decorator

@retry(max_retries=3, delay=2)
def unstable_api():
    import random
    if random.random() < 0.7:
        raise ConnectionError("API 连接失败")
    return "成功"

result = unstable_api()

这个装饰器在网络请求场景很有用,自动重试失败的请求。

实战:缓存装饰器

from functools import wraps

def cache(func):
    cached = {}
    @wraps(func)
    def wrapper(*args):
        if args in cached:
            print(f"命中缓存:{args}")
            return cached[args]
        result = func(*args)
        cached[args] = result
        return result
    return wrapper

@cache
def fibonacci(n):
    if n < 2:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)

print(fibonacci(10))  # 计算
print(fibonacci(10))  # 命中缓存

坑在于:这个简单缓存没有过期机制,如果数据量大会占用内存。Python 3.9+ 可以用 functools.lru_cache

实战:权限检查装饰器

from functools import wraps

def require_permission(permission):
    def decorator(func):
        @wraps(func)
        def wrapper(user, *args, **kwargs):
            if permission not in user.get("permissions", []):
                raise PermissionError(f"用户 {user['name']} 没有 {permission} 权限")
            return func(user, *args, **kwargs)
        return wrapper
    return decorator

@require_permission("admin")
def delete_user(user, user_id):
    print(f"删除用户 {user_id}")

admin = {"name": "admin", "permissions": ["admin", "read"]}
user = {"name": "guest", "permissions": ["read"]}

delete_user(admin, 123)  # 正常执行
# delete_user(user, 123)  # 抛出 PermissionError

踩坑总结

  1. 装饰器本质是闭包,记住外层函数的变量
  2. @wraps(func) 保留原函数的元信息
  3. 带参数的装饰器要再包一层
  4. 装饰器顺序从下往上执行
  5. Python 3.9+ 用 functools.lru_cache 做缓存,比自己实现更可靠

print() 调试到装饰器统一处理,只需要花半小时学习一次,但能省下无数重复代码。

容器化部署的 Python 服务,跑在 K8s 上,平时没问题,流量一大就开始报 ConnectionPool is full。Pod 自动扩容后好了,缩容又报。这篇文章记录一下容器环境下的排查过程。

问题现象

服务部署在 K8s 上,HPA(Horizontal Pod Autoscaler)根据 CPU 使用率自动扩缩容。高峰期 Pod 数量从 3 扩到 10,这时候开始报错:

requests.exceptions.ConnectionError: ConnectionPool(pool_size=10, connections=10, maxsize=10, blocking=False)

坑在于:Pod 扩容后,每个 Pod 都有 20 个 worker 线程,但连接池只有 10 个位置。10 个 Pod 同时调用后端,连接池瞬间满了。

排查过程

第一步:查看 Pod 日志

# 查看 Pod 日志
kubectl logs -f deployment/backend-client --tail=100

# 看到大量 ConnectionPool is full 错误

第二步:确认后端状态

# 检查后端服务是否正常
kubectl exec -it deployment/backend-client -- curl -s http://backend-service:8080/health
# 返回 200,后端没问题

第三步:检查连接池配置

import requests

# 查看当前 Session 的连接池配置
s = requests.Session()
for host, adapter in s.adapters.items():
    print(f"Host: {host}, Pool: {adapter._pool_connections}, Max: {adapter._pool_maxsize}")

发现问题:每个 Pod 的连接池大小是 10,但每个 Pod 有 20 个 worker。10 个 Pod 同时调用,连接池瞬间满了。

问题原因

说白了,就是容器扩缩容导致并发数突增,连接池容量不够。

K8s HPA 扩容时,新 Pod 启动后立即开始处理请求。如果每个 Pod 有 20 个 worker,10 个 Pod 就是 200 个并发。但连接池只有 10 个位置,根本不够用。

坑在于:连接池大小是静态配置的,不会随着 Pod 数量动态调整。

解决方案

方案一:根据 Pod 数量动态调整连接池

import os
import requests
from requests.adapters import HTTPAdapter

# 从环境变量获取 Pod 数量(或 worker 数量)
pod_count = int(os.environ.get("POD_COUNT", 1))
workers_per_pod = int(os.environ.get("WORKERS_PER_POD", 20))

# 动态计算连接池大小
pool_size = pod_count * workers_per_pod

session = requests.Session()
adapter = HTTPAdapter(pool_connections=pool_size, pool_maxsize=pool_size)
session.mount("https://", adapter)
session.mount("http://", adapter)

方案二:使用共享连接池(推荐)

如果后端服务支持 HTTP/2,可以使用共享连接池:

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

session = requests.Session()

# 配置重试策略
retry = Retry(
    total=3,
    backoff_factor=0.1,
    status_forcelist=[500, 502, 503, 504],
)

adapter = HTTPAdapter(
    pool_connections=20,
    pool_maxsize=20,
    max_retries=retry,
)
session.mount("https://", adapter)
session.mount("http://", adapter)

方案三:限制并发数

import asyncio
import aiohttp

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    # 限制并发数为 10
    semaphore = asyncio.Semaphore(10)

    async def bounded_fetch(session, url):
        async with semaphore:
            return await fetch(session, url)

    async with aiohttp.ClientSession() as session:
        tasks = [bounded_fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)

asyncio.run(main())

生产环境最佳实践

  1. 连接池大小 >= 总并发数:总并发数 = Pod 数量 × 每个 Pod 的 worker 数
  2. 使用 HTTP/2:HTTP/2 支持多路复用,一个连接可以并发多个请求
  3. 监控连接池状态:在 Prometheus 里监控连接池使用率
  4. 设置合理的超时timeout=(3, 10) 连接超时 3 秒,读取超时 10 秒
# 生产环境推荐配置
import os
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# 动态计算连接池大小
pod_count = int(os.environ.get("POD_COUNT", 1))
workers_per_pod = int(os.environ.get("WORKERS_PER_POD", 20))
pool_size = pod_count * workers_per_pod

session = requests.Session()

retry = Retry(total=3, backoff_factor=0.1, status_forcelist=[500, 502, 503, 504])
adapter = HTTPAdapter(pool_connections=pool_size, pool_maxsize=pool_size, max_retries=retry)
session.mount("https://", adapter)
session.mount("http://", adapter)

踩坑总结

  1. 容器扩缩容会导致并发数突增,连接池容量不够
  2. 连接池大小是静态配置的,不会随 Pod 数量动态调整
  3. 总并发数 = Pod 数量 × 每个 Pod 的 worker 数
  4. 使用 HTTP/2 可以减少连接数
  5. 生产环境要监控连接池状态

这个问题的坑在于:本地测试没问题,上了 K8s 就报错。关键是理解容器环境下的并发模型,根据实际 Pod 数量和 worker 数量配置连接池。

用Python写了个批量下载脚本,可以实现批量爬取AUTOSAR官网标准文档。没有做多线程,异常处理逻辑也基本没有,仅能在网络良好情况下实现基本功能而已。
默认下载R20-11版本,分类保存到D盘根目录,修改categorycategory_names 理论上可以下载其他版本(AUTOSAR官网不改版的前提下),具体category的值可以打开Chrome/Edge浏览器开发者模式找。

- 阅读剩余部分 -