使用版本：python3.9，centos7.8

python里提供了os模块，可以让我在代码中对linux发送命令，并获取返回值。

这里会讲到两种方法去使用os模块----os.popen和os.system

一.  os.system

os.system('命令字符串')。这个命令适用于一些你只需要知道结果是否成功的命令，因为它只会返回给你一个int类型的数字，0代表成功，非0代表失败。而且当你的命令字符串是一个子进程命令的话，他的返回值并不准确，他不会等待子进程在后台运行结束，只要子进程能够进入运行状态，他就会返回0。

二.  os.popen

在我看来，这个方法在使用子进程做并行或者在获取返回结果上，是很好用的，首先它一共有3个参数，

cmd--命令

mode--方式（默认值为‘r’,可选值为‘w’,一般场景中，r的使用比较普遍）

buffering--缓存值，默认为-1

该方法调用了subprocess里的方法，并最后帮助你关闭了资源，一条龙服务还是很人性化的。所以一般使用时，传一个命令即可，然后它提供了read(),readlines()给你去拿到返回值

即os.popen('cmd').read()

而且笔者认为它强大之处在于，当cmd是一个后台进程，即在命令最后加入&符号的进程，它依旧会等待进程结束并拿到返回值，我也用他帮我将原本需要5分钟+的程序中，加上使用子进程并行将时间缩短到20秒左右。

话不多说，分享实战经验，一个读取jenkins信息并写入时序数据库的流程：

获取相关信息，并发送请求方法：

        if action == 'jenkins':# 获取主机地址host = request.get_host()response_data = {'kind': 'JenkinsInfo.get', 'msg': 'jenkins任务构建信息入库', 'status': 200, 'param': ''}data = request.GETjenkins_name = data.get('jenkins_name', '')if jenkins_name in JENKINS.keys():jenkins_info = JENKINS[jenkins_name]else:response_data.update({'param': '参数无效','status': 403})return JsonResponse(response_data, status=response_data['status'])# 连接jenkinsserver = self.to_jenkins(jenkins_info)all_jobs = server.get_jobs()job_name_list = [x.get('name') for x in all_jobs]cmdb_job = {}# 获取cmdb中的app_namecmdb_apps = AppBasisNew.objects.filter(**{'status': '已上线'}).values('app_name')app_name_list = [x.get('app_name') for x in cmdb_apps]# 判断cmdb中app是否在jenkins中env_list = ['archive-PRE-', 'PRE-', 'bak-PRE-', 'ARCHIVE-PRE-', 'BAK-PRE-']if app_name_list and job_name_list:for app_name in app_name_list:for env in env_list:if env+app_name in job_name_list:cmdb_job[env+app_name] = app_nameif cmdb_job:job_str = str(cmdb_job).replace("{'", "").replace("'}", "").replace("', '", " ").replace("': '", "\|")res = os.popen(f"for job in {job_str}; do curl -G http://{host}/jenkins/pre/to_db?"f"jenkins_name={jenkins_name}\&job_name=$job & done").read()res_str = res.replace('}{', '}, {').join('[]')res_json = json.loads(res_str)response_data.update({'param': res_json})else:response_data.update({'param': 'jenkins与cmdb无匹配数据'})return JsonResponse(response_data, status=response_data['status'])

接受请求，返回相应：

        if action == 'to_db':data = request.GETjenkins_name = data['jenkins_name']job_name = data['job_name'].split('|')[0]project_name = data['job_name'].split('|')[1]server = self.to_jenkins(JENKINS[jenkins_name])build_info_list = []# 最新jenkins项目的构建时间last_timestamp = 0# 连接influxdbdb_client = self.to_influxdb()try:job_info = server.get_job_info(job_name)except jenkins.JenkinsException:db_client.close()return JsonResponse({f'{jenkins_name}': f'{job_name} fail'})if job_info.get('builds'):# 获取该工程名每次构筑的时间：res = db_client.query(f"select max(build_time) from pre_job_info "f"where project_name::tag='{project_name}' and from_jenkins='{jenkins_name}'")last_time = 0if res:last_time = [x[u'max'] for x in res.get_points()][0]for job_build in range(1, len(job_info['builds'])+1):job_time = job_info['builds'][-job_build]['number']try:build_info = server.get_build_info(job_name, job_time)except jenkins.JenkinsException:db_client.close()return JsonResponse({f'{jenkins_name}': f'{job_name} fail'})if build_info.get('timestamp') // 1000 <= last_time:breakif job_build == 1:last_timestamp = build_info.get('timestamp')//1000build_info_dict = ToDict().return_dic(build_info)# 控制台输出build_output = server.get_build_console_output(job_name, job_time)build_out_list = build_output.split('\n')build_agent_name = 'null'project_path = 'null'if jenkins_name == 'JENKINS_PRE1' or jenkins_name == 'JENKINS_PRE3':res_info = ''for i in build_out_list:if 'Running on' in i:res_info = iif res_info:build_agent_name = res_info.split()[-3]project_path = res_info.split()[-1]if jenkins_name == 'JENKINS_PRE2':project_path = build_out_list[1].split()[-1]build_agent_name = build_info_dict.get('builtOn')if job_info.get('lastStableBuild'):last_stable_build = job_info['lastStableBuild']['number']else:last_stable_build = 0if job_info.get('lastSuccessfulBuild'):last_successful_build = job_info['lastSuccessfulBuild']['number']else:last_successful_build = 0if build_info_dict.get('actions_lastBuiltRevision_branch_name'):build_branch_name = build_info_dict.get('actions_lastBuiltRevision_branch_name')code_version = build_info_dict.get('actions_lastBuiltRevision_branch_SHA1')else:build_branch_name = build_info_dict.get('changeSet_revisions_module', 'null')code_version = str(build_info_dict.get('changeSet_revisions_revision', 'null'))if build_info_dict.get('result') == 'SUCCESS':build_successful = Trueelse:build_successful = Falseif build_info_dict.get('actions_parameters_name1') == 'Fabu_Type':fabu_type = build_info_dict.get('actions_parameters_value1')else:fabu_type = build_info_dict.get('actions_parameters_value2', 'all-in-one')res = {'measurement': 'pre_job_info','tags': {'project_name': project_name,'build_result': build_info_dict.get('result'),'project_path': project_path},'time': datetime.fromtimestamp(build_info_dict.get('timestamp') // 1000),'fields': {'build_agent_name': build_agent_name,'build_branch_name': build_branch_name,'build_causer': build_info_dict.get('actions_causes_userName'),'build_exec_time': build_info_dict.get('duration') // 1000,'build_measured_time': int(datetime.now().timestamp()),'build_number': build_info_dict.get('id'),'build_result': build_info_dict.get('result'),'build_result_ordinal': build_info_dict.get('queueId'),'build_scheduled_time': build_info_dict.get('estimatedDuration') // 1000,'build_status_message': build_info_dict.get('building'),'build_successful': build_successful,'build_time': build_info_dict.get('timestamp') // 1000,'last_stable_build': last_stable_build,'last_successful_build': last_successful_build,'project_build_health': job_info['healthReport'][0]['score'],'project_name': project_name,'project_path': project_path,'Code_Version': code_version,'Fabu_Type': fabu_type,'from_jenkins': jenkins_name},}build_info_list.append(res)if build_info_list:db_client.write_points(build_info_list)else:db_client.close()return JsonResponse({f'{jenkins_name}': f'{job_name} has no new data'})# 数据入库验证new_res = db_client.query(f"select build_number from pre_job_info "f"where project_name::tag='{project_name}' and build_time={last_timestamp}")if new_res:db_client.close()return JsonResponse({f'{jenkins_name}': f'{job_name} success'})db_client.close()return JsonResponse({f'{jenkins_name}': f'{job_name} fail'})