如何扩展importlib远程导入系统以支持Numpy的.pyd和.pyi文件？

我正在开发一个基于importlib的Numpy远程导入系统，将本地site-packages中的Numpy文件夹托管在Web服务器上。以下是我的当前代码：

import importlib
import importlib.abc
import importlib.machinery
from types import ModuleType
import sys
import requests

class Finder(importlib.abc.MetaPathFinder):

    def __init__(self, base_url) -> None:
        self.base_url = base_url
    
    def find_spec(self, fullname, path, target=None):
        spec = self.find_py_file_spec(fullname)
        if spec is not None:
            return spec
        spec = self.find_package_spec_init(fullname)
        if spec is not None:
            return spec
        return None
    
 
    def find_py_file_spec(self, fullname):
        if len(fullname.split(".")) == 1:
            url = "{}/{}/{}.py".format(self.base_url,fullname.split(".")[0],fullname.replace(".","/"))
        else:
            url = "{}/{}.py".format(self.base_url,fullname.replace(".","/"))
        print(url)
        source = self.get_source_code(url)
        if source is None:
            print("no thing found")
            return None
        loader = Loader(fullname,source,url)
        return importlib.machinery.ModuleSpec(fullname, loader)
    
    def find_package_spec_init(self,fullname):

        url = "{}/{}/__init__.py".format(self.base_url,fullname.replace(".","/"))
        source = self.get_source_code(url)
        if source is None:
            print("no init found")
            return None
        loader = Loader(fullname,source,url,zip=True)
        return importlib.machinery.ModuleSpec(fullname, loader,is_package=True,)
    
    def get_source_code(self, url):
        try:
            response = requests.get("http://"+url)
            response.raise_for_status()
        except requests.HTTPError:
            return None
            print("import failed due to HTTP")
        
        
        source = response.text
        return source
class Loader(importlib.abc.Loader):

    def __init__(self, name, source_code, url,zip=False) -> None:
        self.name = name
        self.source_code = source_code
        self.url = url
        self.zip = zip
    
    def create_module(self, spec: importlib.machinery.ModuleSpec) -> ModuleType | None:
        module = sys.modules.get(spec.name)
        if module is None:
            module = ModuleType(spec.name)
            sys.modules[spec.name] = module
        return module
    
    def exec_module(self, module: ModuleType) -> None:
        module.__file__ = self.url
        print("url is {}".format(self.url))
        exec(self.source_code, module.__dict__)
        return module

    def get_source(self,name):
        return self.source_code
    
def add_server(ip,port):
    sys.meta_path.append(Finder(f"{ip}:{port}"))


add_server("127.0.0.1",5002)
for f in sys.meta_path:
    print(f)
import numpy

该代码仅能处理仅包含.py文件的包，但Numpy包含.pyd和.pyi文件，请问如何处理这些类型的文件以实现Numpy的成功远程导入？

要支持Numpy中的.pyd（Windows编译扩展模块）、.so（类Unix编译扩展模块）和.pyi（类型存根文件），需要扩展现有Finder和Loader的逻辑，具体实现如下：

1. 扩展Finder，支持多文件类型查找 #

在Finder类中新增扩展模块和类型存根的查找方法，并更新find_spec的优先级（优先加载编译模块，再处理存根和源码）：

def find_extension_spec(self, fullname):
    # 根据平台选择扩展文件后缀
    ext = ".pyd" if sys.platform == "win32" else ".so"
    url = f"{self.base_url}/{fullname.replace('.', '/')}{ext}"
    try:
        # 仅验证文件是否存在，不下载完整内容
        response = requests.head(f"http://{url}")
        response.raise_for_status()
        return importlib.machinery.ModuleSpec(
            fullname,
            ExtensionLoader(fullname, url),
            origin=url
        )
    except requests.HTTPError:
        return None

def find_stub_spec(self, fullname):
    # 查找类型存根文件
    url = f"{self.base_url}/{fullname.replace('.', '/')}.pyi"
    source = self.get_source_code(url)
    if source is None:
        return None
    loader = StubLoader(fullname, source, url)
    return importlib.machinery.ModuleSpec(fullname, loader)

# 更新原find_spec方法
def find_spec(self, fullname, path, target=None):
    # 1. 优先检查编译扩展模块
    spec = self.find_extension_spec(fullname)
    if spec is not None:
        return spec
    # 2. 检查类型存根文件
    spec = self.find_stub_spec(fullname)
    if spec is not None:
        return spec
    # 3. 原有逻辑：检查.py文件
    spec = self.find_py_file_spec(fullname)
    if spec is not None:
        return spec
    # 4. 原有逻辑：检查包的__init__.py
    spec = self.find_package_spec_init(fullname)
    if spec is not None:
        return spec
    return None

2. 实现ExtensionLoader加载编译扩展模块 #

编译模块无法通过exec执行，需要下载到本地临时文件后加载：

import tempfile
import os

class ExtensionLoader(importlib.abc.Loader):
    def __init__(self, name, url):
        self.name = name
        self.url = url

    def create_module(self, spec):
        # 使用默认模块创建逻辑
        return None

    def exec_module(self, module):
        # 下载扩展模块到临时文件
        ext_suffix = ".pyd" if sys.platform == "win32" else ".so"
        with tempfile.NamedTemporaryFile(delete=False, suffix=ext_suffix) as tmp_file:
            response = requests.get(f"http://{self.url}")
            tmp_file.write(response.content)
            tmp_path = tmp_file.name
        
        try:
            # 加载临时文件中的扩展模块
            spec = importlib.util.spec_from_file_location(self.name, tmp_path)
            ext_module = importlib.util.module_from_spec(spec)
            spec.loader.exec_module(ext_module)
            # 将扩展模块的属性复制到当前模块
            for attr_name in dir(ext_module):
                if not attr_name.startswith("__"):
                    setattr(module, attr_name, getattr(ext_module, attr_name))
        finally:
            # 清理临时文件
            os.unlink(tmp_path)

3. 实现StubLoader处理类型存根文件 #

类型存根文件的加载逻辑与.py文件类似，只需标记为存根模块：

class StubLoader(Loader):
    def exec_module(self, module: ModuleType) -> None:
        module.__file__ = self.url
        module.__stub__ = True  # 标记为存根模块，避免影响运行时逻辑
        exec(self.source_code, module.__dict__)

4. 关键注意事项 #

• 包结构一致性：确保Web服务器上的Numpy目录结构与本地site-packages完全一致，包括所有子模块和依赖文件
• 缓存优化：编译模块体积较大，可添加本地缓存机制，避免重复下载相同文件
• 安全风险：远程加载编译模块存在代码注入风险，仅在可信环境中使用
• 跨平台兼容：不同平台的扩展模块格式不通用，需根据运行平台匹配对应的文件

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