fastapi实现文件上传和下载的功能

基础代码

import os
import io
from fastapi import FastAPI, File, UploadFile, Responseapp = FastAPI()# 设置上传文件的保存目录
UPLOAD_FOLDER = 'data'
# 确保上传目录存在
os.makedirs(UPLOAD_FOLDER, exist_ok=True)@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):# 获取文件名filename = file.filename# 读取文件内容file_content = await file.read()# 保存文件到磁盘save_file_path = os.path.join(UPLOAD_FOLDER, filename)with open(save_file_path, 'wb') as f:f.write(file_content)return {"message": "文件上传成功", "filename": file.filename}@app.get("/download/")
def download_file():# 假设你有一个文件的内容和名称file_content = b"abc"file_name = "example.txt"response = Response(content=file_content, media_type="text/plain")response.headers["Content-Disposition"] = f"attachment; filename={file_name}"return responseif __name__ == '__main__':import uvicornuvicorn.run(app, host='0.0.0.0', port=8000)

上传文件并提取文件信息保存到数据库

import os
import uuid
import time
import fastzdp_upload
from fastapi import FastAPI, File, UploadFile, Response,Depends
from sqlmodel import SQLModel, Field, create_engine, Session
from typing import Optional# 创建数据库引擎
sqlite_url = "mysql+pymysql://root:root@127.0.0.1:3306/test?charset=utf8mb4"
engine = create_engine(sqlite_url, echo=True)class FileModel(SQLModel, table=True):id: Optional[int] = Field(default=None, primary_key=True)name: str = Field(index=True)uuid: strsuffix: strmd5: strsave_dir: strpath: strorigin_name: strsize: intadd_time: int# 确保表存在
SQLModel.metadata.drop_all(engine)
SQLModel.metadata.create_all(engine)def get_db():db = Session(engine)try:yield dbfinally:db.close()app = FastAPI()# 设置上传文件的保存目录
UPLOAD_FOLDER = 'data'
# 确保上传目录存在
os.makedirs(UPLOAD_FOLDER, exist_ok=True)@app.post("/upload/")
async def upload_file(file: UploadFile = File(...), db: Session = Depends(get_db),):# 获取文件名filename = file.filenameuuid_value = str(uuid.uuid4())filename = f"{uuid_value}_{filename}"# 读取文件内容file_content = await file.read()# 保存文件到磁盘save_file_path = os.path.join(UPLOAD_FOLDER, filename)with open(save_file_path, 'wb') as f:f.write(file_content)# 文件信息保存到数据库db_file = FileModel(name=filename,uuid=uuid_value,suffix=fastzdp_upload.util.get_suffix(filename),md5=fastzdp_upload.util.get_md5(file_content),save_dir=UPLOAD_FOLDER,path=f"{UPLOAD_FOLDER}/{filename}",origin_name=file.filename,size=len(file_content),add_time = time.time(),)db.add(db_file)db.commit()return {"message": "文件上传成功", "filename": file.filename}@app.get("/download/")
def download_file():# 假设你有一个文件的内容和名称file_content = b"abc"file_name = "example.txt"response = Response(content=file_content, media_type="text/plain")response.headers["Content-Disposition"] = f"attachment; filename={file_name}"return responseif __name__ == '__main__':import uvicornuvicorn.run(app, host='0.0.0.0', port=8000)

使用封装的方法

import fastzdp_upload
from fastapi import FastAPI
from sqlmodel import SQLModel, Field, create_engine, Session# 创建数据库引擎
sqlite_url = "mysql+pymysql://root:root@127.0.0.1:3306/test?charset=utf8mb4"
engine = create_engine(sqlite_url, echo=True)# 确保表存在
SQLModel.metadata.drop_all(engine)
SQLModel.metadata.create_all(engine)def get_db():db = Session(engine)try:yield dbfinally:db.close()app = FastAPI()app.include_router(fastzdp_upload.get_router(get_db))if __name__ == '__main__':import uvicornuvicorn.run(app, host='0.0.0.0', port=8000)