python-background-jobs

来源声明：本技能内容源自 wshobson/agents （MIT License, Copyright (c) 2024 Seth Hobson；完整许可文本见同目录 LICENSE.txt）。 使用、修改与再分发须遵守该 MIT 许可条款并保留版权声明。

中文描述

本技能提供全栈开发相关的最佳实践和模式指导。

---

name: python-background-jobs description: Python 后台作业模式，包括任务队列、工作进程和事件驱动架构。适用于实现异步任务处理、作业队列、长时间运行的操作，或将工作与请求/响应周期解耦。

Python 后台作业与任务队列

将长时间运行或不可靠的工作与请求/响应周期解耦。立即向用户返回结果，同时由后台工作进程异步处理繁重工作。

何时使用此技能

• 处理耗时超过几秒的任务
• 发送电子邮件、通知或 Webhook
• 生成报告或导出数据
• 处理上传内容或媒体转换
• 集成不可靠的外部服务
• 构建事件驱动架构

核心概念

1. 任务队列模式

API 接收请求，将作业加入队列，并立即返回作业 ID。工作进程异步处理作业。

2. 幂等性

任务失败后可能会重试。设计时应确保可以安全地重新执行。

3. 作业状态机

作业在以下状态之间转换：待处理 → 运行中 → 成功/失败。

4. 至少一次投递

大多数队列保证至少一次投递。代码必须能够处理重复任务。

快速入门

本技能使用 Celery 展示示例。Celery 是广泛采用的任务队列。RQ、Dramatiq 和云原生解决方案（AWS SQS、GCP Tasks）等替代方案也同样可行。

from celery import Celery

app = Celery("tasks", broker="redis://localhost:6379")

@app.task
def send_email(to: str, subject: str, body: str) -> None:
    # This runs in a background worker
    email_client.send(to, subject, body)

# In your API handler
send_email.delay("[email protected]", "Welcome!", "Thanks for signing up")

基本模式

模式 1：立即返回作业 ID

对于耗时超过几秒的操作，应返回作业 ID，并进行异步处理。

from uuid import uuid4
from dataclasses import dataclass
from enum import Enum
from datetime import datetime

class JobStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    SUCCEEDED = "succeeded"
    FAILED = "failed"

@dataclass
class Job:
    id: str
    status: JobStatus
    created_at: datetime
    started_at: datetime | None = None
    completed_at: datetime | None = None
    result: dict | None = None
    error: str | None = None

# API endpoint
async def start_export(request: ExportRequest) -> JobResponse:
    """Start export job and return job ID."""
    job_id = str(uuid4())

    # Persist job record
    await jobs_repo.create(Job(
        id=job_id,
        status=JobStatus.PENDING,
        created_at=datetime.utcnow(),
    ))

    # Enqueue task for background processing
    await task_queue.enqueue(
        "export_data",
        job_id=job_id,
        params=request.model_dump(),
    )

    # Return immediately with job ID
    return JobResponse(
        job_id=job_id,
        status="pending",
        poll_url=f"/jobs/{job_id}",
    )

模式 2：Celery 任务配置

为 Celery 任务配置适当的重试和超时设置。

from celery import Celery

app = Celery("tasks", broker="redis://localhost:6379")

# Global configuration
app.conf.update(
    task_time_limit=3600,          # Hard limit: 1 hour
    task_soft_time_limit=3000,      # Soft limit: 50 minutes
    task_acks_late=True,            # Acknowledge after completion
    task_reject_on_worker_lost=True,
    worker_prefetch_multiplier=1,   # Don't prefetch too many tasks
)

@app.task(
    bind=True,
    max_retries=3,
    default_retry_delay=60,
    autoretry_for=(ConnectionError, TimeoutError),
)
def process_payment(self, payment_id: str) -> dict:
    """Process payment with automatic retry on transient errors."""
    try:
        result = payment_gateway.charge(payment_id)
        return {"status": "success", "transaction_id": result.id}
    except PaymentDeclinedError as e:
        # Don't retry permanent failures
        return {"status": "declined", "reason": str(e)}
    except TransientError as e:
        # Retry with exponential backoff
        raise self.retry(exc=e, countdown=2 ** self.request.retries * 60)

模式 3：确保任务具有幂等性

工作进程发生崩溃或超时后可能会重试。设计时应确保可以安全地重新执行。

@app.task(bind=True)
def process_order(self, order_id: str) -> None:
    """Process order idempotently."""
    order = orders_repo.get(order_id)

    # Already processed? Return early
    if order.status == OrderStatus.COMPLETED:
        logger.info("Order already processed", order_id=order_id)
        return

    # Already in progress? Check if we should continue
    if order.status == OrderStatus.PROCESSING:
        # Use idempotency key to avoid double-charging
        pass

    # Process with idempotency key
    result = payment_provider.charge(
        amount=order.total,
        idempotency_key=f"order-{order_id}",  # Critical!
    )

    orders_repo.update(order_id, status=OrderStatus.COMPLETED)

幂等性策略：

1. 写入前检查：执行操作前验证状态
2. 幂等键：在外部服务中使用唯一令牌
3. Upsert 模式：INSERT ... ON CONFLICT UPDATE
4. 去重窗口：记录已处理的 ID，保留 N 小时

模式 4：作业状态管理

持久化作业状态转换，以便查看和调试。

class JobRepository:
    """Repository for managing job state."""

    async def create(self, job: Job) -> Job:
        """Create new job record."""
        await self._db.execute(
            """INSERT INTO jobs (id, status, created_at)
               VALUES ($1, $2, $3)""",
            job.id, job.status.value, job.created_at,
        )
        return job

    async def update_status(
        self,
        job_id: str,
        status: JobStatus,
        **fields,
    ) -> None:
        """Update job status with timestamp."""
        updates = {"status": status.value, **fields}

        if status == JobStatus.RUNNING:
            updates["started_at"] = datetime.utcnow()
        elif status in (JobStatus.SUCCEEDED, JobStatus.FAILED):
            updates["completed_at"] = datetime.utcnow()

        await self._db.execute(
            "UPDATE jobs SET status = $1, ... WHERE id = $2",
            updates, job_id,
        )

        logger.info(
            "Job status updated",
            job_id=job_id,
            status=status.value,
        )

详细示例与模式

详细章节（以 ## Advanced Patterns 开头）位于 references/details.md。如果上面的导航摘要不够充分，请阅读该文件。

最佳实践摘要

1. 立即返回 - 不要让请求因长时间运行的操作而阻塞
2. 持久化作业状态 - 支持状态轮询和调试
3. 确保任务具有幂等性 - 任何失败后都可以安全重试
4. 使用幂等键 - 用于外部服务调用
5. 设置超时 - 同时设置软限制和硬限制
6. 实现 DLQ - 捕获永久失败的任务
7. 记录状态转换 - 跟踪作业状态变化
8. 合理重试 - 对暂时性错误使用指数退避
9. 不要重试永久性故障 - 例如验证错误、无效凭据
10. 监控队列深度 - 在积压增长时发出警报

能力边界

✅ 擅长处理

• 全栈开发相关的最佳实践
• 代码架构设计指导
• 性能优化建议

⚠️ 需要素材

• 具体的项目上下文
• 技术栈信息

❌ 超出范围

• 非技术领域的业务决策
• 具体的代码实现（仅提供指导）

工作流程

步骤 1：需求分析

理解项目需求和技术约束

步骤 2：架构设计

设计系统架构和组件划分

步骤 3：实现指导

提供具体的实现建议和代码模式

步骤 4：代码审查

审查代码质量和最佳实践遵循情况

步骤 5：优化建议

提供性能优化和改进建议

注意事项

1. 过度设计

避免在简单场景中使用复杂的设计模式

2. 性能忽视

不要忽视性能优化，特别是在生产环境中

3. 安全考虑

始终考虑安全性和数据保护

4. 测试覆盖

确保有足够的测试覆盖，特别是边界情况

5. 文档维护

保持文档与代码同步更新