quay/endpoints/api/tag.py

from flask import request

from endpoints.api import (resource, nickname, require_repo_read, require_repo_write,
                           RepositoryParamResource, log_action, NotFound, validate_json_request,
                           path_param)
from endpoints.api.image import image_view
from data import model
from auth.auth_context import get_authenticated_user


@resource('/v1/repository/<repopath:repository>/tag/<tag>')
@path_param('repository', 'The full path of the repository. e.g. namespace/name')
@path_param('tag', 'The name of the tag')
class RepositoryTag(RepositoryParamResource):
  """ Resource for managing repository tags. """
  schemas = {
    'MoveTag': {
      'id': 'MoveTag',
      'type': 'object',
      'description': 'Description of to which image a new or existing tag should point',
      'required': [
        'image',
      ],
      'properties': {
        'image': {
          'type': 'string',
          'description': 'Image identifier to which the tag should point',
        },
      },
    },
  }

  @require_repo_write
  @nickname('changeTagImage')
  @validate_json_request('MoveTag')
  def put(self, namespace, repository, tag):
    """ Change which image a tag points to or create a new tag."""
    image_id = request.get_json()['image']
    image = model.get_repo_image(namespace, repository, image_id)
    if not image:
      raise NotFound()

    original_image_id = None
    try:
      original_tag_image = model.get_tag_image(namespace, repository, tag)
      if original_tag_image:
        original_image_id = original_tag_image.docker_image_id
    except model.DataModelException:
      # This is a new tag.
      pass

    model.create_or_update_tag(namespace, repository, tag, image_id)
    model.garbage_collect_repository(namespace, repository)

    username = get_authenticated_user().username
    log_action('move_tag' if original_image_id else 'create_tag', namespace,
               {'username': username, 'repo': repository, 'tag': tag,
                'image': image_id, 'original_image': original_image_id},
               repo=model.get_repository(namespace, repository))

    return 'Updated', 201

  @require_repo_write
  @nickname('deleteFullTag')
  def delete(self, namespace, repository, tag):
    """ Delete the specified repository tag. """
    model.delete_tag(namespace, repository, tag)
    model.garbage_collect_repository(namespace, repository)

    username = get_authenticated_user().username
    log_action('delete_tag', namespace,
               {'username': username, 'repo': repository, 'tag': tag},
               repo=model.get_repository(namespace, repository))

    return 'Deleted', 204


@resource('/v1/repository/<repopath:repository>/tag/<tag>/images')
@path_param('repository', 'The full path of the repository. e.g. namespace/name')
@path_param('tag', 'The name of the tag')
class RepositoryTagImages(RepositoryParamResource):
  """ Resource for listing the images in a specific repository tag. """
  @require_repo_read
  @nickname('listTagImages')
  def get(self, namespace, repository, tag):
    """ List the images for the specified repository tag. """
    try:
      tag_image = model.get_tag_image(namespace, repository, tag)
    except model.DataModelException:
      raise NotFound()

    parent_images = model.get_parent_images(namespace, repository, tag_image)
    image_map = {}
    for image in parent_images:
      image_map[str(image.id)] = image.docker_image_id

    parents = list(parent_images)
    parents.reverse()
    all_images = [tag_image] + parents

    return {
      'images': [image_view(image, image_map) for image in all_images]
    }