如何用单次asyncio.ensure_future调用结合PyScript读文件与API上传

要实现单次asyncio.ensure_future调用完成"选择图片→读取二进制数据→上传至API"的完整流程，只需将文件读取逻辑与API请求逻辑整合到同一个异步回调中，以下是完整实现代码：

<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="utf-8" />
    <meta name="viewport" content="width=device-width,initial-scale=1" />

    <title>PyScript 图片上传至API示例</title>

    <link rel="stylesheet" href="https://pyscript.net/latest/pyscript.css" />
    <script defer src="https://pyscript.net/latest/pyscript.js"></script>
    <py-config>
      [[fetch]]
      files = ["/request.py"]
    </py-config>
</head>

<body>
    <p>选择本地图片文件，自动上传至API：</p>
    <label for="myfile">选择图片：</label>
    <input type="file" id="myfile" name="myfile" accept="image/*">
    <br><br>
    <div id="upload_status"></div>

    <py-script>
        import asyncio
        import json
        import base64
        from request import request
        from js import document, console
        from pyodide import create_proxy

        async def process_and_upload_file(event):
            file_list = event.target.files.to_py()
            for f in file_list:
                # 读取文件二进制数据并转为base64格式
                array_buffer = await f.arrayBuffer()
                bytes_data = bytes(array_buffer)
                base64_str = base64.b64encode(bytes_data).decode('utf-8')
                # 拼接API要求的data URI格式
                file_data = f"data:{f.type};base64,{base64_str}"
                
                # 构造API请求参数
                upload_preset = "<preset>"  # 替换为你的上传预设
                cloud_name = "<cloudname>"  # 替换为你的Cloudinary云名称
                body = json.dumps({
                    "upload_preset": upload_preset,
                    "file": file_data
                })
                headers = {"Content-type": "application/json"}
                api_url = f"https://api.cloudinary.com/v1_1/{cloud_name}/upload"
                
                # 发送上传请求
                try:
                    response = await request(api_url, body=body, method="POST", headers=headers)
                    response_json = await response.json()
                    status_element = document.getElementById("upload_status")
                    status_element.innerHTML = f"上传成功！状态码: {response.status}<br>返回数据: {json.dumps(response_json, indent=2)}"
                    console.log(f"上传结果: {response_json}")
                except Exception as e:
                    document.getElementById("upload_status").innerHTML = f"上传失败: {str(e)}"
                    console.error(f"上传错误: {e}")

        async def main():
            # 创建文件选择事件的代理回调
            file_event_proxy = create_proxy(process_and_upload_file)
            # 绑定文件选择框的change事件
            file_input = document.getElementById("myfile")
            file_input.addEventListener("change", file_event_proxy, False)

        # 启动主流程，仅需一次asyncio调用
        asyncio.ensure_future(main())
    </py-script>
</body>
</html>

关键修改说明 #

• 文件读取逻辑优化：放弃原示例的text()读取方式，改用arrayBuffer()获取二进制数据，再通过base64.b64encode转为API要求的base64格式，同时拼接文件MIME类型形成标准的data URI。
• 流程整合：将API请求逻辑直接嵌入文件选择的回调函数process_and_upload_file中，实现"选完文件立即上传"的连贯流程。
• 单次asyncio调用：整个初始化流程（绑定事件）仅通过一次asyncio.ensure_future(main())启动，所有后续操作（文件读取、API请求）都在事件回调的异步逻辑中完成。

内容的提问来源于stack exchange，提问作者user1079785