88 lines
3.3 KiB
Python
88 lines
3.3 KiB
Python
from typing import List
|
|
|
|
import requests
|
|
from requests.auth import HTTPBasicAuth
|
|
from zope import component
|
|
from zope.interface import implementer
|
|
|
|
from ceo_common.errors import UserNotFoundError
|
|
from ceo_common.interfaces import IContainerRegistryService, IConfig
|
|
|
|
|
|
@implementer(IContainerRegistryService)
|
|
class ContainerRegistryService:
|
|
def __init__(self):
|
|
cfg = component.getUtility(IConfig)
|
|
self.base_url = cfg.get('registry_base_url')
|
|
if self.base_url.endswith('/'):
|
|
self.base_url = self.base_url[:-1]
|
|
api_username = cfg.get('registry_username')
|
|
api_password = cfg.get('registry_password')
|
|
self.basic_auth = HTTPBasicAuth(api_username, api_password)
|
|
|
|
def _http_request(self, method: str, path: str, **kwargs):
|
|
return requests.request(
|
|
method, self.base_url + path, **kwargs, auth=self.basic_auth)
|
|
|
|
def _http_get(self, path: str, **kwargs):
|
|
return self._http_request('GET', path, **kwargs)
|
|
|
|
def _http_post(self, path, **kwargs):
|
|
return self._http_request('POST', path, **kwargs)
|
|
|
|
def _http_delete(self, path, **kwargs):
|
|
return self._http_request('DELETE', path, **kwargs)
|
|
|
|
def _get_account(self, username: str):
|
|
resp = self._http_get('/users', params={'username': username})
|
|
users = resp.json()
|
|
if len(users) < 1:
|
|
raise UserNotFoundError(username)
|
|
return users[0]
|
|
|
|
def get_accounts(self) -> List[str]:
|
|
# We're only interested in accounts which have a project, so
|
|
# we're actually just going to get a list of projects
|
|
resp = self._http_get('/projects')
|
|
resp.raise_for_status()
|
|
# This project is only owned by the admin account, so we don't
|
|
# want to include it
|
|
project_exceptions = ['library']
|
|
return [
|
|
project['name'] for project in resp.json()
|
|
if project['name'] not in project_exceptions
|
|
]
|
|
|
|
def create_project_for_user(self, username: str):
|
|
user_id = self._get_account(username)['user_id']
|
|
|
|
# Create the project
|
|
resp = self._http_post(
|
|
'/projects', json={'project_name': username, 'public': True})
|
|
# 409 => project already exists (that is OK)
|
|
if resp.status_code != 409:
|
|
resp.raise_for_status()
|
|
|
|
# Add the user as a project admin (role ID 1)
|
|
resp = self._http_post(
|
|
f'/projects/{username}/members',
|
|
json={'role_id': 1, 'member_user': {'user_id': user_id}})
|
|
# 409 => project member already exists (that is OK)
|
|
if resp.status_code != 409:
|
|
resp.raise_for_status()
|
|
|
|
def delete_project_for_user(self, username: str):
|
|
# Delete all of the repositories inside the project first
|
|
resp = self._http_get(f'/projects/{username}/repositories')
|
|
if resp.status_code == 403:
|
|
# For some reason a 403 is returned if the project doesn't exist
|
|
return
|
|
resp.raise_for_status()
|
|
repositories = [repo['name'] for repo in resp.json()]
|
|
for repo in repositories:
|
|
resp = self._http_delete(f'/projects/{username}/repositories/{repo}')
|
|
resp.raise_for_status()
|
|
# Delete the project now that it is empty
|
|
resp = self._http_delete(f'/projects/{username}')
|
|
resp.raise_for_status()
|