FastAPI调用PostgreSQL自定义函数返回异常结果，如何获取目标字符串？

API端点收到无法理解的响应：{ "_row_getter": {}, "_memoized_keys": [ "_row_getter" ] }

FastAPI代码如下：

from typing import Generator
from fastapi import APIRouter, Depends
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.sql import select, func #, column, table

from backend.core.config import settings


SQLALCHEMY_DATABASE_URL = settings.DATABASE_URL
engine = create_engine(SQLALCHEMY_DATABASE_URL)

SessionLocal = sessionmaker(autocommit=False,
                            autoflush=False,
                            bind=engine)

def get_db() -> Generator:
    try:
        db = SessionLocal()
        yield db
    finally:
        db.close()

def check_trc(ist: int, db: Session):
    chk_func = func.chktracer(ist)
    chk_stmt = select(chk_func)
    chk_result = db.execute(chk_stmt)
    db.commit()
    return chk_result

router = APIRouter()

@router.get("/chk/{ist}", response_description=str)
def read_trc(ist: int, db: Session= Depends(get_db)):
    chk = check_trc(ist= ist, db= db)
    return chk

PostgreSQL函数执行的SQL：

-- Upsert user request
INSERT INTO
    trc_log (id)
VALUES
    (p_id)
ON CONFLICT ON CONSTRAINT trc_log_pkey
DO UPDATE
SET req_cnt = EXCLUDED.req_cnt + 1;

-- return status of long running query
SELECT CASE
    WHEN finish_tm IS NULL
    THEN 'inprocess'
    ELSE 'processed'
    END
FROM trc_log
WHERE trc_log.pid = p_ist;

该函数在pgAdmin中返回单行单列的结果（'processed'或'inprocess'），但当前FastAPI代码无法返回期望的字符串，仅返回上述异常响应。

问题根源是db.execute(chk_stmt)返回的是SQLAlchemy的结果集对象，而非直接的字符串值。FastAPI无法正确序列化这个内部对象，因此返回了不符合预期的结构。

修改步骤 #

1. 提取结果集中的字符串值
   修改check_trc函数，使用scalar()方法从单行单列的结果集中提取实际状态值：

def check_trc(ist: int, db: Session):
    chk_func = func.chktracer(ist)
    chk_stmt = select(chk_func)
    chk_result = db.execute(chk_stmt)
    # 提取单行单列的唯一值
    status = chk_result.scalar()
    db.commit()
    return status

2. 明确路由响应类型（推荐）
   将路由的response_description替换为response_model=str，让FastAPI明确返回类型，增强类型校验和文档生成：

@router.get("/chk/{ist}", response_model=str)
def read_trc(ist: int, db: Session= Depends(get_db)):
    chk = check_trc(ist= ist, db= db)
    return chk

补充说明 #

• 如果需要处理无结果的情况，可改用scalar_one_or_none()，避免无匹配数据时抛出异常，之后可根据业务逻辑返回默认值或错误信息。
• 无需手动调用db.commit()，因为PostgreSQL函数内部的INSERT ... ON CONFLICT已经是原子操作，除非你需要在函数外额外提交事务，否则可移除该语句。

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