FastAPI图片上传与数据库存储功能实现咨询

针对你的疑问逐一解答： #

1. FastAPI文件上传配置 #

FastAPI原生支持文件上传，无需复杂全局配置，只需在接口参数中使用UploadFile（来自fastapi模块）即可处理单文件上传。如果需要限制文件大小，通过File类的max_length参数设置，比如限制10MB以内：

from fastapi import File, UploadFile

@app.post("/upload-image")
async def upload_image(file: UploadFile = File(..., max_length=10*1024*1024)):
    # 处理逻辑

另外，确保安装了python-multipart（FastAPI处理表单数据的依赖，执行pip install python-multipart即可）。

2. 获取文件并保存到数据库 #

通过UploadFile对象的read()方法读取二进制内容，再存入数据库对应的字段。核心流程：

• 读取文件二进制数据：file_content = await file.read()
• 读取文件名、文件类型：filename = file.filename，content_type = file.content_type
• 将这些数据存入数据库模型实例，提交会话完成保存

3. 数据库字段类型 #

使用二进制大对象类型：

• SQLAlchemy中用LargeBinary字段，对应数据库的BLOB类型（MySQL的MEDIUMBLOB、PostgreSQL的BYTEA等）
• 建议额外存储filename（字符串类型）和content_type（字符串类型），方便后续返回图片时设置正确的响应头

4. 简化流程的库 #

• 标准库：io（可选，用于处理二进制流操作）
• 第三方包：
  • sqlalchemy：ORM工具，简化数据库CRUD操作
  • pydantic：FastAPI默认依赖，用于数据验证和序列化
  • python-multipart：FastAPI处理文件上传的必需依赖
  • （可选）Pillow：如果需要对上传图片进行裁剪、压缩等预处理

完整代码示例 #

1. 安装依赖 #

pip install fastapi uvicorn sqlalchemy python-multipart

2. 完整代码 #

from fastapi import FastAPI, File, UploadFile, HTTPException
from fastapi.responses import Response
from sqlalchemy import create_engine, Column, Integer, String, LargeBinary
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# 数据库配置（这里用SQLite做示例，可替换为MySQL/PostgreSQL）
SQLALCHEMY_DATABASE_URL = "sqlite:///./images.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# 图片数据库模型
class Image(Base):
    __tablename__ = "images"
    id = Column(Integer, primary_key=True, index=True)
    filename = Column(String, index=True)
    content_type = Column(String)
    data = Column(LargeBinary)

# 创建数据库表
Base.metadata.create_all(bind=engine)

app = FastAPI()

# 依赖：获取数据库会话
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 上传图片接口
@app.post("/upload-image/")
async def upload_image(file: UploadFile = File(...), db=next(get_db())):
    # 验证文件类型（只允许图片）
    allowed_types = ["image/jpeg", "image/png", "image/gif"]
    if file.content_type not in allowed_types:
        raise HTTPException(status_code=400, detail="仅支持JPG、PNG、GIF格式的图片")
    
    # 读取文件内容
    file_content = await file.read()
    
    # 创建图片实例并保存到数据库
    db_image = Image(
        filename=file.filename,
        content_type=file.content_type,
        data=file_content
    )
    db.add(db_image)
    db.commit()
    db.refresh(db_image)
    
    return {"image_id": db_image.id, "filename": db_image.filename}

# 根据ID获取图片接口
@app.get("/image/{image_id}")
def get_image(image_id: int, db=next(get_db())):
    image = db.query(Image).filter(Image.id == image_id).first()
    if not image:
        raise HTTPException(status_code=404, detail="图片不存在")
    
    # 返回图片响应，设置正确的Content-Type
    return Response(content=image.data, media_type=image.content_type)

运行应用 #

uvicorn main:app --reload

测试方法：

• 访问http://localhost:8000/docs，使用自动生成的Swagger UI上传图片
• 上传后得到image_id，访问http://localhost:8000/image/{image_id}即可查看图片

内容的提问来源于stack exchange，提问作者Кирилл Бухалов