zip-extract-on-the-fly.py

# -*- coding:utf-8 -*-
import zipfile
import os.path

# 対象となるファイル
with open("hello.txt", "w") as wf:
    wf.write("hello world")


# zip圧縮する
with zipfile.ZipFile("hello.zip", "w") as zf:
    zf.write("hello.txt", "hello.txt")

assert os.path.exists("hello.zip")


# 普通に展開する
with zipfile.ZipFile("hello.zip") as zf:
    with zf.open("hello.txt") as rf:
        assert b'hello world' == rf.read()


# Network越しに直接渡ってきた場合
import threading
from http import server as s
import contextlib


def make_server(port=6543):
    server_address = ('', port)
    s.HTTPServer.protocol_version = "HTTP/1.0"
    httpd = s.HTTPServer(server_address, s.SimpleHTTPRequestHandler)
    return httpd


@contextlib.contextmanager
def with_http_context():
    server = make_server(6543)
    t = threading.Thread(target=server.handle_request)  # not serve_forever
    try:
        t.start()
        yield
    except Exception as e:
        print("failure: {e}".format(e=e))
    finally:
        t.join()


import urllib.request

with with_http_context():
    with urllib.request.urlopen("http://localhost:6543/hello.zip") as rf:
        # seek()を持っていないので失敗する。
        with zipfile.ZipFile(rf) as zf:
            with zf.open("hello.zip") as r:
                print(r.read())


# Network越しに直接渡ってきた場合にBytesIOに一度結果をまとめる

import urllib.request
from io import BytesIO

with with_http_context():
    with urllib.request.urlopen("http://localhost:6543/hello.zip") as rf:
        with zipfile.ZipFile(BytesIO(rf.read())) as zf:
            with zf.open("hello.txt") as r:
                assert b"hello world" == r.read()