quay/endpoints/v1/tag.py

import logging
import json

from flask import abort, request, jsonify, make_response, session

from util.names import TAG_ERROR, TAG_REGEX
from auth.auth import process_auth
from auth.permissions import (ReadRepositoryPermission,
                              ModifyRepositoryPermission)
from data import model
from endpoints.common import parse_repository_name
from endpoints.decorators import anon_protect
from endpoints.v1 import v1_bp
from endpoints.trackhelper import track_and_log


logger = logging.getLogger(__name__)


@v1_bp.route('/repositories/<repopath:repository>/tags', methods=['GET'])
@process_auth
@anon_protect
@parse_repository_name()
def get_tags(namespace_name, repo_name):
  permission = ReadRepositoryPermission(namespace_name, repo_name)

  if permission.can() or model.repository.repository_is_public(namespace_name, repo_name):
    tags = model.tag.list_repository_tags(namespace_name, repo_name)
    tag_map = {tag.name: tag.image.docker_image_id for tag in tags}
    return jsonify(tag_map)

  abort(403)


@v1_bp.route('/repositories/<repopath:repository>/tags/<tag>', methods=['GET'])
@process_auth
@anon_protect
@parse_repository_name()
def get_tag(namespace_name, repo_name, tag):
  permission = ReadRepositoryPermission(namespace_name, repo_name)

  if permission.can() or model.repository.repository_is_public(namespace_name, repo_name):
    try:
      tag_image = model.tag.get_tag_image(namespace_name, repo_name, tag)
    except model.DataModelException:
      abort(404)

    resp = make_response('"%s"' % tag_image.docker_image_id)
    resp.headers['Content-Type'] = 'application/json'
    return resp

  abort(403)


@v1_bp.route('/repositories/<repopath:repository>/tags/<tag>', methods=['PUT'])
@process_auth
@anon_protect
@parse_repository_name()
def put_tag(namespace_name, repo_name, tag):
  permission = ModifyRepositoryPermission(namespace_name, repo_name)

  if permission.can():
    if not TAG_REGEX.match(tag):
      abort(400, TAG_ERROR)

    docker_image_id = json.loads(request.data)
    model.tag.create_or_update_tag(namespace_name, repo_name, tag, docker_image_id)

    # Store the updated tag.
    if 'pushed_tags' not in session:
      session['pushed_tags'] = {}

    session['pushed_tags'][tag] = docker_image_id

    return make_response('Created', 200)

  abort(403)


@v1_bp.route('/repositories/<repopath:repository>/tags/<tag>', methods=['DELETE'])
@process_auth
@anon_protect
@parse_repository_name()
def delete_tag(namespace_name, repo_name, tag):
  permission = ModifyRepositoryPermission(namespace_name, repo_name)

  if permission.can():
    model.tag.delete_tag(namespace_name, repo_name, tag)
    track_and_log('delete_tag', model.repository.get_repository(namespace_name, repo_name),
                  tag=tag)
    return make_response('Deleted', 200)

  abort(403)