Awesome, use Python to automate service deployment!

When deploying front-end projects recently, you need to first upload the front-end project compressed package to the /tmp directory of the application server through the bastion machine, then enter the application server and use the mv command to Move the compressed file to the Nginx project setting directory, and finally use the unzip command to decompress the file to complete the project deployment.

A careful analysis shows that most operations are repetitive actions, and completing these operations manually will greatly reduce work efficiency.

This article will introduce how to use Python to monitor folders to assist in completing service deployment.

1. Preparation

Here we will introduce a Python dependent library "watchdog"

It can be used to monitor files in a certain file directory File changes include operations such as deletion, modification, and addition. Each operation will call back an event function. We can write custom logic internally to meet our needs.

# 安装依赖包
pip3 install watchdog

Project address:

​https://www.php.cn/link/47841cc9e552bd5c40164db7073b817b​​

2. Let’s do it in practice

First, we need to create a listener to listen to the folder directory:

from watchdog.observers import Observer
...
# 创建一个监听器，用来监听文件夹目录
observer = Observer()
...

Then, create 2 event processing objects.

PS: This object inherits from the "FileSystemEventHandler" class.

They are used to monitor the "/tmp" directory and the "/home/project/frontend" directory respectively. Assume that the event objects are named obj1 and obj2

obj1 is responsible Monitor the /tmp directory, rewrite the "New or Modified" event method, and complete the moving operation of compressed files:

from watchdog.events import *
import ntpath
import shutil
import zipfile
def get_filename(filepath):
 """
 根据文件夹目录，获取文件名称（待后缀）
 :param filepath:
 :return:
 """
 return ntpath.basename(filepath)
class FileMoveHandler(FileSystemEventHandler):
 def __init__(self):
 FileSystemEventHandler.__init__(self)
 ...
 # 文件新建
 def on_created(self, event):
 # 新建文件夹
 if event.is_directory:
 # print("directory created:{0}".format(event.src_path))
 pass
 # 新建文件
 else:
 # print("file created:{0}".format(event.src_path))
 filename = get_filename(event.src_path)
 # 如果属于前端的4个项目压缩包，开始文件夹的操作
 if filename in watch_tags:
 self.start(filename)
...
def on_modified(self, event):
 if event.is_directory:
 # print("directory modified:{0}".format(event.src_path))
 pass
 else:
 # print("file modified:{0}".format(event.src_path))
 filename = get_filename(event.src_path)
 if filename in watch_tags:
 self.start(filename)
 ...
 def start(self, filename):
 """
 文件处理逻辑
 :param filename:
 :return:
 """
 try:
 # 文件名不带后缀
 filename_without_suffix = filename.split(".")[0]
 # 源文件路径（压缩包文件）
 source_file_path = watch_folder + filename
 # 目标文件路径（压缩包文件）
 target_file_path = target_folder + filename
 # 目标项目文件夹（目标项目）
 target_project_path = target_folder + filename_without_suffix
 # 1、复制文件到目标文件夹
 print(f"拷贝源目录{source_file_path},目标文件夹:{target_folder}")
 # 删除目标文件夹下的压缩文件
 if os.path.exists(target_file_path):
 os.remove(target_file_path)
 # 移动文件到目标文件夹中
 shutil.move(source_file_path, target_folder)
 # 2、清空目标文件夹中内的所有文件夹（如果存在）
 # 如果不存在，新建一个文件夹
 if os.path.exists(target_project_path):
 shutil.rmtree(target_project_path, ignore_errors=True)
 print(f"项目{filename_without_suffix}移动成功！")
 except Exception as e:
 print("部署失败，错误原因:", str(e.args))

obj2 is responsible for monitoring the /home/project/frontend directory, and also rewrite the "New or Modified" event method. , complete the decompression action of the compressed file:

...
 def start(self, filename):
 # 文件名不带后缀
 filename_without_suffix = filename.split(".")[0]
 # 目标文件路径（压缩包文件）
 target_file_path = target_folder + filename
 # 目标项目文件夹（目标项目）
 target_project_path = target_folder + filename_without_suffix
 r = zipfile.is_zipfile(target_file_path)
 if r:
 fz = zipfile.ZipFile(target_file_path, 'r')
 for file in fz.namelist():
 fz.extract(file, target_folder)
 else:
 print('这不是一个正常的zip压缩包！')
...

Then, start the listening tasks of the above two events through the listener:

import time
...
if __name__ == "__main__":
 # 待监听的文件夹目录
 watch_folder = "/tmp/"
 # 项目目标文件夹目录
 target_folder = "/home/project/frontend/"
 # 监听文件夹名称，即：项目压缩包名称
 watch_tags = ['proj1.zip', 'proj2.zip', 'proj3.zip', 'proj4.zip']
 # 创建一个监听器，用来监听文件夹目录
 observer = Observer()
 # 创建两个事件处理对象
 move_handler = FileMoveHandler()
 unzip_handler = FileUnzipHandler()
 # 启动监控任务
 # 参数分别是：观察者、监听目录、是否监听子目录
 observer.schedule(move_handler, watch_folder, True)
 observer.schedule(unzip_handler, target_folder, True)
 observer.start()
 try:
 while True:
 time.sleep(1)
 except KeyboardInterrupt:
 observer.stop()
 observer.join()
...

Finally, we use the "nohup" command on the server to let the file The listening program can be run in the background.

# 在后台运行
# 项目文件：watch_folder.py
# 日志文件：watch_folder.log
nohup python3 -u watch_folder.py > watch_folder.log 2>&1 &
# 查看日志：
cat watch_folder.log

3. Summary

Through the above operations, every time I upload the front-end zip compressed project file to the /tmp directory of the application server through the bastion machine, the program will automatically proceed The following operations automatically complete application deployment.

I have uploaded the source code in the article to the backend, and reply with the keyword "watchdog" to obtain the complete source code.

If you think the article is good, please like and share it, because this will be the strongest motivation for me to continue to output more high-quality articles!

The above is the detailed content of Awesome, use Python to automate service deployment!. For more information, please follow other related articles on the PHP Chinese website!