This repository has been archived on 2020-03-24. You can view files and clone it, but cannot push or open issues or pull requests.
quay/endpoints/appr/cnr_backend.py

177 lines
5.2 KiB
Python
Raw Normal View History

2017-03-23 01:52:45 +00:00
import base64
from cnr.exception import raise_package_not_found
from cnr.models.blob_base import BlobBase
from cnr.models.channel_base import ChannelBase
from cnr.models.db_base import CnrDB
from cnr.models.package_base import PackageBase, manifest_media_type
from app import storage
from data.interfaces.appr import oci_app_model
2017-04-13 12:25:47 +00:00
from data.oci_model import blob # TODO these calls should be through oci_app_model
2017-03-23 01:52:45 +00:00
class Blob(BlobBase):
@classmethod
def upload_url(cls, digest):
return "cnr/blobs/sha256/%s/%s" % (digest[0:2], digest)
def save(self, content_media_type):
oci_app_model.store_blob(self, content_media_type)
@classmethod
def delete(cls, package_name, digest):
pass
@classmethod
def _fetch_b64blob(cls, package_name, digest):
blobpath = cls.upload_url(digest)
locations = blob.get_blob_locations(digest)
if not locations:
raise_package_not_found(package_name, digest)
return base64.b64encode(storage.get_content(locations, blobpath))
@classmethod
def download_url(cls, package_name, digest):
blobpath = cls.upload_url(digest)
locations = blob.get_blob_locations(digest)
if not locations:
raise_package_not_found(package_name, digest)
return storage.get_direct_download_url(locations, blobpath)
class Channel(ChannelBase):
""" CNR Channel model implemented against the Quay data model. """
2017-03-23 01:52:45 +00:00
def __init__(self, name, package, current=None):
super(Channel, self).__init__(name, package, current=current)
self._channel_data = None
def _exists(self):
""" Check if the channel is saved already """
return oci_app_model.channel_exists(self.package, self.name)
@classmethod
def get(cls, name, package):
chanview = oci_app_model.fetch_channel(package, name, with_releases=False)
return cls(name, package, chanview.current)
def save(self):
oci_app_model.update_channel(self.package, self.name, self.current)
def delete(self):
oci_app_model.delete_channel(self.package, self.name)
@classmethod
def all(cls, package_name):
return [
Channel(c.name, package_name, c.current) for c in oci_app_model.list_channels(package_name)
]
2017-03-23 01:52:45 +00:00
@property
def _channel(self):
if self._channel_data is None:
self._channel_data = oci_app_model.fetch_channel(self.package, self.name)
return self._channel_data
def releases(self):
""" Returns the list of versions """
return self._channel.releases
def _add_release(self, release):
return oci_app_model.update_channel(self.package, self.name, release)._asdict
def _remove_release(self, release):
oci_app_model.delete_channel(self.package, self.name)
2017-04-13 12:25:47 +00:00
class User(object):
""" User in CNR models """
@classmethod
def get_user(cls, username, password):
""" Returns True if user creds is valid """
return oci_app_model.get_user(username, password)
2017-03-23 01:52:45 +00:00
class Package(PackageBase):
""" CNR Package model implemented against the Quay data model. """
@classmethod
def _apptuple_to_dict(cls, apptuple):
return {
'release': apptuple.release,
'created_at': apptuple.created_at,
'digest': apptuple.manifest.digest,
'mediaType': apptuple.manifest.mediaType,
'package': apptuple.name,
'content': apptuple.manifest.content._asdict()
}
2017-03-23 01:52:45 +00:00
@classmethod
def create_repository(cls, package_name, visibility, owner):
oci_app_model.create_application(package_name, visibility, owner)
@classmethod
def exists(cls, package_name):
return oci_app_model.application_exists(package_name)
@classmethod
def all(cls, organization=None, media_type=None, search=None, username=None, **kwargs):
return [
dict(x._asdict())
for x in oci_app_model.list_applications(namespace=organization, media_type=media_type,
search=search, username=username)
]
2017-03-23 01:52:45 +00:00
@classmethod
def _fetch(cls, package_name, release, media_type):
data = oci_app_model.fetch_release(package_name, release, manifest_media_type(media_type))
return cls._apptuple_to_dict(data)
@classmethod
def all_releases(cls, package_name, media_type=None):
return oci_app_model.list_releases(package_name, media_type)
@classmethod
def search(cls, query, username=None):
return oci_app_model.basic_search(query, username=username)
def _save(self, force=False, **kwargs):
user = kwargs['user']
visibility = kwargs['visibility']
oci_app_model.create_release(self, user, visibility, force)
@classmethod
def _delete(cls, package_name, release, media_type):
oci_app_model.delete_release(package_name, release, manifest_media_type(media_type))
@classmethod
def isdeleted_release(cls, package, release):
return oci_app_model.release_exists(package, release)
def channels(self, channel_class, iscurrent=True):
return [
c.name
for c in oci_app_model.list_release_channels(self.package, self.release, active=iscurrent)
]
2017-03-23 01:52:45 +00:00
@classmethod
def manifests(cls, package, release=None):
return oci_app_model.list_manifests(package, release)
@classmethod
def dump_all(cls, blob_cls):
raise NotImplementedError
class QuayDB(CnrDB):
""" Wrapper Class to embed all CNR Models """
Channel = Channel
Package = Package
Blob = Blob
@classmethod
def reset_db(cls, force=False):
pass