向表中插入数据时如何处理 FastAPI、SQLAlchemy 和 SQL Server 中的触发器？

Question

我已经在数据库中创建了表和触发器， 现在，当我尝试在数据库中创建新记录时，会出现错误。 但是如果我删除触发器，那么我可以在数据库中添加新记录，

这是错误

{ "detail": "创建组织时发生内部服务器错误 (pyodbc.ProgrammingError) ('42000', "[42000] [Microsoft][ODBC Driver 17 for SQL Server][SQL Server]DML 语句的目标表 'organization'如果语句包含没有 INTO 子句的 OUTPUT 子句，则不能有任何启用的触发器。(334) (SQLExecDirectW)") [SQL：INSERT INTO 组织（名称、代码、组织管理员、客户公司、城市、州、国家/地区、创建日期、创建用户、更新日期、更新用户、活动、评论） 输出插入的.id 值 (?, ?, ?, ?, ?, ? , ?, ?, ?, ?, ?, ?, ?)] [参数：('Eric Item Set-1', 'EricTest-64-1', None, None, None, None, None, datetime.datetime(2024, 3, 3, 21, 2, 43, 352446), None ，无，无，无，无）] （此错误的背景位于：https://sqlalche.me/e/20/f405）” }

这是我的代码

---

模型.py

class Organization(Base):
    __tablename__ = "organization"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50), nullable=True)
    code = Column(String(50), nullable=True)
    organization_administrator = Column(Integer, nullable=True)
    customer_company = Column(String(50), nullable=True)
    city = Column(String(50), nullable=True)
    state = Column(String(50), nullable=True)
    country = Column(String(50), nullable=True)
    create_date = Column(DateTime, nullable=True)
    create_user = Column(Integer, nullable=True)
    update_date = Column(DateTime, nullable=True)
    update_user = Column(Integer, nullable=True)
    active = Column(Boolean, nullable=True)
    comments = Column(String, nullable=True)
    _table_args_ = {'implicit_returning':False}

---

架构.py

from typing import Optional ,List
from pydantic import BaseModel
from datetime import datetime


class MyBaseModel(BaseModel):
    class Config():
        from_attributes = True


class OrganizationBase(BaseModel):
    name: Optional[str] = None
    code: Optional[str] = None
    organization_administrator: Optional[int] = None
    customer_company: Optional[str] = None
    city: Optional[str] = None
    state: Optional[str] = None
    country: Optional[str] = None
    create_date: Optional[datetime] = None
    create_user: Optional[int] = None
    update_date: Optional[datetime] = None
    update_user: Optional[int] = None
    active: Optional[bool] = None
    comments: Optional[str] = None


class OrganizationCreate(OrganizationBase):
    name: str
    code: str


class OrganizationUpdate(OrganizationBase):
    pass


class OrganizationInDBBase(OrganizationBase):
    id: int

    class Config:
        from_attributes = True


class Organization(OrganizationInDBBase):
    pass


class Pagination(BaseModel):
    total_records: int
    current_page: int
    total_pages: int


class OrganizationWithPagination(BaseModel):
    data : List[Organization]
    pagination : Pagination

---

端点.py

@router.post('', response_model=OrganizationSchema)
def create(request: OrganizationCreate, db: Session = Depends(get_db)):
    return services.create(request, db)

---

服务.py

def create(request: OrganizationCreate, db: Session):
    try:
        request.create_date = datetime.now()
        request.update_date = None
        instance = Organization(**dict(request))
        db.add(instance)
        db.commit()
        db.refresh(instance)
        return instance
    except Exception as e:
        raise HTTPException(
            status_code=400, detail=f"internal server error while creating organization {e}")

我尝试使用原始查询插入记录，并且能够插入。 但是如果我不想使用原始查询怎么办？ 而且，如果我使用原始查询，那么我无法返回新创建的记录 ID。

注意：我使用的是 SQL Server

Answer 1

从 app.py 导入数据库 然后在你的py代码中使用它

例如

db.session.add(OrganizationCreate)
db.session.commit()
db.session.rollback()
db.session.close()