quay/endpoints/api/search.py

from endpoints.api import (ApiResource, parse_args, query_param, truthy_bool, nickname, resource,
                           require_scope, path_param)
from data import model
from auth.permissions import (OrganizationMemberPermission, ViewTeamPermission,
                              ReadRepositoryPermission, UserAdminPermission,
                              AdministerOrganizationPermission)
from auth.auth_context import get_authenticated_user
from auth import scopes
from app import avatar, get_app_url
from operator import itemgetter

import math

@resource('/v1/entities/<prefix>')
class EntitySearch(ApiResource):
  """ Resource for searching entities. """
  @path_param('prefix', 'The prefix of the entities being looked up')
  @parse_args
  @query_param('namespace', 'Namespace to use when querying for org entities.', type=str,
               default='')
  @query_param('includeTeams', 'Whether to include team names.', type=truthy_bool, default=False)
  @query_param('includeOrgs', 'Whether to include orgs names.', type=truthy_bool, default=False)
  @nickname('getMatchingEntities')
  def get(self, args, prefix):
    """ Get a list of entities that match the specified prefix. """
    teams = []
    org_data = []

    namespace_name = args['namespace']
    robot_namespace = None
    organization = None

    try:
      organization = model.get_organization(namespace_name)

      # namespace name was an org
      permission = OrganizationMemberPermission(namespace_name)
      if permission.can():
        robot_namespace = namespace_name

        if args['includeTeams']:
          teams = model.get_matching_teams(prefix, organization)

        if args['includeOrgs'] and AdministerOrganizationPermission(namespace_name) \
           and namespace_name.startswith(prefix):
          org_data = [{
            'name': namespace_name,
            'kind': 'org',
            'is_org_member': True,
            'avatar': avatar.get_data_for_org(organization),
          }]

    except model.InvalidOrganizationException:
      # namespace name was a user
      user = get_authenticated_user()
      if user and user.username == namespace_name:
        # Check if there is admin user permissions (login only)
        admin_permission = UserAdminPermission(user.username)
        if admin_permission.can():
          robot_namespace = namespace_name

    users = model.get_matching_users(prefix, robot_namespace, organization)

    def entity_team_view(team):
      result = {
        'name': team.name,
        'kind': 'team',
        'is_org_member': True,
        'avatar': avatar.get_data_for_team(team)
      }
      return result

    def user_view(user):
      user_json = {
        'name': user.username,
        'kind': 'user',
        'is_robot': user.robot,
        'avatar': avatar.get_data_for_user(user)
      }

      if organization is not None:
        user_json['is_org_member'] = user.robot or user.is_org_member

      return user_json

    team_data = [entity_team_view(team) for team in teams]
    user_data = [user_view(user) for user in users]

    return {
      'results': team_data + user_data + org_data
    }


  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
    }


@resource('/v1/find/all')
class ConductSearch(ApiResource):
  """ Resource for finding users, repositories, teams, etc. """
  @parse_args
  @query_param('query', 'The search query.', type=str, default='')
  @require_scope(scopes.READ_REPO)
  @nickname('conductSearch')
  def get(self, args):
    """ Get a list of entities and resources that match the specified query. """
    prefix = args['query']
    username = None
    results = []

    def entity_view(entity):
      kind = 'user'
      avatar_data = avatar.get_data_for_user(entity)
      href = '/user/' + entity.username

      if entity.organization:
        kind = 'organization'
        avatar_data = avatar.get_data_for_org(entity)
        href = '/organization/' + entity.username
      elif entity.robot:
        kind = 'robot'
        href = '/user?tab=robots'
        avatar_data = None

      return {
        'kind': kind,
        'avatar': avatar_data,
        'name': entity.username,
        'score': 1,
        'href': href
      }

    if get_authenticated_user():
      username = get_authenticated_user().username

      # Find the matching teams.
      matching_teams = model.get_matching_user_teams(prefix, get_authenticated_user())
      for team in matching_teams:
        results.append({
          'kind': 'team',
          'name': team.name,
          'organization': entity_view(team.organization),
          'avatar': avatar.get_data_for_team(team),
          'score': 2,
          'href': '/organization/' + team.organization.username + '/teams/' + team.name
        })

    # Find the matching repositories.
    matching_repos = model.get_matching_repositories(prefix, username)
    matching_repo_counts = {t[0]: t[1] for t in model.get_repository_pull_counts(matching_repos)}

    for repo in matching_repos:
      results.append({
        'kind': 'repository',
        'namespace': entity_view(repo.namespace_user),
        'name': repo.name,
        'description': repo.description,
        'is_public': repo.visibility.name == 'public',
        'score': math.log(matching_repo_counts.get(repo.id, 1), 10),
        'href': '/repository/' + repo.namespace_user.username + '/' + repo.name
      })

    # Find the matching users, robots and organizations.
    matching_entities = model.get_matching_user_entities(prefix, get_authenticated_user())
    for entity in matching_entities:
      results.append(entity_view(entity))

    return {'results': sorted(results, key=itemgetter('score'), reverse=True)}


@resource('/v1/find/repository')
class FindRepositories(ApiResource):
  """ Resource for finding repositories. """
  @parse_args
  @query_param('query', 'The prefix to use when querying for repositories.', type=str, default='')
  @require_scope(scopes.READ_REPO)
  @nickname('findRepos')
  def get(self, args):
    """ Get a list of repositories that match the specified prefix query. """
    prefix = args['query']

    def repo_view(repo):
      return {
        'namespace': repo.namespace_user.username,
        'name': repo.name,
        'description': repo.description
      }

    username = None
    user = get_authenticated_user()
    if user is not None:
      username = user.username

    matching = model.get_matching_repositories(prefix, username)
    return {
      'repositories': [repo_view(repo) for repo in matching
                       if (repo.visibility.name == 'public' or
                           ReadRepositoryPermission(repo.namespace_user.username, repo.name).can())]
    }