如何在FastAPI的/img-upload/端点将AWS图片URL存入PostgreSQL

1. 确认House模型定义 #

先确保你的House模型已正确包含img_url字段，示例如下：

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class House(Base):
    __tablename__ = "houses"
    
    id = Column(Integer, primary_key=True, index=True)
    img_url = Column(String, nullable=True)
    # 其他业务字段...

2. 修改/img-upload/端点逻辑 #

假设你已实现S3上传并获取URL的核心代码，现在添加数据库更新逻辑：

核心流程 #

• 上传文件到S3并生成可访问的URL
• 通过请求参数（如house_id）定位目标House记录
• 更新记录的img_url字段并提交会话

完整代码示例 #

from fastapi import FastAPI, File, UploadFile, Depends
from sqlalchemy.orm import Session
from .database import get_db  # 已配置好的数据库会话依赖
from .models import House
import boto3
import uuid

app = FastAPI()
s3_client = boto3.client('s3')
BUCKET_NAME = "your-bucket-name"

@app.post("/img-upload/")
async def upload_image(house_id: int, file: UploadFile = File(...), db: Session = Depends(get_db)):
    # 生成唯一文件名避免重复
    unique_filename = f"{uuid.uuid4()}_{file.filename}"
    
    # 上传文件到S3
    s3_client.upload_fileobj(
        file.file,
        BUCKET_NAME,
        unique_filename,
        ExtraArgs={"ACL": "public-read", "ContentType": file.content_type}
    )
    
    # 生成S3图片URL
    img_url = f"https://{BUCKET_NAME}.s3.amazonaws.com/{unique_filename}"
    
    # 定位并更新House记录
    house = db.query(House).filter(House.id == house_id).first()
    if not house:
        return {"error": "目标房屋记录不存在"}
    
    house.img_url = img_url
    db.commit()
    db.refresh(house)  # 刷新对象获取最新数据
    
    return {"img_url": img_url, "updated_house": house}

3. 关键注意事项 #

• 原子性保障：用try-except块处理异常，确保S3上传失败时回滚数据库操作：

try:
    s3_client.upload_fileobj(...)
    house.img_url = img_url
    db.commit()
except Exception as e:
    db.rollback()
    return {"error": str(e)}

• S3权限配置：确保桶的ACL或政策允许生成的URL公开访问（若业务需要）
• 文件命名规则：必须用唯一标识（如UUID）作为文件名前缀，避免同文件名覆盖旧文件

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