ネットワーク越しに送られてくるzipfileを一時ファイルを作成せずに展開した結果が欲しかった。

python

話す内容

  • http requestのresponseがzipファイルで返ってきていた。
  • pythonのzipfileモジュールはseek()が無いfile like objectに対応していない
  • 一時ファイル作るのだるい。
  • どうしよう。

そういう感じで、ネットワーク越しに返ってきたzipファイルを直接展開した結果にアクセス出来ないかなーと思いました。

とりあえず、以下の様にして対象となるzipファイルを作成

# 対象となるファイル
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()

困ったのはネットワーク越しにやってくる時。

ネットワーク越しにやってくることを再現するために、裏側にhttpserverを立ててアクセスする感じのことをしています。(with_http_context)

# 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())

io.BytesIOに渡してあげれば良さそうです

$ python3.3
>>> from io import BytesIO
>>> hasattr(BytesIO(), "seek")
True

一度BytesIOに渡してあげると、BytesIOはseek()を持っているのでzipfile.ZipFileも使える感じになりそうです。

# 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()

https://gist.github.com/podhmo/54254d890811f682a610