Python动态属性赋值与类型注解：延迟导入场景下的挑战与实践

class _ModuleRegistry(object):
    _modules = {}

    def defer_import(
        self,
        import_statement: str,
        import_name: str,
    ):
        """
        注册一个延迟导入的模块或函数。
        import_statement: 完整的导入语句，例如 "from pandas import read_csv"
        import_name: 导入的名称，例如 "read_csv"
        """
        self._modules[import_name] = import_statement
        setattr(self, import_name, None) # 初始设置属性为None

    def __getattribute__(self, __name: str):
        """
        拦截属性访问，实现延迟导入。
        """
        # 排除特殊属性和已知的内部属性
        if (
            __name
            and not __name.startswith("__")
            and __name not in ("defer_import", "_modules")
        ):
            import_statement = self._modules.get(__name)
            if import_statement:
                # 动态执行导入语句
                exec(import_statement, locals())
                # 将导入的对象设置为实例属性
                setattr(self, __name, locals().get(__name))

            # 返回已导入或已设置的属性值
            ret_val = locals().get(__name) # 尝试从当前locals获取
            if ret_val:
                return ret_val
            else:
                # 如果仍未找到，可能是在defer_import中设置的None，或者不存在
                # 此时应再次尝试从实例属性中获取，因为exec可能已更新它
                return super().__getattribute__(__name) 
        else:
            # 对于特殊属性或已知属性，回退到默认行为
            val = super().__getattribute__(__name)
            return val

registry = _ModuleRegistry()
registry.defer_import("from pandas import read_csv", "read_csv")

# 此时，我们希望类型检查器能知道 registry.read_csv 是一个函数
print(registry.read_csv)

from typing import TYPE_CHECKING

# 在类型检查阶段，我们为 registry 对象提供一个具有预期属性的“模拟”
if TYPE_CHECKING:
    # 假设我们知道 registry 会有 defaultdict 和 Namespace 属性
    # 这里我们使用 argparse.Namespace 作为 registry 的一个简单模拟对象，
    # 因为它支持属性赋值，且在 mypy-play 环境中容易验证。
    # 实际应用中，这里应该模拟 _ModuleRegistry 实例的预期属性。
    from collections import defaultdict
    from argparse import Namespace

    # 创建一个模拟的 registry 对象，并为其添加类型检查器期望的属性
    # 注意：这里的 registry 只是一个类型检查时的“幻影”，与运行时实际的 registry 对象不同
    registry = Namespace() 
    registry.defaultdict = defaultdict
    # 模拟 pandas.read_csv
    # from pandas import read_csv # 如果 pandas 可用，可以直接导入
    # registry.read_csv = read_csv
else:
    # 在运行时，使用实际的 _ModuleRegistry 类
    class _ModuleRegistry:
        _modules = {}

        def defer_import(
            self,
            import_statement: str,
            import_name: str,
        ):
            self._modules[import_name] = import_statement
            setattr(self, import_name, None)

        def __getattribute__(self, __name: str):
            if (
                __name
                and not __name.startswith("__")
                and __name not in ("defer_import", "_modules")
            ):
                import_statement = self._modules.get(__name)
                if import_statement:
                    exec(import_statement, locals())
                    setattr(self, __name, locals().get(__name))
                ret_val = locals().get(__name)
                if ret_val:
                    return ret_val
                else:
                    return super().__getattribute__(__name) # fallback
            else:
                val = super().__getattribute__(__name)
                return val

    registry = _ModuleRegistry()

# 运行时实际的 defer_import 调用
# 这里使用 defaultdict 作为示例，与 TYPE_CHECKING 块中的模拟保持一致
registry.defer_import("from collections import defaultdict", "defaultdict")

# 此时，类型检查器会根据 TYPE_CHECKING 块中的定义，
# 推断 registry.defaultdict 的类型。
# Mypy 的 reveal_type 可以帮助我们验证这一点：
# reveal_type(registry.defaultdict)
# Mypy 输出: Revealed type is "Overload(def [_KT, _VT] () -> collections.defaultdict[_KT`1, _VT`2], ...)"

def process_data(file_path: str):
    """
    处理数据文件，延迟导入 pandas。
    """
    from pandas import read_csv # 只有在调用 process_data 时才导入
    df = read_csv(file_path)
    # ... 其他数据处理逻辑
    return df

# 此时 pandas 并未导入
print("pandas 尚未导入") 

# 第一次调用时导入 pandas
data = process_data("my_data.csv") 
print("pandas 已导入并使用")

import types
from typing import Callable, Dict, Any, Optional

class LazyLoader:
    """
    一个简单的延迟加载器，通过函数返回导入的对象。
    """
    def __init__(self):
        self._load_funcs: Dict[str, Callable[[], Any]] = {}
        self._loaded_modules: Dict[str, Any] = {}

    def register_loader(self, name: str, loader_func: Callable[[], Any]):
        """
        注册一个加载函数。当访问 name 时，会调用 loader_func。
        loader_func 应该返回要加载的对象。
        """
        self._load_funcs[name] = loader_func

    def __getattr__(self, name: str) -> Any:
        """
        拦截属性访问，实现延迟加载。
        """
        if name in self._loaded_modules:
            return self._loaded_modules[name]

        if name in self._load_funcs:
            obj = self._load_funcs[name]()
            self._loaded_modules[name] = obj
            return obj

        raise AttributeError(f"'{type(self).__name__}' object has no attribute '{name}'")

# 示例使用
lazy_registry = LazyLoader()

# 注册一个加载 read_csv 的函数
def load_read_csv():
    from pandas import read_csv
    return read_csv

lazy_registry.register_loader("read_csv", load_read_csv)

# 此时 pandas 尚未导入
print("pandas 尚未导入")

# 第一次访问时，load_read_csv 会被调用，并返回 read_csv 函数
csv_reader: Callable = lazy_registry.read_csv
print(f"read_csv 的类型: {type(csv_reader)}")

# 后续访问直接返回已加载的函数
another_csv_reader = lazy_registry.read_csv