啊直接上代码了。
这个代码我用来做jenkins的构建记录的查询和写入的（在成功构建后，命令行执行的形式，执行该代码）
（需要项目源码加v：ct-xiaozhou;非机构，行业变迁，多个好友多点可能性）
import datetime
import sys
import pymysql
class TestMysqlDB:
test_host ='10.255.**.**'
test_port =3306
test_user ='root'
test_password ='123456'
test_db ='test_work'
def __init__(self, db=None):
self.conn = pymysql.connect(host=self.test_host,
port=self.test_port,
user=self.test_user,
password=self.test_password,
db=self.test_db,
charset='utf8'
)
# 使用cursor()方法获取操作游标
self.cursor =self.conn.cursor()
def getData(self, sqlstr):
# 返回一条数据
self.cursor.execute(sqlstr)
return self.cursor.fetchall()
def close(self):
self.cursor.close()
self.conn.close()
def insterJenkins_build_info(self, datalist:list):
# datalist值：1.服务名称；2.git信息，3.dockertag；4.压缩文件名；5.构建信息；6.构建人；
thTime = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
# 构建完成时间
sql =f"""INSERT INTO jenkins_build_info (servicename,gitinfo,dockertag,tarfilename,buildmsg,builduser,env,buildtime,updatetime) VALUES
('{datalist[0]}','{datalist[1]}','{datalist[2]}','{datalist[3]}','{datalist[4]}','{datalist[5]}','{datalist[6]}','{thTime}','{thTime}');"""
try:
# 执行sql语句
self.cursor.execute(sql)
# 提交到数据库执行
mit()
except Exception as e:
print(f'{e}')
self.conn.rollback()
# 关闭数据库连接
self.close()
def dateToStr(thTuple):
temstr =''
for datain thTuple:
temstr =f"{temstr}{str(data[0]).split(':')[1]}\n"
return temstr
if '__main__' == __name__:
"""
命令行执行，第一个参数是文件本身，第二个参数也就是sys.argv[1]定义为数据库操作类型，第三个参数是具体操作的值"""
if "newBuildIn" == sys.argv[1]:
# 1.服务名称|2.git信息|3.dockertag|4.压缩文件名|5.构建信息|6.构建人|7.环境
datalist = sys.argv[2].split('|')
TestMysqlDB().insterJenkins_build_info(datalist)
elif "getHistoryTag" == sys.argv[1]:
# 第一个参数是类型，第二个参数是服务名称
sqlstr =f"SELECT dockertag from jenkins_build_info jbi where servicename ='{sys.argv[2]}' and not(dockertag LIKE '历史版本%') order by updatetime DESC limit 10;"
print(dateToStr(TestMysqlDB().getData(sqlstr)))
