quay/data/registry_model/registry_pre_oci_model.py

# pylint: disable=protected-access

from collections import defaultdict

from data import database
from data import model
from data.registry_model.interface import RegistryDataInterface
from data.registry_model.datatypes import Tag, RepositoryReference, Manifest, LegacyImage


class PreOCIModel(RegistryDataInterface):
  """
  PreOCIModel implements the data model for the registry API using a database schema
  before it was changed to support the OCI specification.
  """

  def find_matching_tag(self, repository_ref, tag_names):
    """ Finds an alive tag in the repository matching one of the given tag names and returns it
        or None if none.
    """
    found_tag = model.tag.find_matching_tag(repository_ref._db_id, tag_names)
    return Tag.for_repository_tag(found_tag)

  def get_most_recent_tag(self, repository_ref):
    """ Returns the most recently pushed alive tag in the repository, if any. If none, returns
        None.
    """
    found_tag = model.tag.get_most_recent_tag(repository_ref._db_id)
    return Tag.for_repository_tag(found_tag)

  def lookup_repository(self, namespace_name, repo_name, kind_filter=None):
    """ Looks up and returns a reference to the repository with the given namespace and name,
        or None if none. """
    repo = model.repository.get_repository(namespace_name, repo_name, kind_filter=kind_filter)
    return RepositoryReference.for_repo_obj(repo)

  def get_manifest_for_tag(self, tag):
    """ Returns the manifest associated with the given tag. """
    try:
      tag_manifest = database.TagManifest.get(tag_id=tag._db_id)
    except database.TagManifest.DoesNotExist:
      return

    return Manifest.for_tag_manifest(tag_manifest)

  def lookup_manifest_by_digest(self, repository_ref, manifest_digest, allow_dead=False):
    """ Looks up the manifest with the given digest under the given repository and returns it
        or None if none. """
    repo = model.repository.lookup_repository(repository_ref._db_id)
    if repo is None:
      return None

    tag_manifest = model.tag.load_manifest_by_digest(repo.namespace_user.username,
                                                     repo.name,
                                                     manifest_digest, allow_dead=allow_dead)
    return Manifest.for_tag_manifest(tag_manifest)

  def create_manifest_label(self, manifest, key, value, source_type_name, media_type_name=None):
    """ Creates a label on the manifest with the given key and value. """
    try:
      tag_manifest = database.TagManifest.get(id=manifest._db_id)
    except database.TagManifest.DoesNotExist:
      return

    model.label.create_manifest_label(tag_manifest, key, value, source_type_name, media_type_name)

  def get_legacy_images(self, repository_ref):
    """
    Returns an iterator of all the LegacyImage's defined in the matching repository.
    """
    repo = model.repository.lookup_repository(repository_ref._db_id)
    if repo is None:
      return None

    all_images = model.image.get_repository_images_without_placements(repo)
    all_images_map = {image.id: image for image in all_images}

    all_tags = model.tag.list_repository_tags(repo.namespace_user.username, repo.name)
    tags_by_image_id = defaultdict(list)
    for tag in all_tags:
      tags_by_image_id[tag.image_id].append(tag)

    return [LegacyImage.for_image(image, images_map=all_images_map, tags_map=tags_by_image_id)
            for image in all_images]

  def get_legacy_image(self, repository_ref, docker_image_id, include_parents=False):
    """
    Returns the matching LegacyImages under the matching repository, if any. If none,
    returns None.
    """
    repo = model.repository.lookup_repository(repository_ref._db_id)
    if repo is None:
      return None

    image = model.image.get_image(repository_ref._db_id, docker_image_id)
    if image is None:
      return None

    parent_images_map = None
    if include_parents:
      parent_images = model.image.get_parent_images(repo.namespace_user.username, repo.name, image)
      parent_images_map = {image.id: image for image in parent_images}

    return LegacyImage.for_image(image, images_map=parent_images_map)


pre_oci_model = PreOCIModel()