quay/endpoints/api/permission_models_pre_oci.py

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()