Python CSC Electronic Office
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
pyceo/ceod/model/ContainerRegistryService.py

86 lines
3.3 KiB

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)
self.projects_to_ignore = cfg.get('registry_projects_to_ignore')
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()
return [
project['name'] for project in resp.json()
if project['name'] not in self.projects_to_ignore
]
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()
# Each repo name has the format project/image
repositories = [repo['name'].split('/')[1] 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()