python - 动态重新加载 Cython 模块

Question

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

来自 this question ，我已经试过了，但没有成功:

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

来自 this question ，我也试过这个，但也没用:

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 文件后，我手动输入了上面列出的命令。

Answer 1

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

1. 一个不变的“runner”部分，它只导入 .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) 的组合。

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