Python元编程与类型校验：DataFrame动态类型映射需求

核心思路：元类+泛型实现静态类型兼容 #

Python虽无TypeScript式的索引类型映射，但可通过元类+标准库泛型实现近似的类型安全效果，让mypy、Pyright等类型检查器识别动态生成的属性与索引操作。

1. 定义泛型RecordDF类（基于元类） #

元类会根据传入的数据类动态生成对应字段的属性，每个属性标注为Mapping[字段类型, 数据类类型]，确保类型检查器能识别索引操作的类型：

from dataclasses import dataclass, fields
from typing import Generic, TypeVar, Mapping, List, Any
import pandas as pd

T = TypeVar('T')

class RecordDFMeta(type):
    def __new__(cls, name, bases, namespace, record_type: type[T]):
        # 为数据类每个字段动态生成属性
        for field in fields(record_type):
            namespace[field.name] = property(
                lambda self, field_name=field.name: self._get_indexer(field_name)
            )
        return super().__new__(cls, name, bases, namespace)

class RecordDF(Generic[T], metaclass=RecordDFMeta, record_type=None):
    def __init__(self, df: pd.DataFrame, record_type: type[T]):
        self.df = df
        self.record_type = record_type
        self._field_series = {f.name: df[f.name] for f in fields(record_type)}

    def _get_indexer(self, field_name: str) -> Mapping[Any, T]:
        # 实现索引逻辑：根据字段值返回对应数据类实例
        series = self._field_series[field_name]
        index_map = {}
        for idx, value in series.items():
            record_data = self.df.iloc[idx].to_dict()
            index_map[value] = self.record_type(**record_data)
        return index_map

    def list(self) -> List[T]:
        # 返回所有数据类实例的列表
        return [self.record_type(**row.to_dict()) for _, row in self.df.iterrows()]

# 示例数据类
@dataclass
class Record:
    name: str
    phone: str
    age: int

# 构造示例DataFrame
df = pd.DataFrame([
    {"name": "Harry", "phone": "123456", "age": 30},
    {"name": "Alice", "phone": "654321", "age": 25}
])

# 创建实例
contacts = RecordDF(df, Record)

2. 类型检查效果验证 #

• 类型检查器会识别contacts.name为Mapping[str, Record]，因此contacts.name['Harry']会被推断为Record类型，支持.phone、.age的自动补全与类型校验。
• contacts.list()会被推断为List[Record]，完全符合需求。

动态从DataFrame生成数据类 #

若不想手动定义Record数据类，可通过dataclasses.make_dataclass结合Pandas的dtypes自动生成：

from dataclasses import make_dataclass
import numpy as np

def infer_record_type(df: pd.DataFrame) -> type:
    # 映射Pandas dtype到Python类型
    dtype_map = {
        'object': str,
        'int64': int,
        'float64': float,
        'bool': bool,
        'datetime64[ns]': pd.Timestamp
    }
    fields = []
    for col, dtype in df.dtypes.items():
        py_type = dtype_map.get(str(dtype), Any)
        fields.append((col, py_type))
    # 动态生成数据类
    return make_dataclass("InferredRecord", fields)

# 使用示例
inferred_record = infer_record_type(df)
contacts = RecordDF(df, inferred_record)
# 类型检查器会识别inferred_record的字段类型，支持自动补全

关键细节说明 #

• 元类的作用：动态为数据类字段生成属性，同时通过泛型T绑定数据类类型，让类型检查器能追踪具体字段类型。
• 索引器实现：_get_indexer返回的Mapping类型是类型检查器识别索引操作的核心，确保contacts.name['Harry']的类型被正确推断。
• 稳定性优化：预存_field_series避免重复从DataFrame取数，数据类构造逻辑统一放在RecordDF中，减少冗余代码。

内容的提问来源于stack exchange，提问作者jroz