quay/endpoints/api/team.py

from flask import request

from endpoints.api import (resource, nickname, ApiResource, validate_json_request, request_error,
                           log_action, Unauthorized, NotFound, internal_only, require_scope,
                           path_param)
from auth.permissions import AdministerOrganizationPermission, ViewTeamPermission
from auth.auth_context import get_authenticated_user
from auth import scopes
from data import model


def team_view(orgname, team):
  view_permission = ViewTeamPermission(orgname, team.name)
  role = model.get_team_org_role(team).name
  return {
    'id': team.id,
    'name': team.name,
    'description': team.description,
    'can_view': view_permission.can(),
    'role': role
  }

def member_view(member):
  return {
    'name': member.username,
    'kind': 'user',
    'is_robot': member.robot,
  }


@resource('/v1/organization/<orgname>/team/<teamname>')
@path_param('orgname', 'The name of the organization')
@path_param('teamname', 'The name of the team')
@internal_only
class OrganizationTeam(ApiResource):
  """ Resource for manging an organization's teams. """
  schemas = {
    'TeamDescription': {
      'id': 'TeamDescription',
      'type': 'object',
      'description': 'Description of a team',
      'required': [
        'role',
      ],
      'properties': {
        'role': {
          'type': 'string',
          'description': 'Org wide permissions that should apply to the team',
          'enum': [
            'member',
            'creator',
            'admin',
          ],
        },
        'description': {
          'type': 'string',
          'description': 'Markdown description for the team',
        },
      },
    },
  }

  @nickname('updateOrganizationTeam')
  @validate_json_request('TeamDescription')
  def put(self, orgname, teamname):
    """ Update the org-wide permission for the specified team. """
    edit_permission = AdministerOrganizationPermission(orgname)
    if edit_permission.can():
      team = None

      details = request.get_json()
      is_existing = False
      try:
        team = model.get_organization_team(orgname, teamname)
        is_existing = True
      except model.InvalidTeamException:
        # Create the new team.
        description = details['description'] if 'description' in details else ''
        role = details['role'] if 'role' in details else 'member'

        org = model.get_organization(orgname)
        team = model.create_team(teamname, org, role, description)
        log_action('org_create_team', orgname, {'team': teamname})

      if is_existing:
        if ('description' in details and
            team.description != details['description']):
          team.description = details['description']
          team.save()
          log_action('org_set_team_description', orgname,
                     {'team': teamname, 'description': team.description})

        if 'role' in details:
          role = model.get_team_org_role(team).name
          if role != details['role']:
            team = model.set_team_org_permission(team, details['role'],
                                                 get_authenticated_user().username)
            log_action('org_set_team_role', orgname, {'team': teamname, 'role': details['role']})
            
      return team_view(orgname, team), 200

    raise Unauthorized()

  @nickname('deleteOrganizationTeam')
  def delete(self, orgname, teamname):
    """ Delete the specified team. """
    permission = AdministerOrganizationPermission(orgname)
    if permission.can():
      model.remove_team(orgname, teamname, get_authenticated_user().username)
      log_action('org_delete_team', orgname, {'team': teamname})
      return 'Deleted', 204

    raise Unauthorized()


@resource('/v1/organization/<orgname>/team/<teamname>/members')
@path_param('orgname', 'The name of the organization')
@path_param('teamname', 'The name of the team')
@internal_only
class TeamMemberList(ApiResource):
  """ Resource for managing the list of members for a team. """
  @nickname('getOrganizationTeamMembers')
  def get(self, orgname, teamname):
    """ Retrieve the list of members for the specified team. """
    view_permission = ViewTeamPermission(orgname, teamname)
    edit_permission = AdministerOrganizationPermission(orgname)

    if view_permission.can():
      team = None
      try:
        team = model.get_organization_team(orgname, teamname)
      except model.InvalidTeamException:
        raise NotFound()
        
      members = model.get_organization_team_members(team.id)
      return {
        'members': {m.username : member_view(m) for m in members},
        'can_edit': edit_permission.can()
      }

    raise Unauthorized()


@resource('/v1/organization/<orgname>/team/<teamname>/members/<membername>')
@path_param('orgname', 'The name of the organization')
@path_param('teamname', 'The name of the team')
@path_param('membername', 'The username of the team member')
class TeamMember(ApiResource):
  """ Resource for managing individual members of a team. """
  @require_scope(scopes.ORG_ADMIN)
  @nickname('updateOrganizationTeamMember')
  def put(self, orgname, teamname, membername):
    """ Add a member to an existing team. """
    permission = AdministerOrganizationPermission(orgname)
    if permission.can():
      team = None
      user = None

      # Find the team.
      try:
        team = model.get_organization_team(orgname, teamname)
      except model.InvalidTeamException:
        raise NotFound()

      # Find the user.
      user = model.get_user(membername)
      if not user:
        raise request_error(message='Unknown user')
        
      # Add the user to the team.
      model.add_user_to_team(user, team)
      log_action('org_add_team_member', orgname, {'member': membername, 'team': teamname})
      return member_view(user)

    raise Unauthorized()

  @require_scope(scopes.ORG_ADMIN)
  @nickname('deleteOrganizationTeamMember')
  def delete(self, orgname, teamname, membername):
    """ Delete an existing member of a team. """
    permission = AdministerOrganizationPermission(orgname)
    if permission.can():
      # Remote the user from the team.
      invoking_user = get_authenticated_user().username
      model.remove_user_from_team(orgname, teamname, membername, invoking_user)
      log_action('org_remove_team_member', orgname, {'member': membername, 'team': teamname})
      return 'Deleted', 204

    raise Unauthorized()