136 lines
5.5 KiB
Python
136 lines
5.5 KiB
Python
# 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, Label
|
|
|
|
|
|
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,
|
|
include_legacy_image=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
|
|
|
|
try:
|
|
tag_manifest = model.tag.load_manifest_by_digest(repo.namespace_user.username,
|
|
repo.name,
|
|
manifest_digest,
|
|
allow_dead=allow_dead)
|
|
except model.tag.InvalidManifestException:
|
|
return None
|
|
|
|
legacy_image = None
|
|
if include_legacy_image:
|
|
legacy_image = self.get_legacy_image(repository_ref, tag_manifest.tag.image.docker_image_id,
|
|
include_parents=True)
|
|
|
|
return Manifest.for_tag_manifest(tag_manifest, legacy_image)
|
|
|
|
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)
|
|
|
|
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 None
|
|
|
|
label = model.label.create_manifest_label(tag_manifest, key, value, source_type_name,
|
|
media_type_name)
|
|
return Label.for_label(label)
|
|
|
|
def list_manifest_labels(self, manifest, key_prefix=None):
|
|
""" Returns all labels found on the manifest. If specified, the key_prefix will filter the
|
|
labels returned to those keys that start with the given prefix.
|
|
"""
|
|
labels = model.label.list_manifest_labels(manifest._db_id, prefix_filter=key_prefix)
|
|
return [Label.for_label(l) for l in labels]
|
|
|
|
def get_manifest_label(self, manifest, label_uuid):
|
|
""" Returns the label with the specified UUID on the manifest or None if none. """
|
|
return Label.for_label(model.label.get_manifest_label(label_uuid, manifest._db_id))
|
|
|
|
def delete_manifest_label(self, manifest, label_uuid):
|
|
""" Delete the label with the specified UUID on the manifest. Returns the label deleted
|
|
or None if none.
|
|
"""
|
|
return Label.for_label(model.label.delete_manifest_label(label_uuid, manifest._db_id))
|
|
|
|
pre_oci_model = PreOCIModel()
|