from . import BaseApiClient, _requests_formatter
import dockerreg.models.v2 as modelsv2
from dockerreg.util.applicable \
import ApplicableClass, ApplicableMethod
import dockerreg.exceptions as ex
from dockerreg.log import LOG
[docs]class RepositoryV2Mixin(object):
@ApplicableMethod(alias="v2.get_tags",formatter=_requests_formatter)
@ApplicableMethod(alias="v2.get_reference_metadata",
formatter=_requests_formatter)
@ApplicableMethod(alias="v2.get_manifest",formatter=_requests_formatter)
[docs] def get_manifest(self,repository,reference,raw=False):
resp = self.get("/%s/manifests/%s" % (repository,reference))
if raw:
return resp
return resp.json()
@ApplicableMethod(alias="v2.put_manifest",formatter=_requests_formatter)
[docs] def put_manifest(self,repository,reference,manifest,media_type=None):
if not media_type:
raise ex.IllegalArgumentError(
"the manifest media_type must be specified")
resp = self.put("/%s/manifests/%s" % (repository,reference),
data=manifest,headers={'Content-type':media_type})
return resp
@ApplicableMethod(alias="v2.delete_manifest_by_tag",
formatter=_requests_formatter)
[docs] def delete_manifest_by_tag(self,repository,tag,raw=False):
# We can only delete manifests by a specific media_type,
# apparently, in newer versions of the docker registry.
# https://github.com/docker/distribution/issues/1637
# https://docs.docker.com/registry/compatibility/#content-addressable-storage-cas
mt = 'application/vnd.docker.distribution.manifest.v2+json'
resp = self.head("/%s/manifests/%s" % (repository,tag),
headers={'Accept':mt})
dcd = resp.headers.get("docker-content-digest")
resp = self.delete("/%s/manifests/%s" % (repository,dcd))
if raw:
return resp
return resp.content
@ApplicableMethod(alias="v2.delete_manifest",formatter=_requests_formatter)
[docs] def delete_manifest(self,repository,reference,raw=False):
resp = self.delete("/%s/manifests/%s" % (repository,reference))
if raw:
return resp
return resp.content
@ApplicableMethod(alias="v2.get_blob",formatter=_requests_formatter)
[docs] def get_blob(self,repository,digest,raw=False):
resp = self.get("/%s/blobs/%s" % (repository,digest))
if raw:
return resp
return resp.content
@ApplicableMethod(alias="v2.get_blob_size",formatter=_requests_formatter)
[docs] def get_blob_size(self,repository,digest):
resp = self.head("/%s/blobs/%s" % (repository,digest))
if resp.status_code == 200:
return int(resp.headers["content-length"])
else:
self._handle_http_error(resp)
@ApplicableMethod(alias="v2.post_blob",formatter=_requests_formatter)
[docs] def post_blob(self,repository,digest,data,raw=False):
#
# POSTing to uploads/ in one shot does not seem to work. So we
# take the alternate "chunking" approach. Sadly, the one-chunk
# approach appears not to work; there seems to be some chunk
# size limit. So we are committed to the real chunking
# strategy, which is slower.
#
CHUNKSIZE = 2**26
resp = self.post(
"/%s/blobs/uploads/" % (repository,)) #params={"digest":digest})
#headers={"Content-Type":"application/octet-stream",
# "Content-Length": '%d' % len(data)})
if resp.status_code == 202:
authhdr = None
authvalue = self.get_cached_auth(
dict(scope="repository:%s:pull,push" % (repository,)))
if authvalue:
LOG.debug("will used cached %s" % (str(authvalue)))
authhdr = "%s %s" % (authvalue.scheme(),authvalue.token())
uuid = resp.headers.get("Docker-Upload-UUID")
location = resp.headers.get("Location")
if not location:
location = "/%s/blobs/uploads/%s" % (repository,uuid)
length = len(data)
start = end = 0
while end < length:
if (start + CHUNKSIZE) > length:
end = length
else:
end = start + CHUNKSIZE
hdrs = { "Content-Type":"application/octet-stream",
"Content-Length": '%d' % (end - start),
"Content-Range": '%d-%d' % (start,end) }
if authhdr:
hdrs["Authorization"] = authhdr
resp = self.patch(
location,url_raw=True,data=data[start:end],headers=hdrs)
location = resp.headers.get("location")
start = end
#location = "/%s/blobs/uploads/" % (repository,)
resp = self.put(
location,url_raw=True,params={"digest":digest},
headers={"Content-Type":"application/octet-stream",
"Content-Length": '0'})
else:
raise ex.DockerRegistryAPIError(
resp.status_code,
errors=['error in single chunk blob upload process'])
if raw:
return resp
return resp.content
@ApplicableMethod(alias="v2.delete_blob",formatter=_requests_formatter)
[docs] def delete_blob(self,repository,digest,raw=False):
resp = self.delete("/%s/blobs/%s" % (repository,digest))
if raw:
return resp
return resp.json()
[docs]class RegistryV2Mixin(object):
@ApplicableMethod(alias="v2.get_repositories",formatter=_requests_formatter)
[docs] def get_repositories(self):
return self.get("/_catalog").json()
@ApplicableClass()
[docs]class ApiClientV2(BaseApiClient,RegistryV2Mixin,RepositoryV2Mixin):
[docs] def __init__(self,host,url_prefix=None,username=None,
auth=None,auth_url=None,verify=True,cert=None,
cache=None,**kwargs):
super(ApiClientV2,self).__init__(
host,url_prefix=url_prefix,username=username,
auth=auth,auth_url=auth_url,verify=verify,cert=cert,
cache=cache,version=2,**kwargs)
[docs] def registry(self):
return modelsv2.RegistryV2(self,name=self.host)