大数据

dataclass于pydantic回顾

dataclass

from dataclasses import dataclass, field
from typing import List


@dataclass
class Book:
    title: str
    author: str
    price: float = 0.0 # 有默认值的,必须放在无默认值的后面
    # default_factory 会为每个实例创建一个新的空列表。
    tags: List[str] = field(default_factory=list) # 可变默认值要用default_factory

# 自动生成__init__
book = Book(title="Python入门", author="张三", price=59.9)
print(book)

pydantic

demo01

from pydantic import BaseModel, ValidationError, Field
from typing import List, Optional


class User(BaseModel):
    id: int
    name: str
    age: int = 18
    email: str
    # 可以使用Field增加校验 比如长度限制
    password: str = Field(min_length=8)


# 自动类型转换
user = User(id="123", name="Alice", email="alice@test.com", password="secret12")
print(user.id)
print(type(user.id))

user_dict = user.model_dump()
print(type(user_dict))
print(user_dict)

user_json = user.model_dump_json()
print(type(user_json))
print(user_json)

# 字典转模型
new_user = User.model_validate(user_dict)
print(type(new_user))
print(new_user)

# 验证失败抛出 ValidationError
# try:
#     user = User(id="abc", name="Bob", email="b@b.com", password="123")
# except ValidationError as e:
#     print(e)

demo02

from typing import List, Dict, Optional, Union, Any, Tuple, Callable


def process_data(
        items: List[int],  # 整数列表
        mapping: Dict[str, float],  # 字符串到浮点数的字典
        maybe: Optional[str] = None,  # 可能是str或None
        union_type: Union[int, str] = 10,  # int或str
        anything: Any = None  # 任意类型
) -> Tuple[int, float]:
    return len(items), sum(mapping.values())

demo03

from dataclasses import dataclass
from pydantic import BaseModel
from typing import Callable, List, Optional
import json


# Pydantic 模型:用来验证收到的JSON请求体
class CreateUserRequest(BaseModel):
    username: str
    age: int
    email: Optional[str] = None


# dataclass: 内部用的轻量数据容器
@dataclass
class UserDB:
    username: str
    age: int
    email: Optional[str] = None


# Callable 类型别名:一个处理函数
# Callable[[入参类型列表], 返回参数类型]
HandlerFunc = Callable[[CreateUserRequest], UserDB]


# 具体的处理函数
def handle_create_user(req: CreateUserRequest) -> UserDB:
    # 这里可以写业务逻辑 然后返回内部模型
    return UserDB(username=req.username, age=req.age, email=req.email)


# 模拟API路由
def api_endpoint(raw_json: str, handler: HandlerFunc) -> str:
    # 1. 用Pydantic解析和验证
    data = CreateUserRequest.model_validate(json.loads(raw_json))
    # 2. 执行业务逻辑
    user_db = handler(data)
    # 返回简单的JSON字符串
    return json.dumps({"status": "ok", "username": user_db.username})


# 测试
json_input = '{"username": "李四", "age": 25}'
result = api_endpoint(json_input, handle_create_user)
print(result)  # {"status": "ok", "username": "李四"}