FastAPI+PostgreSQL通过SQL插入数据后ID冲突问题求助

我搭建了一个连接PostgreSQL数据库的FastAPI应用，原本运行正常，但当通过SQL在后端向其中一张表插入数据后，应用无法识别ID已被占用。

例如：若通过应用录入的最后一条记录ID为10，我通过SQL插入了ID为11-21的10条记录，此时再通过应用录入数据时，它会尝试使用ID11，触发错误；重新录入时它会尝试ID12，以此类推，直到找到未被占用的ID。

我知道可以通过后端插入极高ID（如1000000）的方式规避，但想了解是否有办法让应用知晓当前ID的终止值，从而继续从正确位置生成ID，或其他解决方案。

方法1：配置主键为数据库自增序列（推荐） #

问题根源是当前employee_id仅为普通Integer主键，未关联数据库自增序列。手动SQL插入ID后，应用依赖的序列值未同步更新，导致冲突。

修改模型中的employee_id字段，添加autoincrement=True，让PostgreSQL自动维护关联序列：

class Employees(Base):
    __tablename__ = "employees"
    # 开启自增，PostgreSQL会自动创建并维护对应序列
    employee_id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    first_name = Column(String)
    last_name= Column(String)
    birth_date = Column(Date)
    entry_date = Column(Date)
    nic_number = Column(Integer)
    record_entered_by = Column(Integer, nullable=True)
    active = Column(Boolean, default=True)

    def __str__(self):
        return self.first_name +" "+ self.last_name

如果已经手动插入过较大ID，需要手动更新序列的当前值，确保后续自增从最大值开始：

-- 更新序列为当前最大employee_id +1，序列名默认是「表名_字段名_seq」
SELECT setval('employees_employee_id_seq', (SELECT MAX(employee_id) FROM employees));

方法2：插入前主动查询最大ID #

如果不想修改数据库结构，可在插入新数据前先查询当前表的最大ID，手动指定新ID：

from sqlalchemy import func

@app.post("/index")
async def add_user(request: Request, db: Session = Depends(get_db), fname: str = Form(), lname: str = Form(),
        dob: date = Form(), edate: date = Form(), nic: int = Form()):
    # 若表为空，max_id默认取0
    max_id = db.query(func.max(models.Employees.employee_id)).scalar() or 0
    employee = models.Employees(
        employee_id=max_id + 1,
        first_name=fname,
        last_name=lname,
        birth_date=dob,
        entry_date=edate,
        nic_number=nic
    )
    db.add(employee)
    db.commit()
    return RedirectResponse(url="/index", status_code=status.HTTP_303_SEE_OTHER)

注意：多并发场景下可能出现ID冲突，可添加锁保证原子性：

max_id = db.query(func.max(models.Employees.employee_id)).with_for_update().scalar() or 0

方法3：捕获冲突异常自动重试 #

若上述方法都不适用，可捕获主键冲突异常，自动重试并递增ID：

from sqlalchemy.exc import IntegrityError

@app.post("/index")
async def add_user(request: Request, db: Session = Depends(get_db), fname: str = Form(), lname: str = Form(),
        dob: date = Form(), edate: date = Form(), nic: int = Form()):
    max_retries = 5
    retry_count = 0
    while retry_count < max_retries:
        try:
            employee = models.Employees(
                first_name=fname,
                last_name=lname,
                birth_date=dob,
                entry_date=edate,
                nic_number=nic
            )
            db.add(employee)
            db.commit()
            return RedirectResponse(url="/index", status_code=status.HTTP_303_SEE_OTHER)
        except IntegrityError as e:
            db.rollback()
            retry_count += 1
            # 仅针对主键冲突重试，其他错误直接抛出
            if "unique constraint" in str(e).lower() or "duplicate key" in str(e).lower():
                continue
            raise
    return {"error": "插入失败，ID冲突重试次数过多"}

Employee模型 #

class Employees(Base):
    __tablename__ = "employees"
    employee_id = Column(Integer, primary_key=True, index=True)
    first_name = Column(String)
    last_name= Column(String)
    birth_date = Column(Date)
    entry_date = Column(Date)
    nic_number = Column(Integer)
    record_entered_by = Column(Integer, nullable=True)
    active = Column(Boolean, default=True)


    def __str__(self):
        return self.first_name +" "+ self.last_name

员工数据添加逻辑 #

@app.post("/index")
async def add_user(request: Request, db: Session = Depends(get_db), fname: str = Form(), lname: str = Form(),
        dob: date = Form(), edate: date = Form(), nic: int = Form()):
    
    employee = models.Employees(first_name=fname,
                            last_name=lname,
                            birth_date=dob,
                            entry_date=edate,
                            nic_number=nic)
    db.add(employee)
    db.commit()
    return RedirectResponse(url="/index", status_code=status.HTTP_303_SEE_OTHER)

前端表单 #

<form action="/index" method="POST">
    <!--Add Employee Table-->
    <label class="form-label" for="nic">NIC Number:</label>
    <input class="form-control" type="number" id="nic" name="nic" placeholder="Enter NIC"><br>

    <label class="form-label" for="fname">First name:</label><br>
    <input class="form-control" type="text" id="fname" name="fname" placeholder="Enter First Name"><br>

    <label class="form-label"  for="lname">Last name:</label><br>
    <input class="form-control" type="text" id="lname" name="lname" placeholder="Enter Last Name"><br>


    <label class="form-label" for="dob">Date of birth:</label><br>
    <input class="form-control" type="date" id="dob" name="dob" placeholder="DOB"><br>

    <label class="form-label" for="edate">Date of entry to the public service:</label><br>
    <input class="form-control" type="date" id="edate" name="edate" placeholder="Date of entry to the public service"><br>


    <input type="submit" class="btn btn-primary" value="Add">
    <br>
</form> 

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