Python mysql *** 作 执行sql语句插入查询等

Python mysql *** 作 执行sql语句插入查询等

啊直接上代码了。

这个代码我用来做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)

# 提交到数据库执行

            self.conn.commit()

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)))