quay/data/model/tag.py

from uuid import uuid4

from data.model import image, db_transaction, DataModelException, _basequery
from data.database import (RepositoryTag, Repository, Image, ImageStorage, Namespace,
                           get_epoch_timestamp, db_for_update)


def _tag_alive(query, now_ts=None):
  if now_ts is None:
    now_ts = get_epoch_timestamp()
  return query.where((RepositoryTag.lifetime_end_ts >> None) |
                     (RepositoryTag.lifetime_end_ts > now_ts))


def list_repository_tags(namespace_name, repository_name, include_hidden=False,
                         include_storage=False):
  to_select = (RepositoryTag, Image)
  if include_storage:
    to_select = (RepositoryTag, Image, ImageStorage)

  query = _tag_alive(RepositoryTag
                     .select(*to_select)
                     .join(Repository)
                     .join(Namespace, on=(Repository.namespace_user == Namespace.id))
                     .switch(RepositoryTag)
                     .join(Image)
                     .where(Repository.name == repository_name,
                            Namespace.username == namespace_name))

  if not include_hidden:
    query = query.where(RepositoryTag.hidden == False)

  if include_storage:
    query = query.switch(Image).join(ImageStorage)

  return query


def create_or_update_tag(namespace_name, repository_name, tag_name,
                         tag_docker_image_id, reversion=False):
  try:
    repo = _basequery.get_existing_repository(namespace_name, repository_name)
  except Repository.DoesNotExist:
    raise DataModelException('Invalid repository %s/%s' % (namespace_name, repository_name))

  now_ts = get_epoch_timestamp()

  with db_transaction():
    try:
      tag = db_for_update(_tag_alive(RepositoryTag
                                     .select()
                                     .where(RepositoryTag.repository == repo,
                                            RepositoryTag.name == tag_name), now_ts)).get()
      tag.lifetime_end_ts = now_ts
      tag.save()
    except RepositoryTag.DoesNotExist:
      pass

    try:
      image_obj = Image.get(Image.docker_image_id == tag_docker_image_id, Image.repository == repo)
    except Image.DoesNotExist:
      raise DataModelException('Invalid image with id: %s' % tag_docker_image_id)

    return RepositoryTag.create(repository=repo, image=image_obj, name=tag_name,
                                lifetime_start_ts=now_ts, reversion=reversion)


def create_temporary_hidden_tag(repo, image_obj, expiration_s):
  """ Create a tag with a defined timeline, that will not appear in the UI or CLI. Returns the name
      of the temporary tag. """
  now_ts = get_epoch_timestamp()
  expire_ts = now_ts + expiration_s
  tag_name = str(uuid4())
  RepositoryTag.create(repository=repo, image=image_obj, name=tag_name, lifetime_start_ts=now_ts,
                       lifetime_end_ts=expire_ts, hidden=True)
  return tag_name


def delete_tag(namespace_name, repository_name, tag_name):
  now_ts = get_epoch_timestamp()
  with db_transaction():
    try:
      query = _tag_alive(RepositoryTag
                         .select(RepositoryTag, Repository)
                         .join(Repository)
                         .join(Namespace, on=(Repository.namespace_user == Namespace.id))
                         .where(Repository.name == repository_name,
                                Namespace.username == namespace_name,
                                RepositoryTag.name == tag_name), now_ts)
      found = db_for_update(query).get()
    except RepositoryTag.DoesNotExist:
      msg = ('Invalid repository tag \'%s\' on repository \'%s/%s\'' %
             (tag_name, namespace_name, repository_name))
      raise DataModelException(msg)

    found.lifetime_end_ts = now_ts
    found.save()


def garbage_collect_tags(repo):
  expired_time = get_epoch_timestamp() - repo.namespace_user.removed_tag_expiration_s

  tags_to_delete = list(RepositoryTag
                        .select(RepositoryTag.id)
                        .where(RepositoryTag.repository == repo,
                               ~(RepositoryTag.lifetime_end_ts >> None),
                               (RepositoryTag.lifetime_end_ts <= expired_time))
                        .order_by(RepositoryTag.id))
  if len(tags_to_delete) > 0:
    (RepositoryTag
     .delete()
     .where(RepositoryTag.id << tags_to_delete)
     .execute())


def get_tag_image(namespace_name, repository_name, tag_name):
  def limit_to_tag(query):
    return _tag_alive(query
                      .switch(Image)
                      .join(RepositoryTag)
                      .where(RepositoryTag.name == tag_name))

  images = image.get_repository_images_base(namespace_name, repository_name, limit_to_tag)
  if not images:
    raise DataModelException('Unable to find image for tag.')
  else:
    return images[0]


def list_repository_tag_history(repo_obj, page=1, size=100, specific_tag=None):
  query = (RepositoryTag
           .select(RepositoryTag, Image)
           .join(Image)
           .where(RepositoryTag.repository == repo_obj)
           .where(RepositoryTag.hidden == False)
           .order_by(RepositoryTag.lifetime_start_ts.desc())
           .order_by(RepositoryTag.name)
           .paginate(page, size))

  if specific_tag:
    query = query.where(RepositoryTag.name == specific_tag)

  return query


def revert_tag(repo_obj, tag_name, docker_image_id):
  """ Reverts a tag to a specific image ID. """
  # Verify that the image ID already existed under this repository under the
  # tag.
  try:
    (RepositoryTag
     .select()
     .join(Image)
     .where(RepositoryTag.repository == repo_obj)
     .where(RepositoryTag.name == tag_name)
     .where(Image.docker_image_id == docker_image_id)
     .get())
  except RepositoryTag.DoesNotExist:
    raise DataModelException('Cannot revert to unknown or invalid image')

  return create_or_update_tag(repo_obj.namespace_user.username, repo_obj.name, tag_name,
                              docker_image_id, reversion=True)