Python typing模块深度指南:从类型检查到类型安全编程

# 变量类型注解name: str = "Python"version: float = 3.14is_awesome: bool = Truecount: int = 100pi: float = 3.1415926535# 函数参数和返回值类型注解defgreet(user: str) -> str:"""向用户问好    Args:        user: 用户名，必须是字符串类型    Returns:        问候语字符串    """returnf"Hello, {user}!"# 使用示例greeting = greet("开发者")  # 正确：传入字符串print(greeting)  # 输出: Hello, 开发者!# 类型检查会捕获的错误# greet(42)  # 错误：int无法赋值给str参数# greet(None)  # 错误：None无法赋值给str参数

from typing import Optional# 列表类型注解numbers:list[int] = [1, 2, 3, 4, 5]names:list[str] = ["Alice", "Bob", "Charlie"]# 字典类型注解config: dict[str, str] = {"host": "localhost", "port": "8080"}user_data: dict[str, Optional[str]] = {"name": "Alice", "age": None}# 集合类型注解unique_ids:set[int] = {1, 2, 3, 4, 5}tags:set[str] = {"python", "typing", "static"}# 元组类型注解（固定长度）point: tuple[float, float] = (3.5, 4.2)person: tuple[str, int, bool] = ("Alice", 30, True)# 元组类型注解（可变长度）scores: tuple[int, ...] = (95, 87, 92, 78)# 使用示例def process_users(users:list[str]) -> dict[str, int]:"""处理用户列表，返回用户名和字符数的映射    Args:        users: 用户名字符串列表    Returns:        用户名到名字长度的字典"""return {user: len(user) for user in users}# 调用示例result = process_users(["Alice", "Bob", "Charlie"])print(result)  # 输出: {'Alice': 5, 'Bob': 3, 'Charlie': 7}

from typing import Union# 传统Union写法defprocess_data(data: Union[str, bytes, int]) -> None:"""处理多种类型的数据    Args:        data: 可以是字符串、字节或整数    """if isinstance(data, str):        print(f"字符串: {data}")elif isinstance(data, bytes):        print(f"字节: {data!r}")else:        print(f"整数: {data}")# Python 3.10+推荐写法（使用|运算符）defprocess_data_v2(data: str | bytes | int) -> None:"""功能同上，但语法更简洁    Args:        data: 字符串、字节或整数    """if isinstance(data, str):        print(f"字符串: {data}")elif isinstance(data, bytes):        print(f"字节: {data!r}")else:        print(f"整数: {data}")# 使用示例process_data("hello")     # 正确：传入字符串process_data(b"world")    # 正确：传入字节process_data(42)          # 正确：传入整数# 类型检查会捕获的错误# process_data(3.14)      # 错误：float不在允许类型中# process_data([1, 2, 3]) # 错误：list不在允许类型中

from typing import Optional# 查找用户，可能找不到deffind_user(user_id: int) -> Optional[str]:"""根据用户ID查找用户名    Args:        user_id: 用户ID    Returns:        用户名或None（如果用户不存在）    """    users = {1: "Alice", 2: "Bob", 3: "Charlie"}return users.get(user_id)  # get方法返回Optional[str]# 正确处理Optional返回值username = find_user(1)if username isnotNone:    print(f"找到用户: {username}")# 这里username的类型是str，不是Optional[str]else:    print("用户不存在")# Python 3.10+可以使用|运算符deffind_user_v2(user_id: int) -> str | None:"""功能同上，语法更现代    Args:        user_id: 用户ID    Returns:        用户名或None    """    users = {1: "Alice", 2: "Bob", 3: "Charlie"}return users.get(user_id)# 重要区别：Optional与默认参数不同defconnect(host: Optional[str] = None) -> None:"""连接到指定主机    Args:        host: 主机地址，可选参数    """if host isNone:        host = "localhost"# 提供默认值    print(f"连接到 {host}")

from typing import Literal# HTTP方法约束HttpMethod = Literal["GET", "POST", "PUT", "DELETE", "PATCH"]defhandle_request(method: HttpMethod, url: str) -> None:"""处理HTTP请求    Args:        method: HTTP方法，必须是GET、POST、PUT、DELETE、PATCH之一        url: 请求URL    """    print(f"{method}{url}")# 正确调用handle_request("GET", "/api/users")handle_request("POST", "/api/users")# 类型检查会捕获的错误# handle_request("OPTIONS", "/api/users")  # 错误："OPTIONS"不在允许值中# 状态机示例GameState = Literal["init", "playing", "paused", "game_over"]defchange_state(current: GameState, next_state: GameState) -> bool:"""检查状态转换是否合法    Args:        current: 当前状态        next_state: 目标状态    Returns:        转换是否合法    """    transitions = {"init": ["playing"],"playing": ["paused", "game_over"],"paused": ["playing", "game_over"],"game_over": ["init"]    }return next_state in transitions[current]# 使用示例print(change_state("init", "playing"))    # Trueprint(change_state("playing", "init"))    # False

from typing import TypeVar, Sequence# 无约束的类型变量T = TypeVar('T')deffirst_element(items: Sequence[T]) -> T:"""返回序列的第一个元素    Args:        items: 任意类型的序列    Returns:        序列的第一个元素，类型与序列元素类型相同    """ifnot items:raise ValueError("序列不能为空")return items[0]# 自动推断类型first_int = first_element([1, 2, 3])      # 类型为intfirst_str = first_element(["a", "b", "c"]) # 类型为str# 有约束的类型变量Number = TypeVar('Number', int, float, complex)defadd_numbers(a: Number, b: Number) -> Number:"""数值加法，保持类型一致    Args:        a: 数值类型（int、float或complex）        b: 数值类型，必须与a类型一致    Returns:        相加结果，类型与输入相同    """return a + b# 正确调用print(add_numbers(5, 10))      # 15，类型为intprint(add_numbers(3.14, 2.71)) # 5.85，类型为float# 类型检查会捕获的错误# print(add_numbers(5, 3.14))   # 错误：类型不一致# 有边界的类型变量classAnimal:defspeak(self) -> str:return"..."classDog(Animal):defspeak(self) -> str:return"Woof!"A = TypeVar('A', bound=Animal)defmake_sound(animal: A) -> str:"""所有动物都会叫    Args:        animal: Animal或其子类的实例    Returns:        动物的叫声    """return animal.speak()# 使用示例dog = Dog()print(make_sound(dog))  # 输出: "Woof!"

from typing import Generic, TypeVar, ListT = TypeVar('T')classStack(Generic[T]):"""泛型栈实现    一个类型安全的栈数据结构，可以存储任意类型的元素    """def__init__(self) -> None:"""初始化空栈"""        self._items: List[T] = []defpush(self, item: T) -> None:"""元素入栈        Args:            item: 要入栈的元素        """        self._items.append(item)defpop(self) -> T:"""元素出栈        Returns:            栈顶元素        Raises:            IndexError: 如果栈为空        """ifnot self._items:raise IndexError("栈为空")return self._items.pop()defpeek(self) -> T:"""查看栈顶元素        Returns:            栈顶元素，但不移除        Raises:            IndexError: 如果栈为空        """ifnot self._items:raise IndexError("栈为空")return self._items[-1]defis_empty(self) -> bool:"""检查栈是否为空        Returns:            栈是否为空        """return len(self._items) == 0def__len__(self) -> int:"""栈的大小        Returns:            栈中元素的数量        """return len(self._items)def__str__(self) -> str:"""字符串表示        Returns:            栈的字符串表示        """returnf"Stack({self._items})"# 使用示例int_stack = Stack[int]()  # 创建整数栈int_stack.push(1)int_stack.push(2)int_stack.push(3)print(f"栈顶元素: {int_stack.peek()}")  # 输出: 3print(f"出栈元素: {int_stack.pop()}")   # 输出: 3print(f"栈大小: {len(int_stack)}")      # 输出: 2str_stack = Stack[str]()  # 创建字符串栈str_stack.push("Python")str_stack.push("TypeScript")print(f"字符串栈: {str_stack}")

from typing import Callable, Awaitableimport asyncio# 基本回调函数defon_button_click(callback: Callable[[int], None]) -> None:"""按钮点击事件处理器    Args:        callback: 回调函数，接受一个整数参数    """    print("按钮被点击了")    callback(42)  # 触发回调# 使用示例defhandle_click(data: int) -> None:    print(f"收到点击数据: {data}")on_button_click(handle_click)  # 输出: 按钮被点击了，收到点击数据: 42# 异步回调asyncdefasync_task() -> str:await asyncio.sleep(1)return"任务完成"# 异步回调类型async_callback: Callable[[], Awaitable[str]] = async_task# 任意参数的回调defflexible_handler(callback: Callable[..., str]) -> str:"""接受任意参数的回调函数    Args:        callback: 可接受任意参数的回调函数    Returns:        回调函数的返回值    """return callback("arg1", "arg2", arg3="value")# 使用示例result = flexible_handler(lambda *args, **kwargs: f"args: {args}, kwargs: {kwargs}")print(result)  # 输出: args: ('arg1', 'arg2'), kwargs: {'arg3': 'value'}

from typing import TypedDict, NotRequired# 基础TypedDictclassPerson(TypedDict):"""人员信息字典    Attributes:        name: 姓名，必填        age: 年龄，必填        is_student: 是否为学生，必填    """    name: str    age: int    is_student: bool# 使用示例alice: Person = {"name": "Alice","age": 25,"is_student": False}# 类型检查会验证键值类型# bob: Person = {"name": "Bob"}  # 错误：缺少age和is_student# 可选字段（Python 3.11+）classConfig(TypedDict):"""配置信息字典    Attributes:        host: 主机地址，必填        port: 端口号，必填        timeout: 超时时间，可选        ssl: 是否启用SSL，可选    """    host: str    port: int    timeout: NotRequired[float]  # 可选字段    ssl: NotRequired[bool]       # 可选字段# 正确：只提供必填字段config1: Config = {"host": "localhost", "port": 8080}# 正确：提供所有字段config2: Config = {"host": "example.com","port": 443,"timeout": 30.0,"ssl": True}# 继承TypedDictclassEmployee(TypedDict):"""员工信息基础字典    Attributes:        employee_id: 员工ID        department: 部门名称    """    employee_id: int    department: strclassManager(Employee, total=False):"""经理信息字典，继承员工信息    Attributes:        team_size: 团队规模，可选        budget: 预算，可选    """    team_size: int    budget: NotRequired[float]# 使用示例manager_data: Manager = {"employee_id": 1001,"department": "Engineering","team_size": 8,"budget": 500000.0}

from typing import Protocol, runtime_checkable# 基础ProtocolclassDrawable(Protocol):"""可绘制对象的协议"""defdraw(self) -> None:"""绘制方法"""        ...# 实现协议的具体类classCircle:defdraw(self) -> None:        print("绘制圆形")classSquare:defdraw(self) -> None:        print("绘制正方形")defrender_shape(shape: Drawable) -> None:"""渲染可绘制对象    Args:        shape: 符合Drawable协议的对象    """    shape.draw()# 正确：Circle和Square都符合Drawable协议circle = Circle()square = Square()render_shape(circle)  # 输出: "绘制圆形"render_shape(square)  # 输出: "绘制正方形"# 运行时检查（Python 3.8+）@runtime_checkableclassSerializable(Protocol):"""可序列化对象的协议"""defserialize(self) -> str:"""序列化方法"""        ...classDataModel:defserialize(self) -> str:return"序列化数据"# 运行时检查model = DataModel()print(isinstance(model, Serializable))  # 输出: True# 泛型Protocolfrom typing import TypeVarT = TypeVar('T', covariant=True)classContainer(Protocol[T]):"""泛型容器协议"""defget_item(self) -> T:"""获取容器中的元素"""        ...classIntContainer:defget_item(self) -> int:return42classStrContainer:defget_item(self) -> str:return"hello"defprocess_container(container: Container[str]) -> None:"""处理字符串容器    Args:        container: 包含字符串的容器    """    item = container.get_item()    print(f"容器内容: {item}")# 使用示例process_container(StrContainer())  # 正确# process_container(IntContainer())  # 错误：类型不匹配

from typing import Annotatedfrom typing_extensions import Doc  # Python 3.9+# 添加文档字符串的类型注解defprocess(    data: Annotated[str, Doc("要处理的文本数据")],    max_len: Annotated[int, Doc("最大长度限制")]) -> str:"""处理文本数据    Args:        data: 要处理的文本        max_len: 最大处理长度    Returns:        处理后的文本    """if len(data) > max_len:return data[:max_len] + "..."return data# 使用示例result = process("这是一个很长的文本需要被截断", 10)print(result)  # 输出: 这是一个很长的...# 验证约束的类型注解PositiveInt = Annotated[int, "必须是正整数"]NonEmptyStr = Annotated[str, "不能为空字符串"]defvalidate_input(    id: PositiveInt,    name: NonEmptyStr) -> None:"""验证输入数据    Args:        id: 正整数ID        name: 非空字符串名称    """    print(f"ID: {id}, 名称: {name}")# 正确调用validate_input(1001, "Alice")  # 正确# 第三方库集成示例（FastAPI风格）from fastapi import Querydefsearch_items(    q: Annotated[str, Query(min_length=3, max_length=50)],    page: Annotated[int, Query(ge=1, le=100)] = 1) -> dict:"""搜索接口    Args:        q: 搜索关键词，长度3-50        page: 页码，1-100之间    Returns:        搜索结果    """return {"query": q, "page": page}# 使用示例print(search_items("Python", 2))  # 输出: {'query': 'Python', 'page': 2}

from typing import TypeGuard, Union, Any# 类型守卫函数defis_string_list(value: list[Any]) -> TypeGuard[list[str]]:"""检查列表是否只包含字符串    Args:        value: 任意列表    Returns:        如果是字符串列表返回True    """return all(isinstance(item, str) for item in value)defsafe_process(strings: Union[list[str], list[int]]) -> None:"""安全处理字符串或整数列表    Args:        strings: 字符串列表或整数列表    """if is_string_list(strings):# 在这里，类型检查器知道strings是list[str]for s in strings:            print(s.upper())  # 安全调用字符串方法else:# 这里是list[int]        total = sum(strings)        print(f"总和: {total}")# 使用示例safe_process(["hello", "world"])  # 输出: HELLO WORLDsafe_process([1, 2, 3, 4])        # 输出: 总和: 10# 高级类型守卫：泛型类型守卫T = TypeVar('T')deffilter_by_type(    items: list[Any],    target_type: type[T]) -> TypeGuard[list[T]]:"""过滤出指定类型的元素    Args:        items: 任意列表        target_type: 目标类型    Returns:        如果所有元素都是目标类型返回True    """return all(isinstance(item, target_type) for item in items)defprocess_mixed_data(data: list[Union[str, int, float]]) -> None:"""处理混合类型数据    Args:        data: 包含字符串、整数、浮点数的列表    """if filter_by_type(data, str):# 所有元素都是字符串        print(f"字符串列表: {', '.join(data)}")elif filter_by_type(data, int):# 所有元素都是整数        print(f"整数总和: {sum(data)}")else:# 包含浮点数或其他类型        print(f"混合数据: {data}")

from typing import TypeVar, Generic, Protocolfrom abc import abstractmethod# 约束类型变量Comparable = TypeVar('Comparable', int, float, str)defmax_value(a: Comparable, b: Comparable) -> Comparable:"""返回两个可比较值的最大值    Args:        a: 第一个值        b: 第二个值，必须与a类型相同    Returns:        较大的值    """return a if a >= b else b# 使用示例print(max_value(10, 20))        # 输出: 20print(max_value(3.14, 2.71))    # 输出: 3.14print(max_value("apple", "banana"))  # 输出: "banana"# 边界类型变量classDisplayable(Protocol):"""可显示对象的协议"""    @abstractmethoddefdisplay(self) -> str:"""返回对象的字符串表示"""        ...D = TypeVar('D', bound=Displayable)classProduct:def__init__(self, name: str, price: float):        self.name = name        self.price = pricedefdisplay(self) -> str:returnf"{self.name}: ${self.price:.2f}"defshow_items(items: list[D]) -> None:"""显示可显示对象的列表    Args:        items: 可显示对象列表    """for item in items:        print(item.display())# 使用示例products = [    Product("Laptop", 999.99),    Product("Mouse", 29.99),    Product("Keyboard", 79.99)]show_items(products)

from typing import TypeAlias, Union, Optionalfrom typing_extensions import NotRequiredfrom datetime import datetime# 基础类型别名UserId: TypeAlias = intUserName: TypeAlias = strEmailAddress: TypeAlias = str# 复杂类型别名UserData: TypeAlias = dict[str, Union[str, int, None]]ApiResponse: TypeAlias = dict[str, Union[dict, list, str, int, bool, None]]# TypedDict别名from typing import TypedDictclassUserProfile(TypedDict):"""用户信息字典"""    id: UserId    name: UserName    email: EmailAddress    age: Optional[int]    created_at: datetime    updated_at: datetime# 嵌套泛型别名from typing import TypeVar, List, DictT = TypeVar('T')NestedList: TypeAlias = List[List[T]]StringDict: TypeAlias = Dict[str, Union[str, int, float, bool, None]]# 使用示例defprocess_user_profile(profile: UserProfile) -> str:"""处理用户信息    Args:        profile: 用户信息字典    Returns:        格式化后的用户信息    """returnf"用户: {profile['name']}, 邮箱: {profile['email']}"# 正确调用profile: UserProfile = {"id": 1001,"name": "Alice","email": "alice@example.com","age": 25,"created_at": datetime.now(),"updated_at": datetime.now()}print(process_user_profile(profile))

from typing import TypedDict, List, Optional, NotRequiredfrom datetime import datetimefrom dataclasses import dataclassfrom enum import Enum# 枚举类型定义classUserStatus(str, Enum):    ACTIVE = "active"    INACTIVE = "inactive"    SUSPENDED = "suspended"# 请求类型定义classCreateUserRequest(TypedDict):"""创建用户API请求类型"""    username: str    email: str    password: str    age: NotRequired[int]  # 可选字段    status: NotRequired[UserStatus]  # 可选字段，必须是枚举值# 响应类型定义classUserResponse(TypedDict):"""用户API响应类型"""    id: int    username: str    email: str    age: Optional[int]    status: UserStatus    created_at: datetime    updated_at: datetimeclassPaginatedResponse(TypedDict, total=False):"""分页响应类型"""    data: List[UserResponse]    total: int    page: int    per_page: int    total_pages: int# API服务接口定义from typing import ProtocolclassUserService(Protocol):"""用户服务协议"""defcreate_user(self, data: CreateUserRequest) -> UserResponse:"""创建用户"""        ...defget_user(self, user_id: int) -> Optional[UserResponse]:"""获取用户信息"""        ...deflist_users(        self,         page: int = 1,         per_page: int = 20,        status: Optional[UserStatus] = None    ) -> PaginatedResponse:"""列出用户"""        ...# 具体实现@dataclassclassDatabaseUserService:"""数据库用户服务实现"""defcreate_user(self, data: CreateUserRequest) -> UserResponse:"""创建用户实现        Args:            data: 用户创建数据        Returns:            创建的用户信息        """# 实际数据库操作        print(f"创建用户: {data['username']}")return {"id": 1,"username": data["username"],"email": data["email"],"age": data.get("age"),"status": data.get("status", UserStatus.ACTIVE),"created_at": datetime.now(),"updated_at": datetime.now()        }defget_user(self, user_id: int) -> Optional[UserResponse]:"""获取用户实现        Args:            user_id: 用户ID        Returns:            用户信息或None        """# 模拟数据库查询if user_id == 1:return {"id": 1,"username": "Alice","email": "alice@example.com","age": 25,"status": UserStatus.ACTIVE,"created_at": datetime.now(),"updated_at": datetime.now()            }returnNone# 使用示例defmain() -> None:# 创建用户服务    service = DatabaseUserService()# 创建用户请求（类型安全）    request: CreateUserRequest = {"username": "alice","email": "alice@example.com","password": "secure123","age": 25,"status": UserStatus.ACTIVE    }# 调用服务    user = service.create_user(request)    print(f"创建成功: {user['username']}")# 获取用户    found_user = service.get_user(1)if found_user:        print(f"找到用户: {found_user['email']}")if __name__ == "__main__":    main()

from typing import TypedDict, TypeVar, Generic, Callable, Optional, Unionfrom datetime import datefrom decimal import Decimalfrom dataclasses import dataclass# 原始数据格式classRawUserData(TypedDict):"""原始用户数据格式"""    user_id: str    name: str    birth_date: str  # "YYYY-MM-DD"    salary: str      # 字符串形式的数字    department: str# 验证后数据格式classValidatedUser(TypedDict):"""验证后的用户数据格式"""    id: int    name: str    birth_date: date    salary: Decimal    department_id: int# 泛型验证器T = TypeVar('T')@dataclassclassValidator(Generic[T]):"""泛型验证器基类"""    validation_func: Callable[[dict], T]    error_message: str = "数据验证失败"defvalidate(self, data: dict) -> T:"""验证数据        Args:            data: 原始数据字典        Returns:            验证后的数据        Raises:            ValueError: 如果验证失败        """try:return self.validation_func(data)except (ValueError, KeyError, TypeError) as e:raise ValueError(f"{self.error_message}: {e}")# 具体验证器实现classUserValidator(Validator[ValidatedUser]):"""用户数据验证器"""def__init__(self):        super().__init__(            validation_func=self._validate_user,            error_message="用户数据验证失败"        )def_validate_user(self, raw_data: RawUserData) -> ValidatedUser:"""验证用户数据        Args:            raw_data: 原始用户数据        Returns:            验证后的用户数据        """# 验证用户IDifnot raw_data["user_id"].isdigit():raise ValueError("用户ID必须是数字")# 验证出生日期try:            birth_date = date.fromisoformat(raw_data["birth_date"])except ValueError:raise ValueError("出生日期格式错误，应为YYYY-MM-DD")# 验证薪资try:            salary = Decimal(raw_data["salary"])except ValueError:raise ValueError("薪资格式错误")# 部门映射        dept_map = {"IT": 1, "HR": 2, "Finance": 3, "Sales": 4}        dept_id = dept_map.get(raw_data["department"])if dept_id isNone:raise ValueError(f"无效的部门: {raw_data['department']}")return {"id": int(raw_data["user_id"]),"name": raw_data["name"],"birth_date": birth_date,"salary": salary,"department_id": dept_id        }# 数据转换管道@dataclassclassDataPipeline:"""数据处理管道"""def__init__(self, *validators: Validator):        self.validators = validatorsdefprocess(self, data: dict) -> dict:"""处理数据        Args:            data: 原始数据        Returns:            处理后的数据        """        result = datafor validator in self.validators:            result = validator.validate(result)return resultdefbatch_process(self, data_list: list[dict]) -> list[dict]:"""批量处理数据        Args:            data_list: 原始数据列表        Returns:            处理后的数据列表        """        results = []        errors = []for i, data in enumerate(data_list):try:                result = self.process(data)                results.append(result)except ValueError as e:                errors.append({"index": i, "data": data, "error": str(e)})if errors:            print(f"处理完成，成功: {len(results)}，失败: {len(errors)}")for error in errors:                print(f"索引 {error['index']} 失败: {error['error']}")return results# 使用示例defmain() -> None:# 创建验证器    user_validator = UserValidator()# 创建处理管道    pipeline = DataPipeline(user_validator)# 准备测试数据    test_data: RawUserData = {"user_id": "1001","name": "Alice","birth_date": "1995-03-15","salary": "85000.50","department": "IT"    }# 处理数据try:        validated = pipeline.process(test_data)        print(f"验证成功:")        print(f"  ID: {validated['id']}")        print(f"  姓名: {validated['name']}")        print(f"  出生日期: {validated['birth_date']}")        print(f"  薪资: {validated['salary']}")        print(f"  部门ID: {validated['department_id']}")except ValueError as e:        print(f"验证失败: {e}")# 批量处理示例    batch_data = [        {"user_id": "1002", "name": "Bob", "birth_date": "1990-07-20", "salary": "75000", "department": "HR"},        {"user_id": "invalid", "name": "Charlie", "birth_date": "1985-12-10", "salary": "n/a", "department": "Finance"},        {"user_id": "1003", "name": "David", "birth_date": "2000-01-30", "salary": "60000", "department": "Sales"}    ]    print("\n批量处理结果:")    results = pipeline.batch_process(batch_data)    print(f"成功处理 {len(results)} 条数据")if __name__ == "__main__":    main()

from typing import Protocol, TypeVar, Generic, runtime_checkable, Anyfrom abc import abstractmethodimport importlibimport sys# 泛型类型变量T = TypeVar('T')@runtime_checkableclassPlugin(Protocol[T]):"""插件协议    所有插件必须实现的接口    """    @abstractmethoddefinitialize(self) -> None:"""初始化插件        在插件开始处理数据前调用        """        ...    @abstractmethoddefprocess(self, data: T) -> T:"""处理数据        Args:            data: 输入数据        Returns:            处理后的数据        """        ...    @abstractmethoddefcleanup(self) -> None:"""清理插件资源        在插件处理完成后调用        """        ...# 插件管理器classPluginManager(Generic[T]):"""类型安全的插件管理器    管理和调度插件的执行    """def__init__(self) -> None:"""初始化插件管理器"""        self._plugins: list[Plugin[T]] = []defregister(self, plugin: Plugin[T]) -> None:"""注册插件        Args:            plugin: 符合Plugin协议的对象        """ifnot isinstance(plugin, Plugin):raise TypeError(f"插件必须实现Plugin协议，但收到: {type(plugin)}")        self._plugins.append(plugin)defload_from_module(self, module_name: str) -> None:"""从模块加载插件        Args:            module_name: 模块名称        """try:            module = importlib.import_module(module_name)# 查找所有实现Plugin协议的类for attr_name in dir(module):# 跳过私有属性if attr_name.startswith('_'):continue                attr = getattr(module, attr_name)# 检查是否是可实例化的类ifnot isinstance(attr, type):continue# 跳过Protocol本身if attr is Plugin:continue# 检查是否实现了Plugin协议if issubclass(attr, Plugin) and attr isnot Plugin:try:                        plugin_instance = attr()                        self.register(plugin_instance)                        print(f"从模块 {module_name} 加载插件: {attr_name}")except Exception as e:                        print(f"实例化插件 {attr_name} 失败: {e}")except ImportError as e:            print(f"加载模块 {module_name} 失败: {e}")defprocess_data(self, data: T) -> T:"""通过所有插件处理数据        Args:            data: 输入数据        Returns:            经过所有插件处理后的数据        """        result = datafor plugin in self._plugins:try:                plugin.initialize()                result = plugin.process(result)                plugin.cleanup()except Exception as e:                print(f"插件执行失败: {e}")# 可以选择继续执行其他插件，或者终止continuereturn resultdefclear_plugins(self) -> None:"""清空所有插件"""        self._plugins.clear()# 具体插件实现classStringUppercasePlugin:"""字符串大写插件"""definitialize(self) -> None:        print("大写插件初始化完成")defprocess(self, data: str) -> str:"""将字符串转换为大写        Args:            data: 输入字符串        Returns:            大写字符串        """return data.upper()defcleanup(self) -> None:        print("大写插件清理完成")classStringReversePlugin:"""字符串反转插件"""definitialize(self) -> None:        print("反转插件初始化完成")defprocess(self, data: str) -> str:"""反转字符串        Args:            data: 输入字符串        Returns:            反转后的字符串        """return data[::-1]defcleanup(self) -> None:        print("反转插件清理完成")# 动态插件加载系统classPluginLoader:"""插件加载器    支持从目录、包、远程等多种方式加载插件    """    @staticmethoddefload_from_directory(directory: str, package_prefix: str = "") -> list[Plugin[Any]]:"""从目录加载插件        Args:            directory: 目录路径            package_prefix: 包前缀        Returns:            加载的插件列表        """import osimport importlib.util        plugins = []ifnot os.path.isdir(directory):            print(f"目录不存在: {directory}")return plugins# 遍历目录中的Python文件for filename in os.listdir(directory):if filename.endswith('.py') andnot filename.startswith('_'):                module_name = filename[:-3]  # 去掉.py后缀if package_prefix:                    full_module_name = f"{package_prefix}.{module_name}"else:                    full_module_name = module_name# 动态导入模块                filepath = os.path.join(directory, filename)                spec = importlib.util.spec_from_file_location(full_module_name, filepath)if spec isNone:                    print(f"无法为文件 {filepath} 创建spec")continue                module = importlib.util.module_from_spec(spec)                sys.modules[full_module_name] = moduletry:                    spec.loader.exec_module(module)# 查找插件类for attr_name in dir(module):                        attr = getattr(module, attr_name)if (isinstance(attr, type) and                            issubclass(attr, Plugin) and                            attr isnot Plugin):try:                                plugin_instance = attr()                                plugins.append(plugin_instance)                                print(f"从文件 {filename} 加载插件: {attr_name}")except Exception as e:                                print(f"实例化插件 {attr_name} 失败: {e}")except Exception as e:                    print(f"导入模块 {full_module_name} 失败: {e}")return plugins# 使用示例defmain() -> None:# 创建字符串插件管理器    manager = PluginManager[str]()# 注册插件    manager.register(StringUppercasePlugin())    manager.register(StringReversePlugin())# 处理数据    input_data = "hello world"    result = manager.process_data(input_data)    print(f"\n输入: {input_data}")    print(f"输出: {result}")  # 输出: "DLROW OLLEH"# 动态加载示例    print("\n动态加载插件:")# 示例：从当前目录加载插件（需要实际文件存在）# plugins = PluginLoader.load_from_directory(".", "plugins")# for plugin in plugins:#     manager.register(plugin)    print("插件系统演示完成")if __name__ == "__main__":    main()

# 模块A: models.pyfrom typing import TYPE_CHECKINGfrom typing import Listif TYPE_CHECKING:from services import UserService  # 只在类型检查时导入classUser:def__init__(self, service: "UserService"):# 字符串前向引用        self.service = service# 模块B: services.pyfrom typing import Listfrom models import User  # 这里会出现循环引用

# 解决方案1：使用字符串前向引用from typing import TYPE_CHECKINGif TYPE_CHECKING:# 只在类型检查时导入，运行时不会导入from services import UserServiceclassUser:def__init__(self, service: "UserService"):# 字符串引用        self.service = service# 解决方案2：使用__future__.annotations（Python 3.7+）from __future__ import annotationsclassNode:defconnect(self, other: Node) -> None:"""无需前向引用，自动处理循环引用"""        self.neighbor = other# 解决方案3：使用typing模块的ForwardReffrom typing import ForwardRefUserServiceRef = ForwardRef('UserService')classUser:def__init__(self, service: UserServiceRef):        self.service = service# 在实际导入后，更新ForwardReffrom services import UserServiceUserServiceRef._update_forward_refs(UserService=UserService)

from typing import TypeVar, SequenceT = TypeVar('T')deffirst_element(items: Sequence[T]) -> T:return items[0]# mypy可能无法准确推断嵌套列表的类型nested_list: list[list[int]] = [[1, 2], [3, 4]]# first = first_element(nested_list)  # mypy可能报错

# 解决方案1：显式类型注解from typing import TypeVar, Sequence, ListT = TypeVar('T')deffirst_element(items: Sequence[T]) -> T:return items[0]# 添加显式类型注解nested_list: list[list[int]] = [[1, 2], [3, 4]]first: list[int] = first_element(nested_list)  # 明确指定返回类型# 解决方案2：使用更具体的类型变量from typing import TypeVar, ListT = TypeVar('T')NestedList = List[List[T]]defprocess_nested(items: NestedList[T]) -> List[T]:"""处理嵌套列表，返回每个子列表的第一个元素"""return [sublist[0] for sublist in items]# 正确推断类型result = process_nested([[1, 2], [3, 4]])  # 返回list[int]: [1, 3]# 解决方案3：使用泛型类而非泛型函数from typing import Generic, TypeVar, ListT = TypeVar('T')classListProcessor(Generic[T]):def__init__(self, data: List[List[T]]):        self.data = datadeffirst_elements(self) -> List[T]:return [sublist[0] for sublist in self.data]# 使用示例processor = ListProcessor([[1, 2], [3, 4]])result = processor.first_elements()  # 正确推断为List[int]

# 动态属性访问defget_attr(obj: Any, attr_name: str) -> Any:return getattr(obj, attr_name)  # 类型检查器无法确定返回类型# 类型转换问题defparse_data(data: dict) -> Any:# 动态创建对象return type('DynamicClass', (), data)()

# 解决方案1：使用TypeVar和绑定from typing import TypeVar, AnyT = TypeVar('T', bound=Any)defsafe_getattr(obj: T, attr_name: str) -> Any:"""安全的属性访问，明确返回类型不确定性"""    result = getattr(obj, attr_name, None)# 这里无法确定具体类型，返回Anyreturn result# 解决方案2：使用结构化类型检查from typing import Protocol, runtime_checkablefrom typing_extensions import get_type_hints@runtime_checkableclassHasName(Protocol):    name: strdefprocess_named(obj: Any) -> str:"""处理具有name属性的对象"""if isinstance(obj, HasName):return obj.nameelse:return"Unknown"# 解决方案3：显式类型转换与验证from typing import cast, Type, Anydefvalidate_and_cast(data: dict, target_type: Type[T]) -> T:"""验证数据并转换为目标类型    Args:        data: 原始数据        target_type: 目标类型    Returns:        转换后的类型化数据    """# 这里执行实际验证逻辑ifnot all(hasattr(data, attr) for attr in target_type.__annotations__):raise TypeError(f"数据不符合 {target_type.__name__} 类型要求")return cast(T, data)  # 显式类型转换# 使用示例classUser:    name: str    age: intraw_data = {'name': 'Alice', 'age': 25}user = validate_and_cast(raw_data, User)print(f"用户: {user.name}, 年龄: {user.age}")

import requests  # 没有原生类型提示# mypy: error: Module has no attribute "get"

# 解决方案1：安装types包# pip install types-requestsimport requests# 解决方案2：手动添加类型存根# my_stubs/requests/__init__.pyifrom typing import Any, Optional, Dict, UnionclassResponse:    status_code: int    text: strdefjson(self) -> Any: ...defraise_for_status(self) -> None: ...defget(    url: str,    params: Optional[Dict[str, Any]] = None,    **kwargs: Any) -> Response: ...# 解决方案3：使用类型忽略注释import requests  # type: ignore# 或者针对特定行response = requests.get("https://api.example.com")  # type: ignore# 解决方案4：配置mypy忽略特定模块# mypy.ini[mypy-requests.*]ignore_missing_imports = True[mypy-pandas.*]ignore_missing_imports = True

from typing import Union, Optional, List, Dict, TypeVar, Generic, Callableimport sysT = TypeVar('T')U = TypeVar('U')defcomplex_function(    data: Union[List[Dict[str, Union[int, str, None]]],     callback: Optional[Callable[[T], U]] = None,    options: Optional[Dict[str, Any]] = None) -> Optional[Union[List[T], Dict[str, Any]]]:# 复杂的逻辑...pass

# 解决方案1：使用类型别名简化from typing import TypeAlias, List, Dict, Union, Optional, Any# 定义清晰的类型别名UserData: TypeAlias = Dict[str, Union[int, str, None]]UserList: TypeAlias = List[UserData]Processor: TypeAlias = Callable[[Any], Any]defsimplified_function(    data: UserList,    callback: Optional[Processor] = None,    options: Optional[Dict[str, Any]] = None) -> Optional[Union[List[Any], Dict[str, Any]]]:"""简化后的函数，类型定义更清晰"""# 清晰的逻辑...pass# 解决方案2：使用dataclass替代复杂字典from dataclasses import dataclassfrom typing import Optional, List@dataclassclassUser:"""用户数据类，比字典更清晰"""    name: str    age: int    email: Optional[str] = None    tags: Optional[List[str]] = Nonedefprocess_users(users: List[User]) -> List[str]:"""处理用户列表，类型更安全"""return [user.name for user in users]# 解决方案3：模块化类型定义# 在types.py中集中定义类型# types.pyfrom typing import TypeAlias, List, Dict, UnionApiResponse: TypeAlias = Dict[str, Union[List[Dict], str, int, None]]ConfigDict: TypeAlias = Dict[str, Union[str, int, bool, float]]# 在主模块中导入使用from types import ApiResponse, ConfigDictdefapi_handler(response: ApiResponse) -> None:"""处理API响应，类型引用更简洁"""# 清晰的逻辑...pass

# mypy.ini 高级配置示例[mypy]python_version = 3.12strict = Truewarn_unused_configs = Truedisallow_any_unimported = Truedisallow_any_expr = True[mypy-plugins.*]plugins =     mypy_django_plugin,    mypy_pydantic_plugin

{"python.analysis.autoImportCompletions": true,"python.analysis.indexing": true,"python.analysis.packageIndexDepths": [    {"name": "fastapi", "depth": 3},    {"name": "pydantic", "depth": 2}  ]}