我正在尝试自动更新我的 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 文件后,我手动输入了上面列出的命令。
最佳答案
所以我从来没有找到正确的方法来做到这一点,而是通过将我的程序分成两个独立的部分来解决这个问题:
- 一个不变的“runner”部分,它只导入 .SO 文件(或 Windows 上的 .PYD 文件)并在其中运行一个函数
- 包含核心逻辑的实际 .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) 的组合。
您当然可以以完全不同的方式实现更新检查。
关于python - 动态重新加载 Cython 模块,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/59906132/