py-registry.py

import json,base64
import requests, re
from collections import Counter

class DockerRegistryClient(object):
    def __init__(self, registry="http://reg.xxx.xxx:5000", auth='xxx:xxxxx', api_timeout=None):
        self.registry = registry
        self.auth = auth
        self.method_kwargs = {}
        self.scheme = 'application/vnd.docker.distribution.manifest.v2+json'

        if api_timeout is not None:
            self.method_kwargs['timeout'] = api_timeout

    def _http_response(self, url, method=requests.get, data=None, content_type=None,
                       schema=None, **kwargs):
        """url -> full target url
           method -> method from requests
           data -> request body
           kwargs -> url formatting args
        """
        if schema is None:
            schema = self.scheme
        header = {
            'content-type': content_type or 'application/json',
            'Accept': schema,
            "Authorization": 'Basic %s' % base64.b64encode(
                    self.auth.encode('utf-8')
                ).decode('ascii')
                ,
            }

        if data:
            data = json.dumps(data)
        path = url.format(**kwargs)
        response = method(self.registry + path,
                          data=data, headers=header, **self.method_kwargs)
        response.raise_for_status()

        return response

    def _http_call(self, url, method=requests.get, data=None, **kwargs):
        """url -> full target url
           method -> method from requests
           data -> request body
           kwargs -> url formatting args
        """
        c = Counter()   
        # header 里可能会返回下一页的url参数
        # Link: </v2/_catalog?last=wps%2Fweboffice-storage-provider&n=100>; rel="next"
        while True:
            response = self._http_response(url, method, data=data, **kwargs)
            if not response.content:
                return {}

            c.update(response.json())

            if 'Link' in response.headers:
                url = response.headers['Link'].split(';')[0][1:-1]
                continue

            break
        return dict(c)

    def get_repository(self):
        """GET /v2/_catalog"""
        url = '/v2/_catalog'
        data = self._http_call(url, method=requests.get)
        return data['repositories'] or []

    def get_tags(self, repository, return_with_name=False):
        """repository -> repos []string
        """
        url = '/v2/%s/tags/list'
        data = []
        for repo in repository:
            resp_data = self._http_call(url % repo, method=requests.get)['tags']
            if not resp_data: # 删掉 manifest 下 tags 为 null {"name":"test/pause","tags":null}
                continue
            if return_with_name:
                resp_data = ['%s:%s' % (repo, x) for x in resp_data ]
            data.extend(resp_data)

        return data
    def get_all_from_registry(self, re_pattern=''):
        repos = self.get_repository()
        data = self.get_tags(repos, return_with_name=True)
        if re_pattern:
            prog = re.compile(re_pattern)
            return [x for x in data if prog.search(x)]
        return data

    def get_manifest(self, name):
        img_part = name.split(':')
        if len(img_part) != 2:
            return ''
        url = '/v2/%s/manifests/%s'

        data = self._http_response(url % (img_part[0], img_part[1]))
        return data.headers['Docker-Content-Digest'] or ''

    def delete_manifest(self, img_name, manifest):
        """传入镜像的sha256删除镜像
        """
        url = '/v2/%s/manifests/%s' % (img_name, manifest)
        response = self._http_response(url, method=requests.delete)
        return response.status_code == 202

    def get_layer(self, name):
        img_part = name.split(':')
        if len(img_part) != 2:
            return ''
        url = '/v2/%s/manifests/%s'

        data = self._http_call(url % (img_part[0], img_part[1]))
        return data['layers'] or []

    def delete_blobs(self, img_name, manifest):
        """传入镜像的sha256删除镜像
        """
        url = '/v2/%s/blobs/%s' % (img_name, manifest)
        response = self._http_response(url, method=requests.delete)
        return response.status_code == 202

    def delete_image(self, image):
        # 删掉层会导致同层的其他名字镜像给gc掉
        img_name_without_tag = image.split(':')[0]
        layers = self.get_layer(image)
        if layers:
            for layer in layers:
                self.delete_blobs(img_name_without_tag, layer['digest'])
            digest = self.get_manifest(image)
            self.delete_manifest(img_name_without_tag, digest)


c = DockerRegistryClient()
print(c.get_repository())
print(c.get_tags(['coredns'], return_with_name=True))
print(c.get_all_from_registry('5.1.7'))
digest = c.get_manifest('test/pause:3.4')
print(c.delete_image('test/pause:3.4'))