from app import app, storage as store from data import model from endpoints.v1.models_interface import DockerRegistryV1DataInterface, Repository from util.morecollections import AttrDict class PreOCIModel(DockerRegistryV1DataInterface): """ PreOCIModel implements the data model for the v1 Docker Registry protocol using a database schema before it was changed to support the OCI specification. """ def placement_locations_and_path_docker_v1(self, namespace_name, repo_name, image_id): image, placements = model.image.get_image_and_placements(namespace_name, repo_name, image_id) if image is None: return None, None locations = {placement.location.name for placement in placements} return locations, model.storage.get_layer_path(image.storage) def docker_v1_metadata(self, namespace_name, repo_name, image_id): repo_image = model.image.get_repo_image(namespace_name, repo_name, image_id) if repo_image is None: return None return AttrDict({ 'namespace_name': namespace_name, 'repo_name': repo_name, 'image_id': image_id, 'checksum': repo_image.v1_checksum, 'compat_json': repo_image.v1_json_metadata,}) def update_docker_v1_metadata(self, namespace_name, repo_name, image_id, created_date_str, comment, command, compat_json, parent_image_id=None): parent_image = None if parent_image_id is not None: parent_image = model.image.get_repo_image(namespace_name, repo_name, parent_image_id) model.image.set_image_metadata(image_id, namespace_name, repo_name, created_date_str, comment, command, compat_json, parent=parent_image) def storage_exists(self, namespace_name, repo_name, image_id): repo_image = model.image.get_repo_image_and_storage(namespace_name, repo_name, image_id) if repo_image is None or repo_image.storage is None: return False if repo_image.storage.uploading: return False layer_path = model.storage.get_layer_path(repo_image.storage) return store.exists(repo_image.storage.locations, layer_path) def store_docker_v1_checksums(self, namespace_name, repo_name, image_id, checksum, content_checksum): repo_image = model.image.get_repo_image_and_storage(namespace_name, repo_name, image_id) if repo_image is None or repo_image.storage is None: return with model.db_transaction(): repo_image.storage.content_checksum = content_checksum repo_image.v1_checksum = checksum repo_image.storage.save() repo_image.save() def is_image_uploading(self, namespace_name, repo_name, image_id): repo_image = model.image.get_repo_image_and_storage(namespace_name, repo_name, image_id) if repo_image is None or repo_image.storage is None: return False return repo_image.storage.uploading def update_image_uploading(self, namespace_name, repo_name, image_id, is_uploading): repo_image = model.image.get_repo_image_and_storage(namespace_name, repo_name, image_id) if repo_image is None or repo_image.storage is None: return repo_image.storage.uploading = is_uploading repo_image.storage.save() return repo_image.storage def update_image_sizes(self, namespace_name, repo_name, image_id, size, uncompressed_size): model.storage.set_image_storage_metadata(image_id, namespace_name, repo_name, size, uncompressed_size) def get_image_size(self, namespace_name, repo_name, image_id): repo_image = model.image.get_repo_image_and_storage(namespace_name, repo_name, image_id) if repo_image is None or repo_image.storage is None: return None return repo_image.storage.image_size def create_bittorrent_pieces(self, namespace_name, repo_name, image_id, pieces_bytes): repo_image = model.image.get_repo_image_and_storage(namespace_name, repo_name, image_id) if repo_image is None or repo_image.storage is None: return model.storage.save_torrent_info(repo_image.storage, app.config['BITTORRENT_PIECE_SIZE'], pieces_bytes) def image_ancestry(self, namespace_name, repo_name, image_id): try: image = model.image.get_image_by_id(namespace_name, repo_name, image_id) except model.InvalidImageException: return None parents = model.image.get_parent_images(namespace_name, repo_name, image) ancestry_docker_ids = [image.docker_image_id] ancestry_docker_ids.extend([parent.docker_image_id for parent in parents]) return ancestry_docker_ids def repository_exists(self, namespace_name, repo_name): repo = model.repository.get_repository(namespace_name, repo_name) return repo is not None def create_or_link_image(self, username, namespace_name, repo_name, image_id, storage_location): repo = model.repository.get_repository(namespace_name, repo_name) model.image.find_create_or_link_image(image_id, repo, username, {}, storage_location) def create_temp_hidden_tag(self, namespace_name, repo_name, image_id, expiration): repo_image = model.image.get_repo_image(namespace_name, repo_name, image_id) if repo_image is None: return repo = repo_image.repository model.tag.create_temporary_hidden_tag(repo, repo_image, expiration) def list_tags(self, namespace_name, repo_name): return model.tag.list_repository_tags(namespace_name, repo_name) def create_or_update_tag(self, namespace_name, repo_name, image_id, tag_name): model.tag.create_or_update_tag(namespace_name, repo_name, tag_name, image_id) def find_image_id_by_tag(self, namespace_name, repo_name, tag_name): try: tag_image = model.tag.get_tag_image(namespace_name, repo_name, tag_name) except model.DataModelException: return None return tag_image.docker_image_id def delete_tag(self, namespace_name, repo_name, tag_name): model.tag.delete_tag(namespace_name, repo_name, tag_name) def change_user_password(self, user, new_password): model.user.change_password(user, new_password) def get_repository(self, namespace_name, repo_name): repo = model.repository.get_repository(namespace_name, repo_name) if repo is None: return None return _repository_for_repo(repo) def create_repository(self, namespace_name, repo_name, user=None): model.repository.create_repository(namespace_name, repo_name, user) def repository_is_public(self, namespace_name, repo_name): return model.repository.repository_is_public(namespace_name, repo_name) def validate_oauth_token(self, token): return bool(model.oauth.validate_access_token(token)) def get_sorted_matching_repositories(self, search_term, filter_username=None, offset=0, limit=25): repos = model.repository.get_filtered_matching_repositories( search_term, filter_username=filter_username, offset=offset, limit=limit) return [_repository_for_repo(repo) for repo in repos] def is_namespace_enabled(self, namespace_name): namespace = model.user.get_namespace_user(namespace_name) return namespace is not None and namespace.enabled def _repository_for_repo(repo): """ Returns a Repository object representing the Pre-OCI data model instance of a repository. """ return Repository( id=repo.id, name=repo.name, namespace_name=repo.namespace_user.username, description=repo.description, is_public=model.repository.is_repository_public(repo), kind=model.repository.get_repo_kind_name(repo),) pre_oci_model = PreOCIModel()