首页 > 编程语言 >harbor命令行-python脚本-调用http API

harbor命令行-python脚本-调用http API

时间：2024-03-17 14:11:06浏览次数：23  
标签：http headers harbor format auth python json data response

因为生产环境没有UI界面，也没有开放端口资源出来，无法通过web界面来管理harbor，故调用harbor提供的httpAPI来获取harbor信息和使用常用的功能。

import requests
import json
import urllib.parse

my_auth=('admin', 'Lovedan@971220')
harbor_addr="https://127.0.0.1"

def 查询项目信息():
    headers = {'Content-Type': 'application/json',}
    response = requests.get('{}/api/v2.0/projects'.format(harbor_addr), headers=headers, verify=False, auth=my_auth)
    response_json = json.loads(response.text)
    for data in response_json:
        print("项目名：{}\t项目ID：{}\t创建时间：{}\t是否公开：{}".format(data['name'],data['project_id'],data['creation_time'],data['metadata']['public']));
def 新增项目(项目名:str,公开:bool,配额:int):
    '''
    配额：项目可以使用的最大空间，若不想限制，请传递-1
    '''
    headers = {'Content-Type': 'application/json',}
    json_data={"project_name": 项目名,
  "public": 公开,
  "metadata": {
    "public": str(公开).lower(),
    "enable_content_trust": "false",
    "enable_content_trust_cosign": "false",
    "prevent_vul": "false",
    "auto_scan": "false",
    "reuse_sys_cve_allowlist": "false"
  },
  "storage_limit": -1,}
    response = requests.post('{}/api/v2.0/projects'.format(harbor_addr),headers=headers,json=json_data,verify=False,auth=my_auth,)
    if(200 == response.status_code or 201 == response.status_code):
        return True
    else:
        response_json = json.loads(response.text)
        print(response_json['errors'][0]["message"])
        return False;
def 删除项目(项目ID:int)->bool:
    headers = {'Content-Type': 'application/json',}
    response = requests.delete('{}/api/v2.0/projects/{}'.format(harbor_addr,项目ID), headers=headers, verify=False, auth=my_auth)
    if(200 == response.status_code):
        return True
    else:
        response_json = json.loads(response.text)
        print(response_json['errors'][0]["message"])
        return False;
def 获取项目摘要信息(项目id:int):
    headers = {'Content-Type': 'application/json',}
    response = requests.get('{}/api/v2.0/projects/{}/summary'.format(harbor_addr,项目id), headers=headers, verify=False, auth=my_auth)
    data = json.loads(response.text)
    print("镜像数量：{}\n配额大小：{}\n已使用大小：{}".format(data['repo_count'],data["quota"]["hard"]["storage"],友好显示大小(data["quota"]["used"]["storage"])))
def 获取项目的日志(project_name:str,page:int=1,num:int=100):
    '''
    默认获取最新的一百条日志
    日志ID  用户名称    资源名称    标记    操作    时间
    '''
    headers = {'Content-Type': 'application/json',}
    response = requests.get('{}/api/v2.0/projects/{}/logs'.format(harbor_addr,project_name), headers=headers, verify=False, auth=my_auth)
    response_json = json.loads(response.text)
    for data in response_json:
        print("{}\t{}\t{}\t{}\t{}\t{}".format(data['id'],data['username'],data['resource'],data['resource_type'],data['operation'],data['op_time']))
def 获取项目成员信息(项目id:int):
    headers = {'Content-Type': 'application/json',}
    response = requests.get('{}/api/v2.0/projects/{}/members'.format(harbor_addr,项目id), headers=headers, verify=False, auth=my_auth)
    response_json = json.loads(response.text)
    for data in response_json:
        print("项目成员ID：{}\t项目ID：{}\t账号名称：{}\t角色：{}".format(data['id'],data['project_id'],data['entity_name'],data['role_name']))
def 查询harbor健康度()->tuple:
    '''
    7个组件，按照百分比进行计算，返回元组，第一个元素表示健康百分比。第二个元素是详细健康值
    '''
    headers = {'Content-Type': 'application/json',}
    response = requests.get('{}/api/v2.0/health'.format(harbor_addr), headers=headers, verify=False, auth=my_auth)
    response_json = json.loads(response.text)
    #0表示不健康，1表示健康
    status_dict={"core":0,"database":0,"jobservice":0,"portal":0,"redis":0,"registry":0,
    "registryctl":0,}
    healthy_num=0
    for data in response_json['components']:
        if("healthy" == data['status']):
            healthy_num+=1
            status_dict[data['name']]=1
    healthy=healthy_num/len(status_dict)
    print('健康度：{:.2%}'.format(healthy))
    return (healthy,status_dict);
def 查询普通用户列表():
    headers = {'Content-Type': 'application/json',}
    response = requests.get('{}/api/v2.0/users'.format(harbor_addr), headers=headers, verify=False, auth=my_auth)
    response_json = json.loads(response.text)
    for data in response_json:
        print("账户名称：{}\t用户全名：{}\t创建时间：{}\t邮箱：{}\t用户ID：{}\t管理员:{}\t备注:{}".format(data['username'],data['realname'],data['creation_time'],data['email'],data['user_id'],data['sysadmin_flag'],data['comment']))
def 创建普通用户(用户信息:dict)->bool:
    '''
    用户信息 = {账号名称:"",密码:"",用户全名:"",邮箱:"",备注:""}
    '''
    json_data = {
    'email': 用户信息['邮箱'],
    'realname': 用户信息['用户全名'],
    'comment': 用户信息['备注'],
    'password': 用户信息['密码'],
    'username': 用户信息['账号名称'],
    }
    headers = {
    'Content-Type': 'application/json',
    }
    response = requests.post('{}/api/v2.0/users'.format(harbor_addr),headers=headers,json=json_data,verify=False,auth=my_auth,)
    if(200 == response.status_code or 201 == response.status_code):
        return True
    else:
        response_json = json.loads(response.text)
        print(response_json['errors'][0]["message"])
        return False;
def 删除普通用户(账号id:int)->bool:
    headers = {'Content-Type': 'application/json',}
    response = requests.delete('{}/api/v2.0/users/{}'.format(harbor_addr,账号id), headers=headers, verify=False, auth=my_auth)
    if(200 == response.status_code):
        return True
    else:
        response_json = json.loads(response.text)
        print(response_json['errors'][0]["message"])
        return False;
def 设置用户管理权限(user_id:int,sysadmin_flag:bool):
    '''
    sysadmin_flag：将用户切换为管理员或不切换为管理员
    '''
    json_data = {"sysadmin_flag": sysadmin_flag}
    headers = {'Content-Type': 'application/json',}
    response = requests.put('{}/api/v2.0/users/{}/sysadmin'.format(harbor_addr,user_id),headers=headers,json=json_data,verify=False,auth=my_auth,)
    if(200 == response.status_code or 201 == response.status_code):
        return True
    else:
        response_json = json.loads(response.text)
        print(response_json['errors'][0]["message"])
        return False;
def 修改用户密码(user_id:int,pw:str):
    json_data = {"new_password": pw}
    headers = {'Content-Type': 'application/json',}
    response = requests.put('{}/api/v2.0/users/{}/password'.format(harbor_addr,user_id),headers=headers,json=json_data,verify=False,auth=my_auth,)
    if(200 == response.status_code or 201 == response.status_code):
        return True
    else:
        response_json = json.loads(response.text)
        print(response_json['errors'][0]["message"])
        return False;
def 项目和存储库的统计():
    headers = {'Content-Type': 'application/json',}
    response = requests.get('{}/api/v2.0/statistics'.format(harbor_addr), headers=headers, verify=False, auth=my_auth)
    response_json = json.loads(response.text)
    print("私有项目数量：{}\n私有镜像数量：{}\n公开项目数量：{}\n公开镜像数量：{}\n项目总计数量：{}\n镜像总计数量：{}\n已用存储：{}\n".format(response_json['private_project_count'],response_json['private_repo_count'],response_json['public_project_count'],response_json['public_repo_count'],response_json['total_project_count'],response_json['total_repo_count'],友好显示大小(response_json['total_storage_consumption'])))
def 友好显示大小(size:int)->str:
    '''
    给定正整数，默认以字节为单位，来计算最佳的单位显示，返回的结果中仅保留两位小数
    '''
    if(size<1024):
        return "{}B".format(size)
    elif(size<1048576):
        return "{:.2f}KB".format(size/1024)
    elif(size<1073741824):
        return "{:.2f}MB".format(size/1024/1024)
    elif(size<1099511627776):
        return "{:.2f}GB".format(size/1024/1024/1024)
    elif(size<(1024*1024*1024*1024*1024)):
        return "{:.2f}TB".format(size/1024/1024/1024/1024)
    else:
        return "{}B".format(size)
def 获取项目下的镜像列表(项目名称:str):
    headers = {'Content-Type': 'application/json',}
    response = requests.get('{}/api/v2.0/projects/{}/repositories'.format(harbor_addr,项目名称), headers=headers, verify=False, auth=my_auth)
    response_json = json.loads(response.text)
    for data in response_json:
        print('镜像ID：{}\n项目ID：{}\n镜像名称：{}\n拉取次数：{}\n创建时间：{}\n更新时间：{}\n版本数量：{}\n'.format(data['id'],data['project_id'],data['name'],data['pull_count'],data['creation_time'],data['update_time'],data['artifact_count']))
def 获取镜像详情(项目名称:str,镜像名称:str,是否输出镜像元数据:bool=False,输出路径:str=None):
    '''
    镜像名称中，不需要包含项目名前缀，比如说通过函数《获取项目下的镜像列表》输出的镜像名称中就带有项目名前缀，本函数中不需要此前缀。否则将会获取不到结果。
    倘若函数《获取项目下的镜像列表》输出的镜像名称为【k8s/kube-apiserver】，那么应该本函数的形参【镜像名称】值应该为kube-apiserver
    示例调用：获取镜像详情('k8s','mysql',True,r'/root/mysql_docker_元数据.txt')
    '''
    if('/' in 镜像名称):
        镜像名称 = urllib.parse.quote(urllib.parse.quote(镜像名称,'utf-8'),'utf-8')
    headers = {'Content-Type': 'application/json',}
    response = requests.get('{}/api/v2.0/projects/{}/repositories/{}/artifacts'.format(harbor_addr,项目名称,镜像名称), headers=headers, verify=False, auth=my_auth)
    response_json = json.loads(response.text)
    total_size=0
    for data in response_json:
        print("摘要ID：{}\n标签：{}\n大小：{}".format(data['digest'],data['tags'][0]['name'],友好显示大小(data['size'])))
        total_size+=data['size']
    print('总大小：{}'.format(友好显示大小(total_size)))
    if(是否输出镜像元数据):
        with open(输出路径,'w',encoding='utf-8')as out:
            for data in response_json:
                tag = data['tags'][0]['name']
                str_=json.dumps(data['extra_attrs'],indent='\t')
                print("{}:{}\n{}".format(镜像名称,tag,str_),file=out);
def 删除特定版本的镜像(项目名称:str,镜像名称:str,摘要ID:str=None,标签:str=None):
    '''
    摘要ID、标签 可以根据函数《获取镜像详情》获取得到
    摘要id和标签二者传递一个即可，若是都传递，优先使用摘要ID
    '''
    if('/' in 镜像名称):
        镜像名称 = urllib.parse.quote(urllib.parse.quote(镜像名称,'utf-8'),'utf-8')
    headers = {'Content-Type': 'application/json',}
    if None != 摘要ID:
        reference=摘要ID
    else:
        reference=标签
    response = requests.delete('{}/api/v2.0/projects/{}/repositories/{}/artifacts/{}'.format(harbor_addr,项目名称,镜像名称,reference), headers=headers, verify=False, auth=my_auth)
    if(200 == response.status_code):
        return True
    else:
        response_json = json.loads(response.text)
        print(response_json['errors'][0]["message"])
        return False;
def 删除镜像(项目名称:str,镜像名称:str):
    '''
    删除给定名称的镜像，将该镜像的所有版本都一并删除
    '''
    if('/' in 镜像名称):
        镜像名称 = urllib.parse.quote(urllib.parse.quote(镜像名称,'utf-8'),'utf-8')
    headers = {'Content-Type': 'application/json',}
    response = requests.delete('{}/api/v2.0/projects/{}/repositories/{}'.format(harbor_addr,项目名称,镜像名称), headers=headers, verify=False, auth=my_auth)
    if(200 == response.status_code):
        return True
    else:
        response_json = json.loads(response.text)
        print(response_json['errors'][0]["message"])
        return False;
def 获取系统磁盘信息():
    '''此函数输出的时整个主机的harbor所在磁盘分区使用情况，并非表示harbor的使用情况'''
    headers = {'Content-Type': 'application/json',}
    response = requests.get('{}/api/v2.0/systeminfo/volumes'.format(harbor_addr), headers=headers, verify=False, auth=my_auth)
    response_json = json.loads(response.text)
    total = response_json['storage'][0]["total"]
    free=response_json['storage'][0]["free"]
    print("总的可用大小:{}\n已使用大小：{}\n使用率：{:.2%}".format(友好显示大小(total),友好显示大小(total-free),(total-free)/total))
def 获取常规系统信息():
    headers = {'Content-Type': 'application/json',}
    response = requests.get('{}/api/v2.0/systeminfo'.format(harbor_addr), headers=headers, verify=False, auth=my_auth)
    response_json = json.loads(response.text)
    print(json.dumps(response_json,indent='\t'))
    
print(查询harbor健康度())
# 查询项目信息()
# 创建普通用户({'邮箱':'[email protected]','密码':"Lovedan@971220",'用户全名':'test','账号名称':'test','备注':'ceshi'})
View Code

 

标签：http,headers,harbor,format,auth,python,json,data,response
From： https://www.cnblogs.com/love-DanDan/p/18078519

相关文章

  • 全栈的自我修养 ———— python爬虫爬取斗破苍穹小说并保存到文档里！看完保证会灵活使
    如果赶时间请直接看第四大步中的源码，好好看完这篇博客，这里保证大家以后彻底搞懂获得一个网站怎么来爬取！！！！一、准备二、选入合适的爬取的目标1、如何看出网站做了反爬2、合适的网站三、理思路1、选择合适的页面2、选择合适的元素三、爬取1、获取所有章节的链接获取网站上......
  • Dynamo PythonScript 代码速查手册By九哥
    你好，这里是BIM的乐趣，我是九哥今天给大家带来的是我的知识工程的第二套知识库，这套知识库不是教程，是一套完整的笔记，里面包含了大量的Python代码。当然这里也没有实现太多的具体逻辑，单纯的是通过Python在Dynamo中实现一些RevitAPI的方法，我的目的就是省事，写代码的时......
  • AI回答记录：python中有重载与重写的概念吗？
    问：python中有重载与重写的概念吗？GPT4-turbo:在Python中，严格来说没有像其他一些编程语言（如Java、C++）那样的重载（Overloading）机制，但是可以通过其他方式实现类似的功能。Python中有重写（Overriding）的概念。重载（Overloading）：Python不支持传统的方法重载，即在同一个类中定义多个同名......
  • Python实战：爬取小红书
    有读者在公众号后台询问爬取小红书，今天他来了。本文可以根据关键词，在小红书搜索相关笔记，并保存为excel表格。爬取的字段包括笔记标题、作者、笔记链接、作者主页地址、作者头像、点赞量。一、先看效果1、爬取搜索页2、爬取结果保存到本地excel表格运行我写的爬虫，......
  • Python中有重写和重载的方法吗？
    问1：Python中有重写和重载的方法吗？先说结论：Python因为语言的特性，没有重载，但是有重写的方法。问2：什么是重写？1.先看代码：classCalculator:defadd(self,a,b):print("我执行了吗？")returna+b+100defadd(self,a,b):returnflo......
  • python自动化——Requests——教程截图笔记复习
                                                                           123123......
  • Python基本语法
    Python基本语法Python是一种简单易学、功能强大的编程语言，其基本语法涵盖了变量、数据类型、运算符等方面。本文将介绍Python中的基本语法要点，帮助初学者快速入门。变量在Python中，变量是用来存储数据的容器。定义变量时不需要指定数据类型，Python会根据赋给变量的值自动确......
  • Python中的模块管理：提高代码效率和可维护性
    在编程中，随着项目的增长，维护和管理代码变得越来越复杂。Python通过模块化来解决这个问题，使得代码更加有序，易于理解和维护。本文将探讨Python中模块的概念，如何用自定义模块管理函数，以及处理命名冲突的方法。模块的概念模块是Python程序结构的基本组成部分，它帮助我们将代码分割......
  • python疑难杂症（9）---python的数据类型字典（dict）的创建、访问、修改、删除等方法汇总
    在Python中，字典（Dictionary）是一种内置的数据烈性，是无序的数据结构，用于存储键值对（key-value）。字典中的每个元素由一个键（key）和一个对应的值（value）组成，键和值之间使用冒号（:）进行分隔，每个键值对之间使用逗号（,）进行分隔。字典中的键必须是唯一的，而值可以是任意类型的对象，字典可以用来存......
  • 基于Django旅游景区景点订票系统设计与实现（Pycharm+Python+Mysql）
     博主介绍：黄菊华老师《Vue.js入门与商城开发实战》《微信小程序商城开发》图书作者，CSDN博客专家，在线教育专家，CSDN钻石讲师；专注大学生毕业设计教育和辅导。所有项目都配有从入门到精通的基础知识视频课程，学习后应对毕业设计答辩。项目配有对应开发文档、开题报告、任务书、P......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99