使用Python FastAPI将JSON数据存入PostgreSQL遇问题求助

问题背景 #

我是Python和FastAPI的新手，正在开发一个项目，需求是上传JSON文件到接口，将其中的车辆数据存入PostgreSQL数据库。目前能成功将上传的JSON解析为Python字典，但无法完成数据库写入操作，求修改建议或方向指引。

相关代码细节 #

vehicles.json 结构（含500-600条数据） #

{
  "vehicleList": [
    {
      "id": 1,
      "coordinate": {
        "latitude": 54.5532316,
        "longitude": 67.0087783
      },
      "condition": "GOOD"
    },
    {
      "id": 2,
      "coordinate": {
        "latitude": 37.442316,
        "longitude": 38.0087783
      },
      "condition": "BREAKDOWM"
    }
  ]
}

connect.py（数据库表定义） #

from sqlalchemy import Table, Column, Integer, String, Float, ARRAY, MetaData

metadata_obj = MetaData()

vehicle_data = Table(
    "vehicle_data",
    metadata_obj,
    Column("coordinates", ARRAY(Float), nullable=False),
    Column("condition", String(6), nullable=False),
    Column("id", Integer, primary_key=True),
)

router.py（上传接口代码） #

from fastapi import APIRouter, File, UploadFile
from fastapi.responses import JSONResponse
import json
from psycopg2.extras import execute_values

router = APIRouter()

@router.post("/upload")
async def upload_json(file: UploadFile = File(...)):
    try:
        # Read the uploaded file as bytes
        contents = await file.read()
        # Decode the bytes to string assuming it's JSON
        decoded_content = contents.decode("utf-8")
        # Parse the JSON content
        json_data = json.loads(decoded_content)
        fields = [
            'coordinates', #List of floats
            'condition', #str
            'id' #int
        ]
        for item in json_data:
            my_data = [tuple(item[field] for field in fields) for item in json_data]
            insert_query = "INSERT INTO vehicle_data (coordinates, condition, id) VALUES %s"
            execute_values(insert_query, tuple(my_data))
        return JSONResponse(status_code=200, content={"message": "JSON file uploaded successfully", "data": json_data})
    
    except Exception as e:
        return JSONResponse(content={"error": str(e)}, status_code=500)

当前状态 #

• 能成功读取上传的JSON文件并解析为Python字典
• 无法将数据写入PostgreSQL，仅返回500错误

错误分析与修正方案 #

核心问题点 #

1. JSON遍历错误：JSON外层是vehicleList键，实际数据列表在json_data["vehicleList"]中，但代码直接遍历json_data（遍历的是字典的键，而非数据列表）。
2. 字段不匹配：JSON中是coordinate嵌套对象，代码误用coordinates字段名，且未将经纬度转换为数据库要求的数组格式。
3. 循环逻辑冗余：外层for item in json_data完全多余，内部生成my_data时重复遍历json_data，导致逻辑混乱。
4. 数据库连接缺失：代码没有获取活跃数据库连接的逻辑，execute_values无法直接执行。

修改后的router.py代码 #

from fastapi import APIRouter, File, UploadFile
from fastapi.responses import JSONResponse
import json
from sqlalchemy import create_engine
from connect import vehicle_data  # 导入表定义

router = APIRouter()

# 替换为你的PostgreSQL连接字符串
DATABASE_URL = "postgresql://user:password@localhost/dbname"
engine = create_engine(DATABASE_URL)

@router.post("/upload")
async def upload_json(file: UploadFile = File(...)):
    try:
        # 读取并解析JSON
        contents = await file.read()
        decoded_content = contents.decode("utf-8")
        json_data = json.loads(decoded_content)
        vehicle_list = json_data.get("vehicleList", [])
        
        if not vehicle_list:
            return JSONResponse(status_code=400, content={"error": "JSON文件中无vehicleList数据"})
        
        # 转换数据格式：将coordinate转为[纬度, 经度]数组
        insert_data = []
        for vehicle in vehicle_list:
            coordinates = [vehicle["coordinate"]["latitude"], vehicle["coordinate"]["longitude"]]
            insert_data.append({
                "coordinates": coordinates,
                "condition": vehicle["condition"],
                "id": vehicle["id"]
            })
        
        # 批量插入数据库
        with engine.connect() as conn:
            insert_stmt = vehicle_data.insert()
            conn.execute(insert_stmt, insert_data)
            conn.commit()
        
        return JSONResponse(
            status_code=200,
            content={"message": f"成功导入{len(insert_data)}条车辆数据"}
        )
    
    except Exception as e:
        return JSONResponse(content={"error": str(e)}, status_code=500)

额外注意事项 #

• 确保DATABASE_URL中的用户名、密码、主机、数据库名与你的PostgreSQL实例匹配。
• 表定义中condition字段是String(6)，但示例中的BREAKDOWM是8个字符，会导致插入失败，建议将表定义改为String(20)或修正JSON中的拼写（应为BREAKDOWN）。
• 若追求更高批量插入性能，可改用psycopg2的execute_values，替换插入部分代码：

import psycopg2
from psycopg2.extras import execute_values

# 替换SQLAlchemy插入逻辑
conn = psycopg2.connect(DATABASE_URL)
cur = conn.cursor()
insert_query = "INSERT INTO vehicle_data (coordinates, condition, id) VALUES %s"
values = [(item["coordinates"], item["condition"], item["id"]) for item in insert_data]
execute_values(cur, insert_query, values)
conn.commit()
cur.close()
conn.close()

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