FastAPI中如何设置HTTP错误？请帮我排查代码问题

存在的问题 #

• 重复执行SQL查询：连续两次执行select * from Employee;，完全冗余，增加数据库不必要的负载，也降低接口性能。
• 硬编码列索引取值风险大：通过i[3]、i[4]获取username和password，完全依赖表的列顺序。一旦表结构调整（比如新增列、调整列顺序），会直接导致取值错误，引发业务异常。
• 错误判断逻辑不准确：当前通过if not (username)判断是否抛404，只有当username列表为空（即无任何员工数据）时才触发，但如果表中有员工数据但username字段全为空，username列表会是多个空字符串，此时if not username条件不成立，不会抛出异常，不符合“无有效用户则返回404”的需求。而且判断是否存在数据，直接检查查询结果results_as_dict是否为空更合理。
• 异步函数中使用同步数据库操作可能阻塞事件循环：如果get_db返回的是同步数据库连接（如SQLAlchemy同步Cursor），在异步函数read_data中直接调用同步的conn.execute会阻塞FastAPI的事件循环，导致接口性能下降，甚至无法处理其他请求。

修复后的代码示例 #

from fastapi import HTTPException, status, Depends
from typing import List
# 导入你的其他依赖：app, EmployeeSchema, get_db（注意如果是同步连接要处理异步兼容）

@app.get("/employees", response_model=List[EmployeeSchema])
async def read_data(conn=Depends(get_db)):
    # 只执行一次查询
    temp = conn.execute("select * from Employee;")
    results_as_dict = temp.mappings().all()
    
    # 直接判断查询结果是否为空
    if not results_as_dict:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail={"detail": "用户未找到"}
        )
    
    # 如果需要获取username和password，用字段名而非索引
    usernames = [emp["username"] for emp in results_as_dict]
    passwords = [emp["password"] for emp in results_as_dict]
    
    print(results_as_dict)
    print("用户名列表", usernames)
    print("密码列表", passwords)
    
    return results_as_dict

额外优化建议 #

• 如果使用的是同步数据库驱动，建议将数据库操作放到线程池中执行，避免阻塞异步事件循环，示例：

  import asyncio
  from concurrent.futures import ThreadPoolExecutor
  
  executor = ThreadPoolExecutor()
  
  async def read_data(conn=Depends(get_db)):
      def sync_query():
          temp = conn.execute("select * from Employee;")
          return temp.mappings().all()
      
      results_as_dict = await asyncio.get_event_loop().run_in_executor(executor, sync_query)
      
      # 后续逻辑同上
  

• 建议在SQL查询中明确指定需要的字段，而非select *，既提高查询效率，也避免不必要的字段返回，比如select id, name, username, password from Employee;

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