（十九）后台导出的状态与独立会话

1. 先交回任务编号

现在希望把当前书架生成一份可再次读取的快照。接口先保存一条 queued 任务，返回 202 和任务位置；响应发送后，后台函数读取当前用户的书，保存结果并把状态改为 done。客户端再读取任务位置。

# 项目文件：api_project/src/bookapi/jobs.py
import logging

from sqlalchemy import select

from .models import Book, ExportJob
from .schemas import BookOut

logger = logging.getLogger(__name__)


def build_export(factory, job_id: str):
    # 后台任务另开 Session；不接收路由的 Session 或 ORM 对象。
    try:
        with factory.begin() as session:
            job = session.get(ExportJob, job_id)
            statement = (
                select(Book).where(Book.owner_id == job.owner_id).order_by(Book.id)
            )
            job.items = [
                BookOut.model_validate(book).model_dump(mode="json")
                for book in session.scalars(statement)
            ]
            job.status = "done"  # with 正常结束会提交，再关闭这个 Session。
    except Exception as exc:
        logger.error("导出任务失败：%s", type(exc).__name__)
        # 前一次事务已经回滚；用新的事务保存失败状态。
        with factory.begin() as session:
            job = session.get(ExportJob, job_id)
            job.status = "failed"

后台函数接收工厂和任务编号，再创建自己的会话。factory.begin() 同时管理会话与事务：成功提交，出错回滚，最后关闭。job.items 替换为 JSON 能保存的列表，和 job.status="done" 一起提交。

函数执行失败后，第一次事务回滚，再用新事务把任务标成 failed。logger.error 只记录异常类别。任务记录本身也有 owner_id，查询任务时同样核对归属。

# 项目文件：api_project/src/bookapi/routes/exports.py
import secrets

from fastapi import APIRouter, BackgroundTasks, Request, Response

from ..dependencies import CurrentIdentity, DbSession
from ..errors import Problem, commit_changes
from ..jobs import build_export
from ..models import ExportJob
from ..schemas import ExportOut

router = APIRouter(prefix="/exports", tags=["导出"])


@router.post("", response_model=ExportOut, status_code=202)
def start_export(
    session: DbSession,
    identity: CurrentIdentity,
    tasks: BackgroundTasks,
    request: Request,
    response: Response,
):
    with commit_changes(session):
        job = ExportJob(
            id=secrets.token_hex(16), owner_id=identity.user.id, status="queued"
        )
        session.add(job)
        session.flush()
        result = ExportOut.model_validate(job)
    # 先提交任务记录，再安排响应后的工作。
    tasks.add_task(build_export, request.app.state.session_factory, job.id)
    response.headers["Location"] = f"/exports/{job.id}"
    return result


@router.get("/{job_id}", response_model=ExportOut)
def read_export(job_id: str, session: DbSession, identity: CurrentIdentity):
    job = session.get(ExportJob, job_id)
    if job is None:
        raise Problem(404, "export_not_found", "没有找到这项导出")
    if job.owner_id != identity.user.id:
        raise Problem(403, "forbidden", "这项导出属于其他用户")
    return job

BackgroundTasks 是 FastAPI 提供的任务收集对象。tasks.add_task(function, arg1, arg2) 保存要在响应后执行的调用。传入的是工厂和编号，任务不会继续操作请求中的 Session 或 ORM 对象；请求与后台工作各自完成资源收尾。

先提交 queued 任务返回 202，响应后另开会话读取书籍，提交 done 快照，再由客户端下一次 GET 读取，失败分支回滚并标为 failed

快照反映后台函数实际读取时的书架，提交导出与后台读取之间发生的书籍修改也可能进入结果。后台任务运行在当前服务进程；若程序在 queued 后被结束，重启仍能看到 queued，但内存中的待执行调用已经消失。需要重启后自动领取任务，可以继续给持久化任务表添加领取、重试与恢复流程，或接独立任务队列。这里先观察一次响应后任务的完整执行，相关接口见 FastAPI BackgroundTasks。