在 FastAPI 中实现文件上传并通过 FTP 服务将文件存储到远程 FTP 服务器是一个常见的需求，尤其在需要将文件存储到远程服务器的场景中。FastAPI 提供了文件上传的支持，而 Python 的 `ftplib` 库可以用于实现 FTP 客户端功能。 ### 文件上传与 FTP 集成的实现步骤 FastAPI 使用 `File` 和 `UploadFile` 来处理上传的文件。上传的文件可以通过 `UploadFile` 对象读取其内容，然后将其发送到远程 FTP 服务器。 以下是一个完整的示例： 1. **安装依赖库** 确保安装 FastAPI 和 `python-multipart` 以支持文件上传： ```bash pip install fastapi uvicorn python-multipart ``` 2. **实现 FastAPI 文件上传与 FTP 上传** 下面的代码展示了如何实现文件上传，并通过 FTP 客户端将文件发送到远程服务器： ```python from fastapi import FastAPI, UploadFile, File, HTTPException from ftplib import FTP import io app = FastAPI() # 配置FTP服务器信息 FTP_HOST = "ftp.example.com" FTP_USER = "your_ftp_username" FTP_PASS = "your_ftp_password" FTP_PORT = 21 def upload_to_ftp(file_name: str, file_content: bytes): try: with FTP() as ftp: ftp.connect(FTP_HOST, FTP_PORT) ftp.login(FTP_USER, FTP_PASS) ftp.storbinary(f"STOR {file_name}", io.BytesIO(file_content)) return True except Exception as e: raise HTTPException(status_code=500, detail=f"FTP upload failed: {str(e)}") @app.post("/upload/") async def upload_file(file: UploadFile = File(...)): if not file: raise HTTPException(status_code=400, detail="No file uploaded") # 读取文件内容 file_content = await file.read() # 上传到FTP服务器 upload_to_ftp(file.filename, file_content) return {"message": f"File {file.filename} uploaded and sent to FTP server successfully."} ``` 3. **运行 FastAPI 服务** 使用 Uvicorn 运行 FastAPI 服务： ```bash uvicorn main:app --reload ``` 4. **测试文件上传接口** 使用 FastAPI 自带的文档界面（默认地址为 `http://localhost:8000/docs`）测试 `/upload/` 接口，上传文件后，文件将被发送到远程 FTP 服务器。 ### 安全性与错误处理 - **FTP 连接安全性** 上述示例使用的是明文 FTP 协议，不适用于生产环境。建议使用 `ftplib.FTP_TLS` 或 SFTP（通过 `paramiko` 库）来加密连接。 - **异常处理** 在实际应用中，应确保 FTP 连接失败、文件读取失败等情况下的错误处理机制，避免服务崩溃。 - **文件存储路径** 可以在 FTP 服务器上指定文件存储路径，例如将文件上传到 `/uploads` 目录。 ### 性能优化 - **异步 FTP 上传** 如果需要提高性能，可以使用异步 FTP 客户端库，如 `aioftp`，以便在上传文件时不会阻塞主线程。 - **文件缓存** 对于大文件上传，可以考虑将文件先存储在本地临时目录，再通过后台任务上传到 FTP 服务器，避免阻塞请求处理过程。 ### 示例：使用 `aioftp` 实现异步 FTP 上传 ```python import aioftp import asyncio async def async_upload_to_ftp(file_name: str, file_content: bytes): try: async with aioftp.ClientSession(FTP_HOST, FTP_USER, FTP_PASS) as client: await client.upload(file_name, io.BytesIO(file_content)) return True except Exception as e: raise HTTPException(status_code=500, detail=f"Async FTP upload failed: {str(e)}") @app.post("/async_upload/") async def async_upload_file(file: UploadFile = File(...)): if not file: raise HTTPException(status_code=400, detail="No file uploaded") file_content = await file.read() await async_upload_to_ftp(file.filename, file_content) return {"message": f"File {file.filename} uploaded and sent to FTP server asynchronously."} ``` ---