如何使用FastAPI接收文件URL参数并将文件保存到磁盘？

FastAPI本身没有直接内置接收文件URL并自动下载保存的功能，但你可以通过简单的代码实现这个需求，下面一步步说明：

为什么UploadFile = File(...)不好使 #

UploadFile和File(...)是用来处理客户端直接上传二进制文件数据的，只认文件本身，不认字符串格式的URL。你传的是URL地址，不是文件流，所以FastAPI自然识别不了。

实现方案：把URL当普通参数接收，手动下载保存 #

你可以把文件URL作为普通的查询参数或请求体参数传给接口，然后在路由函数内部完成下载和磁盘写入操作——用你熟悉的requests就行，示例代码如下：

from fastapi import FastAPI
import requests
from pathlib import Path

app = FastAPI()

@app.post("/save-file-from-url")
def save_file_from_url(file_url: str, save_dir: str = "./downloads"):
    # 先创建保存目录，不存在就自动生成
    Path(save_dir).mkdir(parents=True, exist_ok=True)
    
    try:
        # 流式下载文件，避免大文件占满内存
        response = requests.get(file_url, stream=True)
        response.raise_for_status()  # 捕获HTTP请求错误（比如404、500）
        
        # 从URL里提取文件名，也可以让客户端自定义传入
        file_name = Path(file_url).name
        full_save_path = Path(save_dir) / file_name
        
        # 分块写入磁盘
        with open(full_save_path, "wb") as f:
            for chunk in response.iter_content(chunk_size=8192):
                f.write(chunk)
        
        return {"status": "成功", "保存路径": str(full_save_path)}
    except Exception as e:
        return {"status": "失败", "错误信息": str(e)}

进阶优化建议 #

• 加个URL合法性校验，比如检查是否以http://或https://开头
• 限制下载文件的大小，防止恶意请求拖垮服务器
• 允许客户端自定义文件名，不用硬从URL里提取
• 想用异步的话，可以换成aiohttp代替requests，更贴合FastAPI的异步特性：

  import aiohttp
  

from fastapi import FastAPI
from pathlib import Path

app = FastAPI()

@app.post("/save-file-from-url-async")
async def save_file_from_url_async(file_url: str, save_dir: str = "./downloads"):
Path(save_dir).mkdir(parents=True, exist_ok=True)
file_name = Path(file_url).name
full_save_path = Path(save_dir) / file_name

try:
    async with aiohttp.ClientSession() as session:
        async with session.get(file_url) as response:
            response.raise_for_status()
            with open(full_save_path, "wb") as f:
                async for chunk in response.content.iter_chunked(8192):
                    f.write(chunk)
    return {"status": "成功", "保存路径": str(full_save_path)}
except Exception as e:
    return {"status": "失败", "错误信息": str(e)}

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