Python3之harbor sdk api

Posted 2021-02-05 vglede

环境python3

源代码：

# coding=utf8
# Autor : Alaways V
# Time  : 2019/1/22 15:31
# File  : harbor.py
# Software PyCharm

import json
import urllib3
import requests
from pprint import pprint

urllib3.disable_warnings()

class HarborApi(object):
    def __init__(self, url, username, passwd, protocol="https"):
        \'\'\'
        init the request
        :param url: url address or doma
        :param username:
        :param passwd:
        :param protect:
        \'\'\'
        self.url = url
        self.username = username
        self.passwd =passwd
        self.protocol = protocol


    def login_get_session_id(self):
        \'\'\'
        by the login api to get the session of id
        :return:
        \'\'\'
        harbor_version_url = "%s://%s/api/systeminfo"%(self.protocol, self.url)
        header_dict = {
            \'User-Agent\': \'Mozilla/5.0 (Windows NT 6.1; Win64; x64; rv:64.0) Gecko/20100101 Firefox/64.0\', \\
            \'Content-Type\': \'application/x-www-form-urlencoded\'
        }
        data_dict = {
            "principal": self.username,
            "password": self.passwd
        }
        v_req_handle = requests.get(harbor_version_url, verify=False)
        self.harbor_version = v_req_handle.json()["harbor_version"]
        if self.harbor_version.startswith("v1.4"):
            req_url = "%s://%s/login" % (self.protocol, self.url)
            self.session_id_key = \'beegosessionID\'
        elif self.harbor_version.startswith("v1.7"):
            req_url = "%s://%s/c/login" % (self.protocol, self.url)
            self.session_id_key = "sid"
        else:
            raise ConnectionError("the %s version is not to supply!"%self.harbor_version)
        req_handle = requests.post(req_url, data=data_dict, headers=header_dict, verify=False)
        if 200 == req_handle.status_code:
            self.session_id = req_handle.cookies.get(self.session_id_key)
            return self.session_id
        else:
            raise Exception("login error,please check your account info!"+ self.harbor_version)


    def logout(self):
        requests.get(\'%s://%s/logout\' %(self.protocol, self.url),
                     cookies={self.session_id_key: self.session_id})
        raise Exception("successfully logout")

    def project_info(self):
        project_url = "%s://%s/api/projects" %(self.protocol, self.url)
        req_handle = requests.get(project_url, cookies={self.session_id_key: self.session_id}, verify=False)
        if 200 == req_handle.status_code:
            return req_handle.json()
        else:
            raise Exception("Failed to get the project info。")

    def repository_info(self, project_id):
        repository_url = \'%s://%s/api/repositories?project_id=%s\' %(self.protocol, self.url, project_id)
        req_handle = requests.get(repository_url, cookies={self.session_id_key: self.session_id}, verify=False)
        if 200 == req_handle.status_code:
            return req_handle.json()
        else:
            raise Exception("Failed to get the repository info。")

    def tags_info(self, repository_name):
        tags_url = \'%s://%s/api/repositories/%s/tags\' %(self.protocol, self.url, repository_name)
        req_handle = requests.get(tags_url, cookies={self.session_id_key: self.session_id}, verify=False)
        if 200 == req_handle.status_code:
            return req_handle.json()
        else:
            raise Exception("Failed to get the tags info。")

• 获取harbor api对象
  • 　　harbor_api = HarborApi("opsharbor.pay.net", "admin", "Harbor12345")
• 登录并获取session_id：
  • 　　print(harbor_api.login_get_session_id())

 

• 查询项目列表
  • 　　print(harbor_api.login_get_session_id())
  • 　　pprint(harbor_api.project_info())

 

• 查询镜像仓库
  • 　　print(harbor_api.login_get_session_id())
  • 　　pprint(harbor_api.repository_info(3))

• 查询镜像列表
  • 　print(harbor_api.login_get_session_id())
  • 　pprint(harbor_api.tags_info("c-front/c-front"))