FASTAPI Excel文件下载接口故障排查求助

一、修复数据库数据获取问题 #

现有代码仅构建了SQLAlchemy查询对象，未实际从数据库拉取数据，且未传入数据库会话，导致无数据写入Excel。

1. 给send_report_user添加db参数，从接口传入：

# 接口中调用后台任务时传入db
background_tasks.add_task(send_report_user, current_user, db)

# 修改send_report_user函数定义
def send_report_user(current_user, db: Session):

2. 执行查询获取实际用户数据：

# 替换原users、users_enable、users_disable的定义
users = db.execute(query.order_by(User.last_name)).scalars().all()
users_enable = db.execute(query.filter(User.is_active == True).order_by(User.last_name)).scalars().all()
users_disable = db.execute(query.filter(User.is_active == False).order_by(User.last_name)).scalars().all()

二、修复文件路径与写入逻辑 #

1. 正确使用save_path，确保目录存在，处理Windows路径转义：

import os
import datetime

save_path = r'C:\miProyecto'  # 使用原始字符串避免转义问题
os.makedirs(save_path, exist_ok=True)  # 确保目录存在，不存在则创建

excel_file_name = f'Nomina_AIG_{datetime.datetime.now().date().strftime("%d-%m-%Y")}.xlsx'
full_file_path = os.path.join(save_path, excel_file_name)
workbook = xlsxwriter.Workbook(full_file_path)  # 使用完整路径创建文件

2. 修复工作表数据写入错误：原代码中Usuarios activos工作表创建后未写入数据就切换到下一个工作表，同时移除不必要的行号写入：

# Usuarios activos 工作表写入数据
worksheet = workbook.add_worksheet(name='Usuarios activos')
row = 1
col = 0
# 写入表头
for id, name in enumerate(title_columns):
    worksheet.write(0, id, name)
    worksheet.set_column(0, id, 15)
# 写入用户数据
for row_data in users_enable:
    data = [row_data.id, row_data.last_name, row_data.first_name, row_data.department_id, row_data.email, row_data.phone]
    worksheet.write_row(row, col, data)
    row += 1

# Usuarios inactivos 工作表
worksheet = workbook.add_worksheet(name='Usuarios inactivos')
row = 1
col = 0
worksheet.write_row(0, 0, title_columns)
for row_data in users_disable:
    data = [row_data.id, row_data.last_name, row_data.first_name, row_data.department_id, row_data.email, row_data.phone]
    worksheet.write_row(row, col, data)
    row += 1

# Usuarios totales 工作表
worksheet = workbook.add_worksheet(name='Usuarios totales')
row = 1
col = 0
worksheet.write_row(0, 0, title_columns)
for row_data in users:
    data = [row_data.id, row_data.last_name, row_data.first_name, row_data.department_id, row_data.email, row_data.phone]
    worksheet.write_row(row, col, data)
    row += 1

三、修复依赖传递问题 #

send_report_user中不要直接调用deps.s3_auth()，而是从接口传入已初始化的S3客户端：

# 接口中调用后台任务时传入s3
background_tasks.add_task(send_report_user, current_user, db, s3)

# 修改send_report_user函数定义
def send_report_user(current_user, db: Session, s3: BaseClient):

四、实现Excel文件下载功能 #

根据文件大小选择合适的方案：

方案1：后台生成后返回S3下载链接（适合大文件） #

生成文件后上传到S3，生成预签名下载链接（可通过邮件或后续查询接口交付用户）：

# 在send_report_user末尾添加S3上传逻辑
with open(full_file_path, 'rb') as f:
    s3.upload_fileobj(f, 'your-bucket-name', excel_file_name)

# 生成有效期1小时的预签名下载链接
download_url = s3.generate_presigned_url(
    'get_object',
    Params={'Bucket': 'your-bucket-name', 'Key': excel_file_name},
    ExpiresIn=3600
)
# 可将download_url存入数据库，或通过邮件发送给current_user

方案2：同步生成并直接返回文件（适合小文件） #

取消后台任务，直接在接口中生成文件并返回下载：

from fastapi import FileResponse
from io import BytesIO

@router.get("/users/export", status_code=200)
async def export_users(
    *,
    db: Session = Depends(deps.get_db),
    current_user: models.User = Depends(deps.get_current_active_dashboard)
):
    # 构建查询逻辑（同send_report_user中的query构建）
    if crud.user.is_only_superuser(user= current_user):
        query = (select(User).where(User.role_id !=4))
    elif current_user.department_id == 2:
        query = (select(User)).where(and_(User.role_id != 1, User.department_id !=1 ))
    else:
        query = (select(User)).where(User.department_id == current_user.department_id, User.role_id != 1, User.department_id !=1 )
    
    users = db.execute(query.order_by(User.last_name)).scalars().all()
    users_enable = db.execute(query.filter(User.is_active == True).order_by(User.last_name)).scalars().all()
    users_disable = db.execute(query.filter(User.is_active == False).order_by(User.last_name)).scalars().all()

    # 内存中生成Excel
    output = BytesIO()
    workbook = xlsxwriter.Workbook(output)
    title_columns = ['ID', 'Apellido','Nombre','Departamento','Correo electrónico','Teléfono']
    
    # 写入三个工作表逻辑（同前文修复后的写入代码）
    # ...
    
    workbook.close()
    output.seek(0)
    excel_file_name = f'Nomina_AIG_{datetime.datetime.now().date().strftime("%d-%m-%Y")}.xlsx'

    return FileResponse(
        output,
        media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
        filename=excel_file_name
    )

五、添加错误处理 #

在send_report_user中添加异常捕获，避免后台任务崩溃无日志：

import logging

def send_report_user(current_user, db: Session, s3: BaseClient):
    try:
        # 原函数逻辑
    except Exception as e:
        logging.error(f"生成Excel报告失败: {str(e)}")
        # 可添加告警逻辑，比如邮件通知管理员

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