You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何用带类型注解的泛型函数重构Python中查找Pydantic模型的相似函数?

问题描述

我定义了多个函数,用于根据属性值在Pydantic模型列表中查找特定的Pydantic模型,代码示例如下:

SocketIOUserSessionID = str
RoomWithIndex = tuple[Room, RoomIndex]
RSStateWithIndex = tuple[RSState, int]
RSPacketSavedRecordWithIndex = tuple[RSPacketSavedRecordsContainer, int]


def find_room_by_id(self, id: UUID | str, where: list[Room]) -> RoomWithIndex | None:
    room = next(filter(lambda room: room.id == id, where), None)
    if room is None:
        return None
    index = where.index(room)
    return room, index

def find_room_by_session(self, session: SocketIOUserSessionID, where: list[Room]) -> RoomWithIndex | None:
    room = next(filter(lambda room: session in room.sessions, where), None)
    if room is None:
        return None
    index = where.index(room)
    return room, index

def find_rs_state_by_room_id(self, room_id: str, where: list[RSState]) -> RSStateWithIndex | None:
    rs_state = next(filter(lambda rs_state: rs_state.room_id == room_id, where), None)
    if rs_state is None:
        return None
    index = where.index(rs_state)
    return rs_state, index

def find_saved_record_by_room_id(self, room_id: str, where: list[RSPacketSavedRecordsContainer]) -> RSPacketSavedRecordWithIndex | None:
    saved_record = next(filter(lambda saved_records: saved_records.room_id == room_id, where), None)
    if saved_record is None:
        return None
    index = where.index(saved_record)
    return saved_record, index

请问如何编写带类型注解的泛型函数来重构这类代码?我了解functools.singledispatch装饰器,但不确定是否适用于此场景。

解决方案

1. 泛型函数+谓词函数抽离核心逻辑

所有这些查找函数的核心逻辑高度一致:遍历列表找到第一个匹配元素,返回元素与索引(无匹配则返回None)。我们可以把这部分通用逻辑抽成泛型函数,将匹配条件作为参数传入,避免重复代码。

代码示例:

from typing import TypeVar, Callable, Optional, List, Tuple
from pydantic import BaseModel

# 定义绑定到Pydantic模型的泛型类型
T = TypeVar('T', bound=BaseModel)
FoundWithIndex = Tuple[T, int] | None

def find_with_index(self, predicate: Callable[[T], bool], where: List[T]) -> FoundWithIndex:
    match = next(filter(predicate, where), None)
    if not match:
        return None
    return match, where.index(match)

基于这个泛型函数,原来的各个查找函数可以简化为:

def find_room_by_id(self, id: UUID | str, where: List[Room]) -> FoundWithIndex:
    return self.find_with_index(lambda room: room.id == id, where)

def find_room_by_session(self, session: SocketIOUserSessionID, where: List[Room]) -> FoundWithIndex:
    return self.find_with_index(lambda room: session in room.sessions, where)

def find_rs_state_by_room_id(self, room_id: str, where: List[RSState]) -> FoundWithIndex:
    return self.find_with_index(lambda rs: rs.room_id == room_id, where)

def find_saved_record_by_room_id(self, room_id: str, where: List[RSPacketSavedRecordsContainer]) -> FoundWithIndex:
    return self.find_with_index(lambda rec: rec.room_id == room_id, where)

2. 属性匹配场景的进一步简化

如果大部分查找都是基于属性值相等的逻辑,可以再封装一个更针对性的泛型函数:

from typing import Any

def find_by_attr(self, attr_name: str, attr_value: Any, where: List[T]) -> FoundWithIndex:
    return self.find_with_index(lambda item: getattr(item, attr_name) == attr_value, where)

这样像find_rs_state_by_room_id这类函数可以进一步缩短:

def find_rs_state_by_room_id(self, room_id: str, where: List[RSState]) -> FoundWithIndex:
    return self.find_by_attr("room_id", room_id, where)

3. 关于functools.singledispatch的适用性

singledispatch的核心是根据第一个参数的类型分发不同实现,而你的场景中,不同查找函数的差异是匹配条件而非参数类型,因此用它意义不大。除非你需要针对不同模型类型(比如Room、RSState)定义完全独立的查找逻辑,这时才适合使用,但这种写法会让代码更冗余,不如泛型+谓词的方式灵活简洁。

4. 类型提示优化

如果希望返回类型更精确(比如明确返回Tuple[Room, int]而非Tuple[T, int]),可以利用泛型绑定让IDE自动推断更准确的类型:

TModel = TypeVar('TModel', bound=BaseModel)
FoundModelWithIndex = Tuple[TModel, int] | None

def find_with_index(self, predicate: Callable[[TModel], bool], where: List[TModel]) -> FoundModelWithIndex:
    match = next(filter(predicate, where), None)
    if not match:
        return None
    return match, where.index(match)

当传入List[Room]时,IDE会自动识别返回类型为Tuple[Room, int] | None。

内容的提问来源于stack exchange,提问作者Slava Pasedko

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.06.23 01:00:15