vbatts/quay
Archived
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

endpoints.appr: move to new v22 format

Browse source
This commit is contained in:
Jimmy Zelinskie 2017-07-25 17:41:55 -07:00
parent c271b1f386
commit 155cb65f7d
9 changed files with 279 additions and 251 deletions
Download patch file Download diff file Expand all files Collapse all files

49
endpoints/appr/cnr_backend.py
View file

@ -7,8 +7,7 @@ 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
from data.oci_model import blob # TODO these calls should be through oci_app_model
from endpoints.appr.models_oci import model
class Blob(BlobBase):
@ -17,7 +16,7 @@ class Blob(BlobBase):
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)
model.store_blob(self, content_media_type)
@classmethod
def delete(cls, package_name, digest):
@ -26,7 +25,7 @@ class Blob(BlobBase):
@classmethod
def _fetch_b64blob(cls, package_name, digest):
blobpath = cls.upload_url(digest)
locations = blob.get_blob_locations(digest)
locations = model.get_blob_locations(digest)
if not locations:
raise_package_not_found(package_name, digest)
return base64.b64encode(storage.get_content(locations, blobpath))
@ -34,7 +33,7 @@ class Blob(BlobBase):
@classmethod
def download_url(cls, package_name, digest):
blobpath = cls.upload_url(digest)
locations = blob.get_blob_locations(digest)
locations = model.get_blob_locations(digest)
if not locations:
raise_package_not_found(package_name, digest)
return storage.get_direct_download_url(locations, blobpath)
@ -49,29 +48,29 @@ class Channel(ChannelBase):
def _exists(self):
""" Check if the channel is saved already """
return oci_app_model.channel_exists(self.package, self.name)
return model.channel_exists(self.package, self.name)
@classmethod
def get(cls, name, package):
chanview = oci_app_model.fetch_channel(package, name, with_releases=False)
chanview = 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)
model.update_channel(self.package, self.name, self.current)
def delete(self):
oci_app_model.delete_channel(self.package, self.name)
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)
Channel(c.name, package_name, c.current) for c in model.list_channels(package_name)
]
@property
def _channel(self):
if self._channel_data is None:
self._channel_data = oci_app_model.fetch_channel(self.package, self.name)
self._channel_data = model.fetch_channel(self.package, self.name)
return self._channel_data
def releases(self):
@ -79,10 +78,10 @@ class Channel(ChannelBase):
return self._channel.releases
def _add_release(self, release):
return oci_app_model.update_channel(self.package, self.name, release)._asdict
return model.update_channel(self.package, self.name, release)._asdict
def _remove_release(self, release):
oci_app_model.delete_channel(self.package, self.name)
model.delete_channel(self.package, self.name)
class User(object):
@ -91,7 +90,7 @@ class User(object):
@classmethod
def get_user(cls, username, password):
""" Returns True if user creds is valid """
return oci_app_model.get_user(username, password)
return model.get_user(username, password)
class Package(PackageBase):
@ -110,55 +109,55 @@ class Package(PackageBase):
@classmethod
def create_repository(cls, package_name, visibility, owner):
oci_app_model.create_application(package_name, visibility, owner)
model.create_application(package_name, visibility, owner)
@classmethod
def exists(cls, package_name):
return oci_app_model.application_exists(package_name)
return 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,
for x in model.list_applications(namespace=organization, media_type=media_type,
search=search, username=username)
]
@classmethod
def _fetch(cls, package_name, release, media_type):
data = oci_app_model.fetch_release(package_name, release, manifest_media_type(media_type))
data = 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)
return model.list_releases(package_name, media_type)
@classmethod
def search(cls, query, username=None):
return oci_app_model.basic_search(query, username=username)
return 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)
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))
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)
return 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)
for c in model.list_release_channels(self.package, self.release, active=iscurrent)
]
@classmethod
def manifests(cls, package, release=None):
return oci_app_model.list_manifests(package, release)
return model.list_manifests(package, release)
@classmethod
def dump_all(cls, blob_cls):