115 lines
5 KiB
Python
115 lines
5 KiB
Python
from app import avatar
|
|
from data import model
|
|
from permission_models_interface import PermissionDataInterface, UserPermission, TeamPermission, Role, SaveException, DeleteException
|
|
|
|
|
|
class PreOCIModel(PermissionDataInterface):
|
|
"""
|
|
PreOCIModel implements the data model for Permission using a database schema
|
|
before it was changed to support the OCI specification.
|
|
"""
|
|
|
|
def get_repo_permissions_by_user(self, namespace_name, repository_name):
|
|
org = None
|
|
try:
|
|
org = model.organization.get_organization(namespace_name) # Will raise an error if not org
|
|
except model.InvalidOrganizationException:
|
|
# This repository isn't under an org
|
|
pass
|
|
|
|
# Load the permissions.
|
|
repo_perms = model.user.get_all_repo_users(namespace_name, repository_name)
|
|
|
|
if org:
|
|
users_filter = {perm.user for perm in repo_perms}
|
|
org_members = model.organization.get_organization_member_set(org, users_filter=users_filter)
|
|
|
|
def is_org_member(user):
|
|
if not org:
|
|
return False
|
|
|
|
return user.robot or user.username in org_members
|
|
|
|
return [self._user_permission(perm, org is not None, is_org_member(perm.user)) for perm in repo_perms]
|
|
|
|
def get_repo_roles(self, username, namespace_name, repository_name):
|
|
user = model.user.get_user(username)
|
|
if not user:
|
|
return None
|
|
|
|
repo = model.repository.get_repository(namespace_name, repository_name)
|
|
if not repo:
|
|
return None
|
|
|
|
return [self._role(r) for r in model.permission.get_user_repo_permissions(user, repo)]
|
|
|
|
def get_repo_permission_for_user(self, username, namespace_name, repository_name):
|
|
perm = model.permission.get_user_reponame_permission(username, namespace_name, repository_name)
|
|
org = None
|
|
try:
|
|
org = model.organization.get_organization(namespace_name)
|
|
org_members = model.organization.get_organization_member_set(org, users_filter={perm.user})
|
|
is_org_member = perm.user.robot or perm.user.username in org_members
|
|
except model.InvalidOrganizationException:
|
|
# This repository is not part of an organization
|
|
is_org_member = False
|
|
|
|
return self._user_permission(perm, org is not None, is_org_member)
|
|
|
|
def set_repo_permission_for_user(self, username, namespace_name, repository_name, role_name):
|
|
try:
|
|
perm = model.permission.set_user_repo_permission(username, namespace_name, repository_name, role_name)
|
|
org = None
|
|
try:
|
|
org = model.organization.get_organization(namespace_name)
|
|
org_members = model.organization.get_organization_member_set(org, users_filter={perm.user})
|
|
is_org_member = perm.user.robot or perm.user.username in org_members
|
|
except model.InvalidOrganizationException:
|
|
# This repository is not part of an organization
|
|
is_org_member = False
|
|
return self._user_permission(perm, org is not None, is_org_member)
|
|
except model.DataModelException as ex:
|
|
raise SaveException(ex)
|
|
|
|
def delete_repo_permission_for_user(self, username, namespace_name, repository_name):
|
|
try:
|
|
model.permission.delete_user_permission(username, namespace_name, repository_name)
|
|
except model.DataModelException as ex:
|
|
raise DeleteException(ex)
|
|
|
|
def get_repo_permissions_by_team(self, namespace_name, repository_name):
|
|
repo_perms = model.permission.get_all_repo_teams(namespace_name, repository_name)
|
|
return [self._team_permission(perm, perm.team.name) for perm in repo_perms]
|
|
|
|
def get_repo_role_for_team(self, team_name, namespace_name, repository_name):
|
|
return self._role(model.permission.get_team_reponame_permission(team_name, namespace_name, repository_name))
|
|
|
|
def set_repo_permission_for_team(self, team_name, namespace_name, repository_name, role_name):
|
|
try:
|
|
return self._team_permission(model.permission.set_team_repo_permission(team_name, namespace_name, repository_name, role_name), team_name)
|
|
except model.DataModelException as ex:
|
|
raise SaveException(ex)
|
|
|
|
def delete_repo_permission_for_team(self, team_name, namespace_name, repository_name):
|
|
try:
|
|
model.permission.delete_team_permission(team_name, namespace_name, repository_name)
|
|
except model.DataModelException as ex:
|
|
raise DeleteException(ex)
|
|
|
|
def _role(self, permission_obj):
|
|
return Role(role_name=permission_obj.role.name)
|
|
|
|
def _user_permission(self, permission_obj, has_org, is_org_member):
|
|
return UserPermission(role_name=permission_obj.role.name,
|
|
username=permission_obj.user.username,
|
|
is_robot=permission_obj.user.robot,
|
|
avatar=avatar.get_data_for_user(permission_obj.user),
|
|
is_org_member=is_org_member,
|
|
has_org=has_org)
|
|
|
|
def _team_permission(self, permission_obj, team_name):
|
|
return TeamPermission(role_name=permission_obj.role.name,
|
|
team_name=permission_obj.team.name,
|
|
avatar=avatar.get_data_for_team(permission_obj.team))
|
|
|
|
pre_oci_model = PreOCIModel()
|