有一个python项目是用poetry管理依赖的 在构建docker镜像部署的时候也是poetry去管理依赖 但是现在部分业务切到函数计算来用
函数计算则是要给一个函数配置对应的层 所以在开始都是手动去更新层信息 在CI/CD自动话之前 想要一个脚本来自动更新层信息因为会涉及很多正则操作 所以打算用python来做脚本文件。 python 做脚本文件有一个很好用的库 fire
监控最新的依赖
依赖信息都安装在 backend/pyproject.toml 且规定用一个注释 # fc build start来作为分割 标记最新的依赖有哪些
...
alibabacloud-darabonba-stream = "^0.0.1"
aliyun-python-sdk-sts = "^3.1.2"
# fc build start
fire = "^0.5.0"通过正则和字符串来匹配最新的依赖信息。最新的库和版本号
def list_new_packages():
"""找到最新安装的依赖."""
packages = []
with open("backend/pyproject.toml") as f:
lines = f.readlines()
new_package_flag = False
for line in lines:
if not line.strip():
continue
if "[tool.poetry.dev-dependencies]" in line:
break
if new_package_flag:
if line.startswith("#"):
continue
if "#" in line:
line = line.split("#")[0]
package = ""
version = ""
extras = ""
result = tuple(line.split("=", 1))
package, description = result
if description.strip().startswith("{"):
description = description.replace("extras", '"extras"')
description = description.replace("version", '"version"')
description = description.replace("python", '"python"')
description = description.replace("=", ":")
print(description)
infos = json.loads(description)
version = infos.get("version", "").strip("^")
extras = infos.get("extras")
new_package = (
(f"{package.strip()}[{'|'.join(extras)}]=={version}")
if extras
else f"{package.strip()}=={version}"
)
packages.append(new_package)
else:
version = description.strip().strip('"').strip("^")
new_package = f"{package.strip()}=={version}"
packages.append(new_package)
if "# fc build start" in line:
new_package_flag = True
stdout("以下为新增依赖")
stdout(">>>>>>>>>>>>", StdoutType.INFO)
for package in packages:
stdout(f"{package}")
stdout(">>>>>>>>>>>>", StdoutType.INFO)
将最新的依赖 追加写入 requirements.txt
import json
import os
import subprocess
import sys
from enum import Enum
import fire
class StdoutType(str, Enum):
DEFAULT = "default"
INFO = "info"
ERROR = "error"
WARNING = "warning"
DEBUG = "debug"
def list_new_packages() -> list[str]:
"""找到最新安装的依赖."""
packages = []
with open("backend/pyproject.toml") as f:
lines = f.readlines()
new_package_flag = False
for line in lines:
if not line.strip():
continue
if "[tool.poetry.dev-dependencies]" in line:
break
if new_package_flag:
if line.startswith("#"):
continue
if "#" in line:
line = line.split("#")[0]
package = ""
version = ""
extras = ""
result = tuple(line.split("=", 1))
package, description = result
if description.strip().startswith("{"):
description = description.replace("extras", '"extras"')
description = description.replace("version", '"version"')
description = description.replace("python", '"python"')
description = description.replace("=", ":")
print(description)
infos = json.loads(description)
version = infos.get("version", "").strip("^")
extras = infos.get("extras")
new_package = (
(f"{package.strip()}[{'|'.join(extras)}]=={version}")
if extras
else f"{package.strip()}=={version}"
)
packages.append(new_package)
else:
version = description.strip().strip('"').strip("^")
new_package = f"{package.strip()}=={version}"
packages.append(new_package)
if "# fc build start" in line:
new_package_flag = True
stdout("以下为新增依赖")
stdout(">>>>>>>>>>>>", StdoutType.DEBUG)
for package in packages:
stdout(package)
stdout(">>>>>>>>>>>>", StdoutType.DEBUG)
return packages
def write_requirements(packages: list[str]):
"""将新的依赖写入 requirements."""
with open("backend/requirements.txt", "a") as f:
f.writelines([f"{package}\n" for package in packages])
def stdout(info: str, out_type: StdoutType = StdoutType.DEFAULT):
match out_type:
case StdoutType.DEFAULT:
pass
case StdoutType.ERROR:
info = "\033[91m" + info + "\033[0m"
case StdoutType.INFO:
info = "\033[92m" + info + "\033[0m"
case StdoutType.WARNING:
info = "\033[93m" + info + "\033[0m"
case StdoutType.DEBUG:
info = "\033[94m" + info + "\033[0m"
sys.stdout.write(info)
sys.stdout.write(os.linesep)
sys.stdout.flush()
class FcLayerUpdater:
def update(self):
try:
stdout("step1 -> 匹配最新的依赖包", StdoutType.INFO)
packages = list_new_packages()
stdout("step2 -> 将新的依赖写入 requirements", StdoutType.INFO)
write_requirements(packages)
stdout("step3 -> 安装requirements", StdoutType.INFO)
subprocess.Popen(
"pip install -r backend/requirements.txt -t ./fc_layer", shell=True
).wait()
stdout("step4 -> 将文件夹打包为 zip", StdoutType.INFO)
os.system("sudo apt-get install zip")
subprocess.Popen("zip -r fc_layer.zip fc_layer", shell=True).wait()
stdout("step5 -> 上传 zip 到 OSS", StdoutType.INFO)
stdout("step6 -> 通过OSS更新层", StdoutType.INFO)
except Exception as e:
stdout(f"更新失败. 错误原因 {e.__str__()}", StdoutType.ERROR)
sys.exit(1)
except KeyboardInterrupt:
stdout("更新停止", StdoutType.ERROR)
sys.exit(1)
if __name__ == "__main__":
try:
fire.Fire(FcLayerUpdater)
except SystemExit as e:
if e.code != 0:
sys.exit(e.code)
… 没时间接着写了 先撸代码 有缘再见
