參考：
https://blog.csdn.net/weixin_42502089/article/details/147689236
https://www.cnblogs.com/bitterteaer/p/17581746.html
修改下載緩沖區大小
https://ask.csdn.net/questions/8328950

1. 文件下載處理

對于文件下載，FastAPI 提供了 FileResponse 和 StreamingResponse 兩種方式。 FileResponse 適合小文件，而 StreamingResponse 適合大文件，因為它可以分塊返回文件內容。

1.1. 服務端處理

1.1.1. 下載小文件

使用 FileResponse 可以直接下載文件，而無需在內存中加載整個文件。

"""
fastapi + request 上傳和下載功能
"""
from fastapi import FastAPI, UploadFile
from fastapi.responses import FileResponse
import uvicornapp = FastAPI()# filename 下載時設置的文件名
@app.get("/download/small/{filename}")
async def download_small_file(filename: str):print(filename)file_path = "./測試任務.pdf"return FileResponse(file_path, filename=filename, media_type="application/octet-stream")if __name__ == '__main__':uvicorn.run(app, port=8000)

保證當前目錄下有名為“測試任務.pdf”的文件。
然后使用瀏覽器下載：
http://127.0.0.1:8000/download/small/ceshi.pdf

1.1.2. 下載大文件（yield 支持預覽的）

使用 StreamingResponse 可以分塊下載文件，這樣不會占用太多服務器資源，特別適用于大文件的下載。

from fastapi.responses import StreamingResponse
from fastapi import HTTPException
@app.get("/download/big/{filename}")
async def download_big_file(filename: str):def iter_file(path: str):with open(file=path, mode="rb") as tfile:yield tfile.read()# while chunk := tfile.read(1024*1024):  # 1MB 緩沖區#     yield chunkfile_path = "./測試任務.pdf"if not os.path.exists(file_path):raise HTTPException(status_code=404, detail="File not found")# # 支持瀏覽器預覽# return StreamingResponse(content=iter_file(path=file_path), status_code = 200,)# 直接下載return StreamingResponse(iter_file(path=file_path), media_type="application/octet-stream", headers={"Content-Disposition": f"attachment; filename={filename}"})

然后使用瀏覽器下載：
http://127.0.0.1:8000/download/big/ceshi_big.pdf

1.1.3. 下載大文件（bytes）

import io
@app.get("/download/bytes/{filename}")
async def download_bytes_file(filename: str):def read_bytes(path: str):content = "Error"with open(file=path, mode="rb") as tfile:content = tfile.read()# # 失敗，需要轉成bytes輸出# return contentreturn io.BytesIO(content)file_path = "./測試任務.pdf"if not os.path.exists(file_path):raise HTTPException(status_code=404, detail="File not found")# 解決中文名錯誤from urllib.parse import quote# return StreamingResponse(content=read_bytes(path=file_path), media_type="application/octet-stream", #                          headers={"Content-Disposition": "attachment; filename={}".format(quote(filename))})return StreamingResponse(content=read_bytes(path=file_path), media_type="application/octet-stream", headers={"Content-Disposition": "attachment; filename={}".format(quote(filename))})

1.1.4. 提供靜態文件服務

FastAPI 允許開發者使用 StaticFiles 來提供靜態文件服務。這類似于傳統 Web 服務器處理文件的方式。

from fastapi.staticfiles import StaticFiles# app.mount("/static", StaticFiles(directory="static", html=True), name="free")
app.mount("/static", StaticFiles(directory="fonts", html=True), name="free")

尚未測試通過。

1.1.5. 中文文件名錯誤

下載文件時，當傳遞文件名為中文時，報錯。

    # 解決中文名錯誤from urllib.parse import quotereturn StreamingResponse(iter_file(path=file_path), media_type="application/octet-stream", headers={"Content-Disposition": "attachment; filename={}".format(quote(filename))})

1.2. 客戶端處理

參考（還有進度條， 帶有斷點續傳的下載， 帶有超時和重試的下載）：
https://blog.csdn.net/u013762572/article/details/145158401
批量上傳下載
https://blog.csdn.net/weixin_43413871/article/details/137027968

1.2.1. 普通下載

import requests
import os"""方式1，將整個文件下載在保存到本地"""
def download_file_bytes(file_name):# 以下三個地址均可以url = "http://127.0.0.1:8000/download/small/ceshi_samll.pdf"url = "http://127.0.0.1:8000/download/bytes/ceshi_bytes.pdf"url = "http://127.0.0.1:8000/download/big/ceshi_big.pdf"# response = requests.get(url, params={"filename": "1.txt"})response = requests.get(url)# print(response.text)with open(file_name, 'wb') as file:# file.write(response.text)file.write(response.content)if __name__ == '__main__':download_file_bytes("本地測試下載文件bytes.pdf")

1.2.2. 分塊下載

import requests
import os"""方式2，通過流的方式一次寫入8192字節"""
def download_file_big(file_name):# 以下三個地址均可以url = "http://127.0.0.1:8000/download/small/ceshi_samll.pdf"# url = "http://127.0.0.1:8000/download/big/ceshi_big.pdf"# url = "http://127.0.0.1:8000/download/bytes/ceshi_bytes.pdf"# response = requests.get(url, params={"filename": "./測試任務.pdf"}, stream=True)response = requests.get(url, stream=True)with open(file_name, 'wb') as file:for chunk in response.iter_content(chunk_size=8192):file.write(chunk)if __name__ == '__main__':download_file_big("本地測試下載文件big.pdf")

1.2.3. 顯示進度條下載

import requests
import os
from tqdm import tqdmdef download_file_tqdm(file_name):# 以下三個地址均可以# url = "http://127.0.0.1:8000/download/small/ceshi_samll.pdf"# url = "http://127.0.0.1:8000/download/big/ceshi_big.pdf"url = "http://127.0.0.1:8000/download/bytes/ceshi_bytes.pdf"response = requests.get(url, stream=True)if response.status_code == 200:file_size = int(response.headers.get('content-length', 0))# 顯示進度條progress = tqdm(response.iter_content(chunk_size=8192), total=file_size,unit='B', unit_scale=True)with open(file_name, 'wb') as f:for data in progress:f.write(data)return Truereturn Falseif __name__ == '__main__':download_file_tqdm("本地測試下載文件tqdm.pdf")

運行結果：

> python.exe .\fast_client.py
1.92kB [00:00, 14.0kB/s]

1.2.4. 帶有斷點續傳的下載

# 帶有斷點續傳的下載
def resume_download(file_name):# 以下三個地址均可以# url = "http://127.0.0.1:8000/download/small/ceshi_samll.pdf"# url = "http://127.0.0.1:8000/download/big/ceshi_big.pdf"url = "http://127.0.0.1:8000/download/bytes/ceshi_bytes.pdf"# 獲取已下載文件大小initial_pos = os.path.getsize(file_name) if os.path.exists(file_name) else 0# 設置 Headerheaders = {'Range': f'bytes={initial_pos}-'}response = requests.get(url, stream=True, headers=headers)# 追加模式打開文件mode = 'ab' if initial_pos > 0 else 'wb'with open(file_name, mode) as f:for chunk in response.iter_content(chunk_size=8192):if chunk:f.write(chunk)

尚未測試

1.2.5. 帶有超時和重試的下載

# 帶有超時和重試的下載
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
import time
def download_with_retry(file_name, max_retries=3, timeout=30):# 以下三個地址均可以# url = "http://127.0.0.1:8000/download/small/ceshi_samll.pdf"# url = "http://127.0.0.1:8000/download/big/ceshi_big.pdf"url = "http://127.0.0.1:8000/download/bytes/ceshi_bytes.pdf"session = requests.Session()# 設置重試策略retries = Retry(total=max_retries,backoff_factor=1,status_forcelist=[500, 502, 503, 504])session.mount('http://', HTTPAdapter(max_retries=retries))session.mount('https://', HTTPAdapter(max_retries=retries))try:response = session.get(url, stream=True, timeout=timeout)with open(file_name, 'wb') as f:for chunk in response.iter_content(chunk_size=8192):if chunk:f.write(chunk)return Trueexcept Exception as e:print(f"Download failed: {str(e)}")return False

尚未測試

1.2.6. 完整的下載器實現

import requests
from tqdm import tqdm
import os
from pathlib import Path
import hashlibclass FileDownloader:def __init__(self, chunk_size=8192):self.chunk_size = chunk_sizeself.session = requests.Session()def get_file_size(self, url):response = self.session.head(url)return int(response.headers.get('content-length', 0))def get_file_hash(self, file_path):sha256_hash = hashlib.sha256()with open(file_path, "rb") as f:for byte_block in iter(lambda: f.read(4096), b""):sha256_hash.update(byte_block)return sha256_hash.hexdigest()def download(self, url, save_path, verify_hash=None):save_path = Path(save_path)# 創建目錄save_path.parent.mkdir(parents=True, exist_ok=True)# 獲取文件大小file_size = self.get_file_size(url)# 設置進度條progress = tqdm(total=file_size,unit='B',unit_scale=True,desc=save_path.name)try:response = self.session.get(url, stream=True)with save_path.open('wb') as f:for chunk in response.iter_content(chunk_size=self.chunk_size):if chunk:f.write(chunk)progress.update(len(chunk))progress.close()# 驗證文件完整性if verify_hash:downloaded_hash = self.get_file_hash(save_path)if downloaded_hash != verify_hash:raise ValueError("File hash verification failed")return Trueexcept Exception as e:progress.close()print(f"Download failed: {str(e)}")if save_path.exists():save_path.unlink()return Falsedef download_multiple(self, url_list, save_dir):results = []for url in url_list:filename = url.split('/')[-1]save_path = Path(save_dir) / filenamesuccess = self.download(url, save_path)results.append({'url': url,'success': success,'save_path': str(save_path)})return results# 使用示例
downloader = FileDownloader()# 單文件下載
url = "http://127.0.0.1:8000/download/bytes/ceshi_bytes.pdf"
downloader.download(url, save_path="downloads/file.pdf")# # 多文件下載
# urls = [
#     "https://example.com/file1.pdf",
#     "https://example.com/file2.pdf"
# ]
# results = downloader.download_multiple(urls, "downloads")

運行結果：

> python.exe .\fast_client_plus.py
file.pdf: 9.18MB [00:00, 60.2MB/s]

2. 文件上傳處理

FastAPI 提供了 File() 和 UploadFile() 兩個類用于處理文件上傳。 File() 適用于小文件，而 UploadFile() 則更適合處理大文件。在文件上傳時，我們建議使用異步版本的函數，這樣可以避免阻塞服務器。

2.1. 服務端處理

2.1.1. 上傳小文件

使用 File() 時，可以通過 httpie 或 requests 庫來模擬上傳操作。需要注意的是，上傳文件時應該使用表單而不是 JSON，這可以通過在命令中加入 -f 或 --form 參數來指定。

from fastapi import File
@app.post("/upload/small")
async def upload_small_file(small_file: bytes = File()) -> str:return f"file size: {len(small_file)}"

尚未測試

2.1.2. 上傳大文件

對于大文件，建議使用 UploadFile ，因為它會在服務器的磁盤上創建一個臨時文件對象，而不是將整個文件加載到內存中。這樣可以有效避免服務器內存溢出。

from fastapi import UploadFile
import time
import os
@app.post("/upload/big")
async def upload_big_file(big_file: UploadFile= File(...)):filename = f"{str(time.time()).replace('.','')}-{big_file.filename}"path = os.path.join("upload", filename)with open(path, "wb") as f:f.write(big_file.file.read())f.flush()return {"filename": filename,"filesize": big_file.size}

2.2. 客戶端處理

import requests
import osdef upload_file_big(file_path):url = "http://127.0.0.1:8000/upload/big"with open(file_path, 'rb') as f:contents = f.read()response = requests.post(url, files={"file": (os.path.basename(file_path), contents, 'multipart/form-data')})return response.json()if __name__ == '__main__':upload_file_big(r"./example.pdf")

尚未測試