quay/endpoints/api/repository.py

import logging
import json

from flask import request

from data import model
from endpoints.api import (truthy_bool, format_date, nickname, log_action, validate_json_request,
                           require_repo_read, require_repo_write, require_repo_admin,
                           RepositoryParamResource, resource, query_param, parse_args, ApiResource,
                           request_error, require_scope, Unauthorized, NotFound, InvalidRequest)
from auth.permissions import (ModifyRepositoryPermission, AdministerRepositoryPermission,
                              CreateRepositoryPermission, ReadRepositoryPermission)
from auth.auth_context import get_authenticated_user
from auth import scopes


logger = logging.getLogger(__name__)


@resource('/v1/repository')
class RepositoryList(ApiResource):
  """Operations for creating and listing repositories."""
  schemas = {
    'NewRepo': {
      'id': 'NewRepo',
      'type': 'object',
      'description': 'Description of a new repository',
      'required': [
        'repository',
        'visibility',
        'description',
      ],
      'properties': {
        'repository': {
          'type': 'string',
          'description': 'Repository name',
        },
        'visibility': {
          'type': 'string',
          'description': 'Visibility which the repository will start with',
          'enum': [
            'public',
            'private',
          ],
        },
        'namespace': {
          'type': 'string',
          'description': ('Namespace in which the repository should be created. If omitted, the '
                          'username of the caller is used'),
        },
        'description': {
          'type': 'string',
          'description': 'Markdown encoded description for the repository',
        },
      },
    },
  }

  @require_scope(scopes.CREATE_REPO)
  @nickname('createRepo')
  @validate_json_request('NewRepo')
  def post(self):
    """Create a new repository."""
    owner = get_authenticated_user()
    req = request.get_json()

    if owner is None and 'namespace' not in 'req':
      raise InvalidRequest('Must provide a namespace or must be logged in.')

    namespace_name = req['namespace'] if 'namespace' in req else owner.username

    permission = CreateRepositoryPermission(namespace_name)
    if permission.can():
      repository_name = req['repository']
      visibility = req['visibility']

      existing = model.get_repository(namespace_name, repository_name)
      if existing:
        raise request_error(message='Repository already exists')

      visibility = req['visibility']

      repo = model.create_repository(namespace_name, repository_name, owner,
                                     visibility)
      repo.description = req['description']
      repo.save()

      log_action('create_repo', namespace_name, {'repo': repository_name,
                                                 'namespace': namespace_name}, repo=repo)
      return {
        'namespace': namespace_name,
        'name': repository_name
      }, 201

    raise Unauthorized()

  @require_scope(scopes.READ_REPO)
  @nickname('listRepos')
  @parse_args
  @query_param('page', 'Offset page number. (int)', type=int)
  @query_param('limit', 'Limit on the number of results (int)', type=int)
  @query_param('namespace', 'Namespace to use when querying for org repositories.', type=str)
  @query_param('public', 'Whether to include public repositories.', type=truthy_bool, default=True)
  @query_param('private', 'Whether to inlcude private repositories.', type=truthy_bool,
               default=True)
  @query_param('sort', 'Whether to sort the results.', type=truthy_bool, default=False)
  @query_param('count', 'Whether to include a count of the total number of results available.',
               type=truthy_bool, default=False)
  def get(self, args):
    """Fetch the list of repositories under a variety of situations."""
    def repo_view(repo_obj):
      return {
        'namespace': repo_obj.namespace,
        'name': repo_obj.name,
        'description': repo_obj.description,
        'is_public': repo_obj.visibility.name == 'public',
      }

    username = None
    if get_authenticated_user() and args['private']:
      username = get_authenticated_user().username

    response = {}

    repo_count = None
    if args['count']:
      repo_count = model.get_visible_repository_count(username, include_public=args['public'],
                                                      namespace=args['namespace'])
      response['count'] = repo_count

    repo_query = model.get_visible_repositories(username, limit=args['limit'], page=args['page'],
                                                include_public=args['public'], sort=args['sort'],
                                                namespace=args['namespace'])

    response['repositories'] = [repo_view(repo) for repo in repo_query
                                if (repo.visibility.name == 'public' or
                                    ReadRepositoryPermission(repo.namespace, repo.name).can())]

    return response


@resource('/v1/repository/<repopath:repository>')
class Repository(RepositoryParamResource):
  """Operations for managing a specific repository."""
  schemas = {
    'RepoUpdate': {
      'id': 'RepoUpdate',
      'type': 'object',
      'description': 'Fields which can be updated in a repository.',
      'required': [
        'description',
      ],
      'properties': {
        'description': {
          'type': 'string',
          'description': 'Markdown encoded description for the repository',
        },
      }
    }
  }

  @require_repo_read
  @nickname('getRepo')
  def get(self, namespace, repository):
    """Fetch the specified repository."""
    logger.debug('Get repo: %s/%s' % (namespace, repository))

    def tag_view(tag):
      return {
        'name': tag.name,
        'image_id': tag.image.docker_image_id,
        'dbid': tag.image.id
      }

    organization = None
    try:
      organization = model.get_organization(namespace)
    except model.InvalidOrganizationException:
      pass

    is_public = model.repository_is_public(namespace, repository)
    repo = model.get_repository(namespace, repository)
    if repo:
      tags = model.list_repository_tags(namespace, repository)
      tag_dict = {tag.name: tag_view(tag) for tag in tags}
      can_write = ModifyRepositoryPermission(namespace, repository).can()
      can_admin = AdministerRepositoryPermission(namespace, repository).can()
      active_builds = model.list_repository_builds(namespace, repository, 1,
                                                   include_inactive=False)

      return {
        'namespace': namespace,
        'name': repository,
        'description': repo.description,
        'tags': tag_dict,
        'can_write': can_write,
        'can_admin': can_admin,
        'is_public': is_public,
        'is_building': len(list(active_builds)) > 0,
        'is_organization': bool(organization),
        'status_token': repo.badge_token if not is_public else ''
      }

    raise NotFound()
  
  @require_repo_write
  @nickname('updateRepo')
  @validate_json_request('RepoUpdate')
  def put(self, namespace, repository):
    """ Update the description in the specified repository. """
    repo = model.get_repository(namespace, repository)
    if repo:
      values = request.get_json()
      repo.description = values['description']
      repo.save()
      
      log_action('set_repo_description', namespace,
                 {'repo': repository, 'description': values['description']},
                 repo=repo)
      return {
          'success': True
      }
    raise NotFound()

  @require_repo_admin
  @nickname('deleteRepository')
  def delete(self, namespace, repository):
    """ Delete a repository. """
    model.purge_repository(namespace, repository)
    log_action('delete_repo', namespace,
               {'repo': repository, 'namespace': namespace})
    return 'Deleted', 204


@resource('/v1/repository/<repopath:repository>/changevisibility')
class RepositoryVisibility(RepositoryParamResource):
  """ Custom verb for changing the visibility of the repository. """
  schemas = {
    'ChangeVisibility': {
      'id': 'ChangeVisibility',
      'type': 'object',
      'description': 'Change the visibility for the repository.',
      'required': [
        'visibility',
      ],
      'properties': {
        'visibility': {
          'type': 'string',
          'description': 'Visibility which the repository will start with',
          'enum': [
            'public',
            'private',
          ],
        },
      }
    }
  }

  @require_repo_admin
  @nickname('changeRepoVisibility')
  @validate_json_request('ChangeVisibility')
  def post(self, namespace, repository):
    """ Change the visibility of a repository. """
    repo = model.get_repository(namespace, repository)
    if repo:
      values = request.get_json()
      model.set_repository_visibility(repo, values['visibility'])
      log_action('change_repo_visibility', namespace,
                 {'repo': repository, 'visibility': values['visibility']},
                 repo=repo)
      return {
          'success': True
      }