动态重新加载 Cython 模块

我正在尝试自动更新我的 python 程序即时使用的 Cython .so 模块。下载新模块后del module and import modulePython 似乎仍在导入旧版本。

From 这个问题 https://stackoverflow.com/questions/17955194/how-to-reload-a-cython-module-interactively-using-pyximport，我已经尝试过，但没有成功：

from importlib import reload
import pyximport
pyximport.install(reload_support=True)
import module as m
reload(m)

From 这个问题 https://stackoverflow.com/questions/21630060/unloading-a-module-in-python-3-x，我也尝试过这个，但它也不起作用：

del sys.modules['module']
del module
import module

我也尝试过这个但没有成功：

from importlib import reload
import my_module

my_module = reload(my_module)

知道如何即时导入 Cython .SO 文件吗？

update_filename = "my_module.cpython-37m-darwin.so"

if __name__ == '__main__':
    response = check_for_update()
    if response != "No new version available!":
        print (download_update(response))

def check_for_update():
    print("MD5 hash: {}".format(md5(__file__)))
    s = setup_session()
    data = {
        "hash": md5(__file__),
        "type": "md5",
        "platform": platform.system()
    }
    response = s.post(UPDATE_CHECK_URL, json=data)
    return response.text

def download_update(url):
    s = setup_session()
    with s.get(url, stream=True) as r:
        r.raise_for_status()
        with open(update_filename, 'wb') as f:
            for chunk in r.iter_content(chunk_size=8192): 
                if chunk:
                    f.write(chunk)
    return update_filename

下载新的 SO 文件后，我手动输入上面列出的命令。

所以我从来没有找到正确的方法来做到这一点，而是通过将我的程序分成两个单独的部分来解决这个问题：

1. 不变的“运行程序”部分，只需导入 .SO 文件（或 Windows 上的 .PYD 文件）并在其中运行函数
2. 包含核心逻辑的实际 .SO（或 .PYD）文件

不变的脚本检查 .SO/.PYD 文件的更新（使用它的 SHA256 哈希 + 模块版本），当找到更新版本时，它会下载它并替换现有的 .SO/.PYD 文件并重新启动自身，从而加载更新的模块。

当未找到更新时，它会导入本地 .SO/.PYD 文件并运行其中的函数。这种方法在 Windows 和 OSX 上都对我有效。

运行程序脚本 (run.py)

import requests, os, sys
from pathlib import Path
from shutil import move

original_filename = "my_module.cp38-win32.pyd" # the filename of the Cython module to load
update_filename = f"{original_filename}.update"
UPDATE_SERVER = "https://example.com/PROD/update-check"

def check_for_update():
    replace_current_pyd_with_previously_downloaded_update() # Actually perform the update
    # Add your own update check logic here
    # This one checks with {UPDATE_SERVER} for updates to {original_filename} and returns the direct link to the updated PYD file if an update exists
    s = requests.Session()
    data = {
        "hash": sha256(original_filename),
        "type": "sha256",
        "current_version": get_daemon_version(),
        "platform": platform.system()
    }
    response = s.post(UPDATE_SERVER, json=data)
    return response.text # direct link to newer version of PYD file if update exists

def download_update(url):
    # Download updated PYD file from update server and write/replace {update_filename}

def replace_current_pyd_with_previously_downloaded_update():
    print("Checking for previously downloaded update file")
    update_file_path = Path(update_filename)
    if update_file_path.is_file():
        print(f"Update file found! Performing update by replacing {original_filename} with the updated version and deleting {update_filename}")
        move(update_filename, original_filename)
    else:
        print("No previously downloaded update file found. Checking with update server for new versions")

def get_daemon_version():
    from my_module import get_version
    return get_version() # my_module.__version__.lower().strip()

def restart():
    print ("Restarting to apply update...\r\n")
    python = sys.executable
    os.execl(python, python, *sys.argv)

def apply_update():
    restart()

def start_daemon():
    import my_module
    my_module.initiate()
    my_module.start()

if __name__ == "__main__":
    response = None
    print ("Checking to see if an update is available...")
    try:
        response = check_for_update()
    except Exception as ex:
        print ("Unable to check for updates")
        pass
    if response is None:
        print ("Unable to check for software updates. Using locally available version.")
        start_daemon()
    elif response != "No new version available!" and response != '':
        print ("Newer version available. Updating...")
        print ("Update downloaded: {}".format(download_update(response)))
        apply_update()
        start_daemon()
    else:
        print ("Response from update check API: {}\r\n".format(response))
        start_daemon()

.SO/.PYD 文件

实际的 .SO 文件（在本例中为 .PYD 文件）应包含一个名为的方法get_version它应该返回模块的版本，并且您的更新服务器应该包含逻辑来确定更新是否可用于（SHA256 + module_version）的组合。

您当然可以以完全不同的方式实现更新检查。