Python类型系统深度实践：从类型提示到运行时验证

from typing import TypeVar, Generic

# 定义类型变量
T = TypeVar('T')
K = TypeVar('K')
V = TypeVar('V')

class Stack(Generic[T]):
    """泛型栈实现"""
    
    def __init__(self) -> None:
        self._items: list[T] = []
    
    def push(self, item: T) -> None:
        self._items.append(item)
    
    def pop(self) -> T:
        if not self._items:
            raise IndexError("Stack is empty")
        return self._items.pop()
    
    def peek(self) -> T:
        if not self._items:
            raise IndexError("Stack is empty")
        return self._items[-1]
    
    def is_empty(self) -> bool:
        return len(self._items) == 0

# 使用泛型
int_stack: Stack[int] = Stack()
int_stack.push(42)
int_stack.push(100)

str_stack: Stack[str] = Stack()
str_stack.push("hello")
str_stack.push("world")

from typing import TypeVar

# 约束类型变量必须可比较
T = TypeVar('T', bound='Comparable')

class Comparable:
    def __lt__(self, other: 'Comparable') -> bool:
        raise NotImplementedError

class Number(Comparable):
    def __init__(self, value: float) -> None:
        self.value = value
    
    def __lt__(self, other: 'Number') -> bool:
        return self.value < other.value
    
    def __repr__(self) -> str:
        return f"Number({self.value})"

def find_max(items: list[T]) -> T:
    """找到列表中的最大值"""
    if not items:
        raise ValueError("Empty list")
    
    max_item = items[0]
    for item in items[1:]:
        if item < max_item:
            max_item = item
    return max_item

# 使用
numbers = [Number(3.14), Number(2.71), Number(1.41)]
max_num = find_max(numbers)
print(max_num)  # Number(1.41)

from typing import Protocol, runtime_checkable

@runtime_checkable
class Drawable(Protocol):
    """可绘制的对象协议"""
    def draw(self) -> None: ...

@runtime_checkable
class Movable(Protocol):
    """可移动的对象协议"""
    def move(self, dx: float, dy: float) -> None: ...

class Circle:
    def __init__(self, radius: float) -> None:
        self.radius = radius
    
    def draw(self) -> None:
        print(f"Drawing circle with radius {self.radius}")
    
    def move(self, dx: float, dy: float) -> None:
        print(f"Moving circle by ({dx}, {dy})")

class Square:
    def __init__(self, side: float) -> None:
        self.side = side
    
    def draw(self) -> None:
        print(f"Drawing square with side {self.side}")

def render_shapes(shapes: list[Drawable]) -> None:
    """渲染所有形状"""
    for shape in shapes:
        shape.draw()

# 使用
circle = Circle(5.0)
square = Square(10.0)

render_shapes([circle, square])  # 正常工作

# 运行时检查
print(isinstance(circle, Drawable))  # True
print(isinstance(square, Drawable))  # True

from typing import Protocol, TypeVar, Generic

T = TypeVar('T')

class Container(Protocol[T]):
    """容器协议"""
    def get(self, key: str) -> T: ...
    def set(self, key: str, value: T) -> None: ...

class DictContainer(Generic[T]):
    """字典容器实现"""
    def __init__(self) -> None:
        self._data: dict[str, T] = {}
    
    def get(self, key: str) -> T:
        return self._data[key]
    
    def set(self, key: str, value: T) -> None:
        self._data[key] = value

class RedisContainer:
    """模拟Redis容器"""
    def __init__(self):
        self._cache = {}
    
    def get(self, key: str) -> str:
        return self._cache.get(key, "")
    
    def set(self, key: str, value: str) -> None:
        self._cache[key] = value

def process_container(container: Container[str]) -> None:
    """处理字符串容器"""
    container.set("key", "value")
    print(container.get("key"))

# 使用不同的容器实现
dict_container: Container[str] = DictContainer[str]()
redis_container: Container[str] = RedisContainer()

process_container(dict_container)  # 正常工作
process_container(redis_container)  # 正常工作

from typing import Literal, overload

Status = Literal["pending", "running", "success", "failed"]

def update_status(status: Status) -> None:
    print(f"Updating status to: {status}")

# 只能传递特定值
update_status("pending")    # OK
update_status("running")    # OK
update_status("completed")  # Type error!

# 与其他类型组合
def process(mode: Literal["fast", "slow"], count: int) -> None:
    if mode == "fast":
        print(f"Fast processing {count} items")
    else:
        print(f"Slow processing {count} items")

# 枚举值的字面量
from enum import Enum

class Color(Enum):
    RED = "red"
    GREEN = "green"
    BLUE = "blue"

ColorLiteral = Literal[Color.RED, Color.GREEN, Color.BLUE]

def set_color(color: ColorLiteral) -> None:
    print(f"Setting color to {color.value}")

set_color(Color.RED)   # OK
set_color(Color.YELLOW)  # Type error!

from typing import NewType, TypeAlias

# NewType创建新类型（运行时验证）
UserId = NewType('UserId', int)
OrderId = NewType('OrderId', str)

def get_user(user_id: UserId) -> dict:
    return {"id": user_id, "name": "John"}

# 类型检查器会强制区分int和UserId
user_id = UserId(123)
result = get_user(user_id)  # OK
# result = get_user(123)    # Type error!

# 类型别名（无运行时开销）
Coordinates: TypeAlias = tuple[float, float]
Matrix: TypeAlias = list[list[float]]
JSON: TypeAlias = dict[str, "JSON | list[JSON] | str | int | float | bool | None"]

# 复杂类型组合
Result[T, E]: TypeAlias = "Success[T] | Failure[E]"

class Success:
    def __init__(self, value: T) -> None:
        self.value = value

class Failure:
    def __init__(self, error: E) -> None:
        self.error = error

from typing import TypeGuard, Union

# 联合类型
Response = dict[str, Union[str, int, list[str], None]]

def is_string_response(response: Response) -> TypeGuard[dict[str, str]]:
    """类型守卫：收窄为字符串响应"""
    return "data" in response and isinstance(response["data"], str)

def is_list_response(response: Response) -> TypeGuard[dict[str, list[str]]]:
    """类型守卫：收窄为列表响应"""
    return "data" in response and isinstance(response["data"], list)

def process_response(response: Response) -> None:
    if is_string_response(response):
        # 类型收窄为 dict[str, str]
        print(f"String: {response['data'].upper()}")
    elif is_list_response(response):
        # 类型收窄为 dict[str, list[str]]
        for item in response["data"]:
            print(f"Item: {item}")
    else:
        # 其他类型
        print(f"Other: {response}")

# 使用isinstance进行类型收窄
def process_data(data: str | int | None) -> None:
    if isinstance(data, str):
        # 收窄为str
        print(f"String length: {len(data)}")
    elif isinstance(data, int):
        # 收窄为int
        print(f"Integer: {data * 2}")
    else:
        # 收窄为None
        print("No data")

from typing import (
    Generic, TypeVar, Optional, Union, 
    List, Dict, Set, Tuple, Callable
)

# 泛型容器
numbers: List[int] = [1, 2, 3, 4, 5]
mapping: Dict[str, int] = {"a": 1, "b": 2}
unique_ids: Set[int] = {1, 2, 3}

# 元组类型（固定长度和类型）
point: Tuple[int, int] = (10, 20)
rgb: Tuple[int, int, int] = (255, 128, 0)
mixed: Tuple[str, int, bool] = ("hello", 42, True)

# 可变参数元组
def process_coordinates(coords: Tuple[float, ...]) -> None:
    """处理任意数量的坐标点"""
    for x, y in zip(coords[::2], coords[1::2]):
        print(f"Point: ({x}, {y})")

# 函数类型
def callback(result: str) -> int:
    return len(result)

processor: Callable[[str], int] = callback
processor: Callable[[int, int], int] = lambda x, y: x + y

# 可调用类型的复杂形式
from typing import Any, Concatenate, ParamSpec

P = ParamSpec('P')
R = TypeVar('R')

def apply(func: Callable[P, R], *args: P.args, **kwargs: P.kwargs) -> R:
    """泛型函数应用"""
    return func(*args, **kwargs)

result = apply(lambda x, y: x + y, 1, 2)  # result: int

from typing import TypeVar, Generic, Callable, Any

T = TypeVar('T')
K = TypeVar('K')

# 泛型装饰器
def memoize(func: Callable[..., T]) -> Callable[..., T]:
    """记忆化装饰器"""
    cache: dict[tuple, T] = {}
    
    def wrapper(*args: Any, **kwargs: Any) -> T:
        key = (args, tuple(sorted(kwargs.items())))
        if key not in cache:
            cache[key] = func(*args, **kwargs)
        return cache[key]
    
    return wrapper

@memoize
def fibonacci(n: int) -> int:
    if n < 2:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)

# 泛型类工厂
def create_repository(
    entity_type: type[T], 
    id_type: type[K]
) -> type["Repository[T, K]"]:
    """创建通用仓储类"""
    
    class Repository(Generic[T, K]):
        def __init__(self) -> None:
            self._storage: dict[K, T] = {}
        
        def save(self, entity: T, id: K) -> None:
            self._storage[id] = entity
        
        def get(self, id: K) -> Optional[T]:
            return self._storage.get(id)
    
    return Repository

# 使用
class User:
    def __init__(self, name: str) -> None:
        self.name = name

UserRepository = create_repository(User, int)
repo = UserRepository()
repo.save(User("Alice"), 1)
user = repo.get(1)

from pydantic import BaseModel, Field, validator
from typing import Optional, List
from datetime import datetime

class User(BaseModel):
    """带验证的用户模型"""
    id: int
    username: str = Field(..., min_length=3, max_length=50)
    email: str
    age: Optional[int] = Field(None, ge=0, le=150)
    roles: List[str] = Field(default_factory=list)
    created_at: datetime = Field(default_factory=datetime.now)
    
    @validator('email')
    def validate_email(cls, v: str) -> str:
        if '@' not in v:
            raise ValueError('Invalid email address')
        return v.lower()
    
    class Config:
        validate_assignment = True

# 使用
user = User(
    id=1,
    username="john_doe",
    email="John@example.com",
    age=30,
    roles=["admin", "user"]
)

print(user.dict())
print(user.username)  # john_doe (自动转换为小写)

# 从JSON创建
data = {
    "id": 2,
    "username": "jane",
    "email": "jane@example.com"
}
user2 = User.parse_obj(data)
user2.age = -5  # 验证失败!

from typing import get_type_hints, get_origin, get_args
import re

class ValidationError(Exception):
    pass

def validate_object(obj: Any, schema: type) -> list[str]:
    """运行时验证对象"""
    errors = []
    type_hints = get_type_hints(schema)
    
    for field_name, expected_type in type_hints.items():
        value = getattr(obj, field_name, None)
        
        if value is None:
            continue
            
        if not validate_value(value, expected_type):
            errors.append(
                f"Field '{field_name}': expected {expected_type}, "
                f"got {type(value).__name__}"
            )
    
    return errors

def validate_value(value: Any, expected_type: type) -> bool:
    """验证单个值"""
    origin = get_origin(expected_type)
    
    # Optional[T] -> Union[T, None]
    if expected_type is type(None):
        return value is None
    
    # List[T]
    if origin is list:
        item_type = get_args(expected_type)[0]
        return all(validate_value(item, item_type) for item in value)
    
    # Dict[K, V]
    if origin is dict:
        key_type, val_type = get_args(expected_type)
        return all(
            validate_value(k, key_type) and validate_value(v, val_type)
            for k, v in value.items()
        )
    
    # 基础类型
    return isinstance(value, expected_type)

# 使用
class Config:
    host: str = "localhost"
    port: int = 8080
    debug: bool = False
    tags: list[str] = []

config = Config()
config.host = 123  # 类型错误
errors = validate_object(config, Config)
print(errors)  # ["Field 'host': expected , got str"]