remove unused: class DockerImage, class StdoutTee & func os_read_non_blocking()

rc.xsh

@@ -1,6 +1,5 @@
 import contextlib
 import datetime as dt
-import io
 import os
 import re
 import shutil
@@ -92,97 +91,6 @@ $PASSWORD_STORE_UMASK = '027'
 $PASSWORD_STORE_CHARACTER_SET_NO_SYMBOLS = '1-9a-km-zA-HJKLMNPR-Z'
 $PASSWORD_STORE_CHARACTER_SET = $PASSWORD_STORE_CHARACTER_SET_NO_SYMBOLS + '*+!&#@%.\-_'
 
-class DockerImage:
-
-    def __init__(self, image):
-        import json
-        attrs, = json.loads(subprocess.check_output(['sudo', 'docker', 'image', 'inspect', image])
-                            .decode(sys.stdout.encoding))
-        self._id = attrs['Id']
-        self._tags = attrs['RepoTags']
-
-    def __repr__(self):
-        return '{}(id={!r}, tags={!r})'.format(type(self).__name__, self._id, self._tags)
-
-    @classmethod
-    def build(cls, dockerfile_or_path, tag=None):
-        out = io.BytesIO()
-        args = ['sudo', 'docker', 'build']
-        if tag:
-            args.extend(['--tag', tag])
-        with StdoutTee(out) as tee:
-            if os.path.exists(dockerfile_or_path):
-                p = subprocess.Popen(args + [dockerfile_or_path],
-                                     stdin=None, stdout=tee)
-            else:
-                p = subprocess.Popen(args + ['-'],
-                                     stdin=subprocess.PIPE, stdout=tee)
-                p.stdin.write(dockerfile_or_path.encode())
-                p.stdin.close()
-            assert p.wait() == 0, 'docker build failed'
-        image_id, = re.search(rb'^Successfully built (\S+)$', out.getvalue(), re.MULTILINE).groups()
-        return cls(image_id.decode(sys.stdout.encoding))
-
-    @classmethod
-    def pull(cls, image):
-        out = io.BytesIO()
-        with StdoutTee(out) as tee:
-            subprocess.run(['sudo', 'docker', 'image', 'pull', image], stdout=tee)
-        repo_digest, = re.search(rb'^Digest: (sha\S+:\S+)$', out.getvalue(), re.MULTILINE).groups()
-        return cls('{}@{}'.format(image, repo_digest.decode()))
-
-    def run(self, args=[], name=None, detach=False, env={},
-            network=None, publish_ports=[], volumes=[], caps=[]):
-        params = ['sudo', 'docker', 'run', '--rm']
-        if name:
-            params.extend(['--name', name])
-        if detach:
-            params.append('--detach')
-        else:
-            params.extend(['--interactive', '--tty'])
-        params.extend([a for k, v in env.items() for a in ['--env', '{}={}'.format(k, v)]])
-        if network:
-            params.extend(['--network', network])
-        params.extend([a for v in volumes for a in ['--volume', ':'.join(v)]])
-        params.extend('--publish=' + ':'.join([str(a) for a in p]) for p in publish_ports)
-        params.extend(['--security-opt=no-new-privileges', '--cap-drop=all'])
-        params.extend(['--cap-add={}'.format(c) for c in caps])
-        params.append(self._id)
-        params.extend(args)
-        sys.stderr.write('{}\n'.format(shlex_join(params)))
-        subprocess.run(params, check=True)
-
-class StdoutTee:
-
-    def __init__(self, sink):
-        self._sink = sink
-
-    def __enter__(self):
-        self._read_fd, self._write_fd = os.pipe()
-        import threading
-        self._thread = threading.Thread(target=self._loop)
-        self._thread.start()
-        return self
-
-    def _loop(self):
-        while True:
-            try:
-                data = os_read_non_blocking(self._read_fd)
-            except OSError:  # fd closed
-                return
-            if data:
-                self._sink.write(data)
-                sys.stdout.buffer.write(data)
-                sys.stdout.flush()
-
-    def fileno(self):
-        return self._write_fd
-
-    def __exit__(self, exc_type, exc_value, traceback):
-        os.close(self._read_fd)
-        os.close(self._write_fd)
-        self._thread.join()
-
 @contextlib.contextmanager
 def chdir(path):
     import shlex
@@ -244,13 +152,6 @@ def fuse_mount(mount_arg_patterns, mount_point_path):
         sys.stderr.write('{}\n'.format(shlex_join(umount_args)))
         subprocess.check_call(umount_args)
 
-def os_read_non_blocking(fd, buffer_size_bytes=8*1024, timeout_seconds=0.1):
-    import select
-    if fd in select.select([fd], [], [], timeout_seconds)[0]:
-        return os.read(fd, buffer_size_bytes)
-    else:
-        return None
-
 def shlex_join(params):
     import shlex
     assert isinstance(params, list) or isinstance(params, tuple), params