使用Tornado将二进制文件流式传输到Google Storage

Posted 2021-04-21

我正在尝试通过我的服务器将二进制文件从客户端请求流式传输到Google云端存储。

我正在使用Tornado框架将请求中的数据流式传输到服务器，并使用Google Cloud Storage API将文件流式传输到Google -upload_from_file方法。

我是Tornado的新手，我正在使用@stream_request_body装饰器，所以我可以从块中获取请求中的数据并将每个块上传到Google。

我试图打开一个类似文件的对象，我可以将每个块写入，同时将“文件”本身上传到Google。

问题是我在开始编写块之前无法将“文件”上传到Google。

任何援助将不胜感激。

答案

使用Google的HTTP库执行此操作非常棘手，因为它们专为同步使用而设计。您需要将实际上载放在另一个线程上以避免阻止IOLoop。您可以使用os.pipe在Tornado线程和上传线程之间进行通信（将管道的写入端包装在Pipeiostream中，将读取端包装在os.fdopen中）。这是一个未经测试的解决方案草图：

def prepare(self):
    r, w = os.pipe()
    self.write_pipe = tornado.iostream.PipeIOStream(w)
    # Create our "file-like object" for upload_from_file
    self.read_pipe = os.fdopen(r)
    # Create an event for the upload thread to communicate back
    # to tornado when it's done, and save a reference to our IOLoop.
    self.upload_done = tornado.locks.Event()
    self.io_loop = tornado.ioloop.IOLoop.current()
    # Consider using a tornado.locks.Semaphore to limit the number of
    # threads you can create.
    self.thread = threading.Thread(target=self.upload_file)
    self.thread.start()

def upload_file(self):
    google_client.upload_from_file(self.read_pipe)
    # tell the IOLoop thread we're finished
    self.io_loop.add_callback(self.upload_done.set)

async def data_received(self, chunk):
    await self.write_pipe.write(chunk)

async def put(self):  # or post()
    self.write_pipe.close()
    await self.upload_done.wait()
    self.thread.join()
    self.render("upload_done.html")

或者，您可以避免使用谷歌的同步库，并使用底层HTTP API和AsyncHTTPClient执行所有操作。以这种方式排序身份验证是棘手的，但您可以避免线程不匹配。这将涉及在this gist中使用body_producer