Upstream/harbor
1
0
Fork 0
mirror of https://github.com/goharbor/harbor.git synced 2024-12-22 16:48:30 +01:00
Code Issues Projects Releases Wiki Activity

Upgrade python SDK to v1.1.1

Browse Source 
Update Python SDK to v1.1.1

disable verify

remove unused
This commit is contained in:
wangyan 2017-05-04 02:38:46 -07:00
parent 13d9b05dcb
commit a21f321594
1 changed files with 174 additions and 84 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

256
contrib/sdk/harbor-py/harborclient/harborclient.py
View File

@ -4,8 +4,7 @@ import json
import logging import logging
import requests import requests
logging.basicConfig(level=logging.INFO) logging.basicConfig(level=logging.WARNING)
class HarborClient(object): class HarborClient(object):
def __init__(self, host, user, password, protocol="http"): def __init__(self, host, user, password, protocol="http"):
@ -14,55 +13,33 @@ class HarborClient(object):
self.password = password self.password = password
self.protocol = protocol self.protocol = protocol
self.session_id = self.login()
def __del__(self): def __del__(self):
self.logout() self.logout()
def login(self): def login(self):
login_data = requests.post('%s://%s/login' % login_data = requests.post('%s://%s/login' %(self.protocol, self.host),
(self.protocol, self.host),
data={'principal': self.user, data={'principal': self.user,
'password': self.password}) 'password': self.password}, verify=False)
if login_data.status_code == 200: if login_data.status_code == 200:
session_id = login_data.cookies.get('beegosessionID') session_id = login_data.cookies.get('beegosessionID')
self.session_id = session_id
logging.debug("Successfully login, session id: {}".format( logging.debug("Successfully login, session id: {}".format(
session_id)) session_id))
return session_id
else: else:
logging.error("Fail to login, please try again") logging.error("Fail to login, please try again")
return None
def logout(self): def logout(self):
requests.get('%s://%s/logout' % (self.protocol, self.host), requests.get('%s://%s/log_out' % (self.protocol, self.host),
cookies={'beegosessionID': self.session_id}) cookies={'beegosessionID': self.session_id}, verify=False)
logging.debug("Successfully logout") logging.debug("Successfully logout")
# Get project id
def get_project_id_from_name(self, project_name):
registry_data = requests.get(
'%s://%s/api/projects?project_name=%s' %
(self.protocol, self.host, project_name),
cookies={'beegosessionID': self.session_id})
if registry_data.status_code == 200 and registry_data.json():
project_id = registry_data.json()[0]['project_id']
logging.debug(
"Successfully get project id: {}, project name: {}".format(
project_id, project_name))
return project_id
else:
logging.error("Fail to get project id from project name",
project_name)
return None
# GET /search # GET /search
def search(self, query_string): def search(self, query_string):
result = None result = None
path = '%s://%s/api/search?q=%s' % (self.protocol, self.host, path = '%s://%s/api/search?q=%s' % (self.protocol, self.host,
query_string) query_string)
response = requests.get(path, response = requests.get(path,
cookies={'beegosessionID': self.session_id}) cookies={'beegosessionID': self.session_id}, verify=False)
if response.status_code == 200: if response.status_code == 200:
result = response.json() result = response.json()
logging.debug("Successfully get search result: {}".format(result)) logging.debug("Successfully get search result: {}".format(result))
@ -71,12 +48,11 @@ class HarborClient(object):
return result return result
# GET /projects # GET /projects
def get_projects(self, project_name=None, is_public=None): def get_projects(self):
# TODO: support parameter
result = None result = None
path = '%s://%s/api/projects' % (self.protocol, self.host) path = '%s://%s/api/projects' % (self.protocol, self.host)
response = requests.get(path, response = requests.get(path,
cookies={'beegosessionID': self.session_id}) cookies={'beegosessionID': self.session_id}, verify=False)
if response.status_code == 200: if response.status_code == 200:
result = response.json() result = response.json()
logging.debug("Successfully get projects result: {}".format( logging.debug("Successfully get projects result: {}".format(
@ -91,7 +67,7 @@ class HarborClient(object):
path = '%s://%s/api/projects?project_name=%s' % ( path = '%s://%s/api/projects?project_name=%s' % (
self.protocol, self.host, project_name) self.protocol, self.host, project_name)
response = requests.head(path, response = requests.head(path,
cookies={'beegosessionID': self.session_id}) cookies={'beegosessionID': self.session_id}, verify=False)
if response.status_code == 200: if response.status_code == 200:
result = True result = True
logging.debug( logging.debug(
@ -105,16 +81,15 @@ class HarborClient(object):
return result return result
# POST /projects # POST /projects
def create_project(self, project_name, is_public=False): def create_project(self, project_name, is_public=0):
result = False result = False
path = '%s://%s/api/projects' % (self.protocol, self.host) path = '%s://%s/api/projects' % (self.protocol, self.host)
request_body = json.dumps({'project_name': project_name, request_body = json.dumps({'project_name': project_name,
'public': is_public}) 'public': is_public})
response = requests.post(path, response = requests.post(path,
cookies={'beegosessionID': self.session_id}, cookies={'beegosessionID': self.session_id},
data=request_body) data=request_body, verify=False)
if response.status_code == 201 or response.status_code == 500: if response.status_code == 201:
# TODO: the response return 500 sometimes
result = True result = True
logging.debug( logging.debug(
"Successfully create project with project name: {}".format( "Successfully create project with project name: {}".format(
@ -125,15 +100,68 @@ class HarborClient(object):
project_name, response.status_code)) project_name, response.status_code))
return result return result
# GET /projects/{project_id}/members
def get_project_members(self, project_id):
result = None
path = '%s://%s/api/projects/%s/members' % (self.protocol, self.host, project_id)
response = requests.get(path,
cookies={'beegosessionID': self.session_id}, verify=False)
if response.status_code == 200:
result = response.json()
logging.debug(
"Successfully create project with project id: {}".format(
project_id))
else:
logging.error(
"Fail to create project with project id: {}, response code: {}".format(
project_id, response.status_code))
return result
# POST /projects/{project_id}/members
def add_project_member(self, project_id, username, role_id):
result = False
path = '%s://%s/api/projects/%s/members' % (self.protocol, self.host, project_id)
request_str = '{"username": "%s","roles": [%s]}' % (username, role_id)
request_body = json.dumps(json.loads(request_str))
response = requests.post(path,
cookies={'beegosessionID': self.session_id},
data=request_body, verify=False)
if response.status_code == 200:
result = True
logging.debug(
"Successfully add project member with project id: {}".format(
project_id))
else:
logging.error(
"Fail to add project member with project id: {}, response code: {}".format(
project_id, response.status_code))
return result
# DELETE /projects/{project_id}/members/{user_id}
def delete_member_from_project(self, project_id, user_id):
result = False
path = '%s://%s/api/projects/%s/members/%s' % (self.protocol, self.host,
project_id, user_id)
response = requests.delete(path,
cookies={'beegosessionID': self.session_id}, verify=False)
if response.status_code == 200:
result = True
logging.debug("Successfully delete member with id: {}".format(
user_id))
else:
logging.error("Fail to delete member with id: {}, response code: {}"
.format(user_id, response.status_code))
return result
# PUT /projects/{project_id}/publicity # PUT /projects/{project_id}/publicity
def set_project_publicity(self, project_id, is_public): def set_project_publicity(self, project_id, is_public):
result = False result = False
path = '%s://%s/api/projects/%s/publicity?project_id=%s' % ( path = '%s://%s/api/projects/%s/publicity' % (
self.protocol, self.host, project_id, project_id) self.protocol, self.host, project_id)
request_body = json.dumps({'public': is_public}) request_body = json.dumps({'public': is_public})
response = requests.put(path, response = requests.put(path,
cookies={'beegosessionID': self.session_id}, cookies={'beegosessionID': self.session_id},
data=request_body) data=request_body, verify=False)
if response.status_code == 200: if response.status_code == 200:
result = True result = True
logging.debug( logging.debug(
@ -150,26 +178,42 @@ class HarborClient(object):
result = None result = None
path = '%s://%s/api/statistics' % (self.protocol, self.host) path = '%s://%s/api/statistics' % (self.protocol, self.host)
response = requests.get(path, response = requests.get(path,
cookies={'beegosessionID': self.session_id}) cookies={'beegosessionID': self.session_id}, verify=False)
if response.status_code == 200: if response.status_code == 200:
result = response.json() result = response.json()
logging.debug("Successfully get statistics: {}".format(result)) logging.debug("Successfully get statistics: {}".format(result))
else: else:
logging.error("Fail to get statistics result") logging.error("Fail to get statistics result with status code: {}"
.format(response.status_code))
return result return result
# GET /users # GET /users
def get_users(self, user_name=None): def get_users(self):
# TODO: support parameter # TODO: support parameter
result = None result = None
path = '%s://%s/api/users' % (self.protocol, self.host) path = '%s://%s/api/users' % (self.protocol, self.host)
response = requests.get(path, response = requests.get(path,
cookies={'beegosessionID': self.session_id}) cookies={'beegosessionID': self.session_id}, verify=False)
if response.status_code == 200: if response.status_code == 200:
result = response.json() result = response.json()
logging.debug("Successfully get users result: {}".format(result)) logging.debug("Successfully get users result: {}".format(result))
else: else:
logging.error("Fail to get users result") logging.error("Fail to get users result with status code: {}"
.format(response.status_code))
return result
# GET /users/current
def get_user_info(self):
result = None
path = '%s://%s/api/users/current' % (self.protocol, self.host)
response = requests.get(path,
cookies={'beegosessionID': self.session_id}, verify=False)
if response.status_code == 200:
result = response.json()
logging.debug("Successfully get users result: {}".format(result))
else:
logging.error("Fail to get users result with status code: {}"
.format(response.status_code))
return result return result
# POST /users # POST /users
@ -183,7 +227,7 @@ class HarborClient(object):
'comment': comment}) 'comment': comment})
response = requests.post(path, response = requests.post(path,
cookies={'beegosessionID': self.session_id}, cookies={'beegosessionID': self.session_id},
data=request_body) data=request_body, verify=False)
if response.status_code == 201: if response.status_code == 201:
result = True result = True
logging.debug("Successfully create user with username: {}".format( logging.debug("Successfully create user with username: {}".format(
@ -198,14 +242,14 @@ class HarborClient(object):
def update_user_profile(self, user_id, email, realname, comment): def update_user_profile(self, user_id, email, realname, comment):
# TODO: support not passing comment # TODO: support not passing comment
result = False result = False
path = '%s://%s/api/users/%s?user_id=%s' % (self.protocol, self.host, path = '%s://%s/api/users/%s' % (self.protocol, self.host,
user_id, user_id) user_id)
request_body = json.dumps({'email': email, request_body = json.dumps({'email': email,
'realname': realname, 'realname': realname,
'comment': comment}) 'comment': comment})
response = requests.put(path, response = requests.put(path,
cookies={'beegosessionID': self.session_id}, cookies={'beegosessionID': self.session_id},
data=request_body) data=request_body, verify=False)
if response.status_code == 200: if response.status_code == 200:
result = True result = True
logging.debug( logging.debug(
@ -220,28 +264,29 @@ class HarborClient(object):
# DELETE /users/{user_id} # DELETE /users/{user_id}
def delete_user(self, user_id): def delete_user(self, user_id):
result = False result = False
path = '%s://%s/api/users/%s?user_id=%s' % (self.protocol, self.host, path = '%s://%s/api/users/%s' % (self.protocol, self.host,
user_id, user_id) user_id)
response = requests.delete(path, response = requests.delete(path,
cookies={'beegosessionID': self.session_id}) cookies={'beegosessionID': self.session_id}, verify=False)
if response.status_code == 200: if response.status_code == 200:
result = True result = True
logging.debug("Successfully delete user with id: {}".format( logging.debug("Successfully delete user with id: {}".format(
user_id)) user_id))
else: else:
logging.error("Fail to delete user with id: {}".format(user_id)) logging.error("Fail to delete user with id: {}, response code: {}"
.format(user_id, response.status_code))
return result return result
# PUT /users/{user_id}/password # PUT /users/{user_id}/password
def change_password(self, user_id, old_password, new_password): def change_password(self, user_id, old_password, new_password):
result = False result = False
path = '%s://%s/api/users/%s/password?user_id=%s' % ( path = '%s://%s/api/users/%s/password' % (
self.protocol, self.host, user_id, user_id) self.protocol, self.host, user_id)
request_body = json.dumps({'old_password': old_password, request_body = json.dumps({'old_password': old_password,
'new_password': new_password}) 'new_password': new_password})
response = requests.put(path, response = requests.put(path,
cookies={'beegosessionID': self.session_id}, cookies={'beegosessionID': self.session_id},
data=request_body) data=request_body, verify=False)
if response.status_code == 200: if response.status_code == 200:
result = True result = True
logging.debug( logging.debug(
@ -252,13 +297,15 @@ class HarborClient(object):
return result return result
# PUT /users/{user_id}/sysadmin # PUT /users/{user_id}/sysadmin
def promote_as_admin(self, user_id): def promote_as_admin(self, user_id, has_admin_role):
# TODO: always return 404, need more test
result = False result = False
path = '%s://%s/api/users/%s/sysadmin?user_id=%s' % ( path = '%s://%s/api/users/%s/sysadmin' % (
self.protocol, self.host, user_id, user_id) self.protocol, self.host, user_id)
request_body = json.dumps({'has_admin_role': has_admin_role,
'user_id': user_id})
response = requests.put(path, response = requests.put(path,
cookies={'beegosessionID': self.session_id}) cookies={'beegosessionID': self.session_id},
data=request_body, verify=False)
if response.status_code == 200: if response.status_code == 200:
result = True result = True
logging.debug( logging.debug(
@ -277,58 +324,73 @@ class HarborClient(object):
path = '%s://%s/api/repositories?project_id=%s' % ( path = '%s://%s/api/repositories?project_id=%s' % (
self.protocol, self.host, project_id) self.protocol, self.host, project_id)
response = requests.get(path, response = requests.get(path,
cookies={'beegosessionID': self.session_id}) cookies={'beegosessionID': self.session_id}, verify=False)
if response.status_code == 200: if response.status_code == 200:
result = response.json() result = response.json()
logging.debug( logging.debug(
"Successfully get repositories with id: {}, result: {}".format( "Successfully get repositories with id: {}, result: {}".format(
project_id, result)) project_id, result))
else: else:
logging.error("Fail to get repositories result with id: {}".format( logging.error("Fail to get repositories result with id: {}, response code: {}".format(
project_id)) project_id, response.status_code))
return result return result
# DELETE /repositories # DELETE /repositories/{repo_name}/tags/{tag}
def delete_repository(self, repo_name, tag=None): def delete_tag_of_repository(self, repo_name, tag):
# TODO: support to check tag
# TODO: return 200 but the repo is not deleted, need more test
result = False result = False
path = '%s://%s/api/repositories?repo_name=%s' % (self.protocol, path = '%s://%s/api/repositories/%s/tags/%s' % (self.protocol,self.host,
repo_name, tag)
response = requests.delete(path,
cookies={'beegosessionID': self.session_id}, verify=False)
if response.status_code == 200:
result = True
logging.debug("Successfully delete a tag of repository: {}".format(
repo_name))
else:
logging.error("Fail to delete repository with name: {}, response code: {}".format(
repo_name, response.status_code))
return result
# DELETE /repositories/{repo_name}/tags
def delete_tags_of_repository(self, repo_name):
result = False
path = '%s://%s/api/repositories/%s/tags' % (self.protocol,
self.host, repo_name) self.host, repo_name)
response = requests.delete(path, response = requests.delete(path,
cookies={'beegosessionID': self.session_id}) cookies={'beegosessionID': self.session_id}, verify=False)
if response.status_code == 200: if response.status_code == 200:
result = True result = True
logging.debug("Successfully delete repository: {}".format( logging.debug("Successfully delete repository: {}".format(
repo_name)) repo_name))
else: else:
logging.error("Fail to delete repository: {}".format(repo_name)) logging.error("Fail to delete repository with name: {}, response code: {}".format(
repo_name, response.status_code))
return result return result
# Get /repositories/tags # Get /repositories/{repo_name}/tags
def get_repository_tags(self, repo_name): def get_repository_tags(self, repo_name):
result = None result = None
path = '%s://%s/api/repositories/tags?repo_name=%s' % ( path = '%s://%s/api/repositories/%s/tags' % (
self.protocol, self.host, repo_name) self.protocol, self.host, repo_name)
response = requests.get(path, response = requests.get(path,
cookies={'beegosessionID': self.session_id}) cookies={'beegosessionID': self.session_id}, verify=False)
if response.status_code == 200: if response.status_code == 200:
result = response.json() result = response.json()
logging.debug( logging.debug(
"Successfully get tag with repo name: {}, result: {}".format( "Successfully get tag with repo name: {}, result: {}".format(
repo_name, result)) repo_name, result))
else: else:
logging.error("Fail to get tags with repo name: {}".format( logging.error("Fail to get tags with repo name: {}, response code: {}".format(
repo_name)) repo_name, response.status_code))
return result return result
# GET /repositories/manifests # GET /repositories/{repo_name}/tags/{tag}/manifest
def get_repository_manifests(self, repo_name, tag): def get_repository_manifest(self, repo_name, tag):
result = None result = None
path = '%s://%s/api/repositories/manifests?repo_name=%s&tag=%s' % ( path = '%s://%s/api/repositories/%s/tags/%s/manifest' % (
self.protocol, self.host, repo_name, tag) self.protocol, self.host, repo_name, tag)
response = requests.get(path, response = requests.get(path,
cookies={'beegosessionID': self.session_id}) cookies={'beegosessionID': self.session_id}, verify=False)
if response.status_code == 200: if response.status_code == 200:
result = response.json() result = response.json()
logging.debug( logging.debug(
@ -347,7 +409,7 @@ class HarborClient(object):
if count: if count:
path += "?count=%s" % (count) path += "?count=%s" % (count)
response = requests.get(path, response = requests.get(path,
cookies={'beegosessionID': self.session_id}) cookies={'beegosessionID': self.session_id}, verify=False)
if response.status_code == 200: if response.status_code == 200:
result = response.json() result = response.json()
logging.debug( logging.debug(
@ -362,7 +424,7 @@ class HarborClient(object):
result = None result = None
path = '%s://%s/api/logs' % (self.protocol, self.host) path = '%s://%s/api/logs' % (self.protocol, self.host)
response = requests.get(path, response = requests.get(path,
cookies={'beegosessionID': self.session_id}) cookies={'beegosessionID': self.session_id}, verify=False)
if response.status_code == 200: if response.status_code == 200:
result = response.json() result = response.json()
logging.debug("Successfully get logs") logging.debug("Successfully get logs")
@ -370,3 +432,31 @@ class HarborClient(object):
logging.error("Fail to get logs and response code: {}".format( logging.error("Fail to get logs and response code: {}".format(
response.status_code)) response.status_code))
return result return result
# Get /systeminfo
def get_systeminfo(self):
result = None
path = '%s://%s/api/systeminfo' % (self.protocol, self.host)
response = requests.get(path,
cookies={'beegosessionID': self.session_id}, verify=False)
if response.status_code == 200:
result = response.json()
logging.debug(
"Successfully get systeminfo, result: {}".format(result))
else:
logging.error("Fail to get systeminfo, response code: {}".format(response.status_code))
return result
# Get /configurations
def get_configurations(self):
result = None
path = '%s://%s/api/configurations' % (self.protocol, self.host)
response = requests.get(path,
cookies={'beegosessionID': self.session_id}, verify=False)
if response.status_code == 200:
result = response.json()
logging.debug(
"Successfully get configurations, result: {}".format(result))
else:
logging.error("Fail to get configurations, response code: {}".format(response.status_code))
return result