Python Background Jobs

作者 wshobson46891e7e60da无许可证收录于 2026年10月8日更新于 2026年10月8日

Python background job patterns including task queues, workers, and event-driven architecture. Use when implementing async task processing, job queues, long-running operations, or decoupling work from request/response cycles.

仅含说明Software Development

AI 生成的概览

指导 Python 后台任务设计，涵盖任务队列、工作进程、重试、幂等性和任务状态跟踪。

功能: 该技能提供模式和代码示例，用于在 Python 应用中将长时间运行的工作与请求/响应周期解耦。内容涵盖使用 Celery 的任务队列、立即返回任务 ID、配置重试与超时、使任务具备幂等性，以及持久化任务状态转换。它还指向一个参考文件以获取高级模式。
适用场景: 适用于在 Python 中实现异步任务处理、任务队列或长时间运行操作时。适合发送邮件或 webhook、生成报告、处理上传文件以及与不可靠外部服务集成等场景。
运行要求: 不附带脚本，仅为说明性内容。示例假定使用 Python、Celery 以及 Redis 等消息代理，并需要数据库保存任务状态，但这些只是示例而非随附依赖。

Python Background Jobs & Task Queues

Decouple long-running or unreliable work from request/response cycles. Return immediately to the user while background workers handle the heavy lifting asynchronously.

When to Use This Skill

Processing tasks that take longer than a few seconds
Sending emails, notifications, or webhooks
Generating reports or exporting data
Processing uploads or media transformations
Integrating with unreliable external services
Building event-driven architectures

Core Concepts

1. Task Queue Pattern

API accepts request, enqueues a job, returns immediately with a job ID. Workers process jobs asynchronously.

2. Idempotency

Tasks may be retried on failure. Design for safe re-execution.

3. Job State Machine

Jobs transition through states: pending → running → succeeded/failed.

4. At-Least-Once Delivery

Most queues guarantee at-least-once delivery. Your code must handle duplicates.

Quick Start

This skill uses Celery for examples, a widely adopted task queue. Alternatives like RQ, Dramatiq, and cloud-native solutions (AWS SQS, GCP Tasks) are equally valid choices.

python

from celery import Celery
app = Celery("tasks", broker="redis://localhost:6379")
@app.taskdef send_email(to: str, subject: str, body: str) -> None:    # This runs in a background worker    email_client.send(to, subject, body)
# In your API handlersend_email.delay("[email protected]", "Welcome!", "Thanks for signing up")

from celery import Celery
app = Celery("tasks", broker="redis://localhost:6379")
@app.taskdef send_email(to: str, subject: str, body: str) -> None:    # This runs in a background worker    email_client.send(to, subject, body)
# In your API handlersend_email.delay("[email protected]", "Welcome!", "Thanks for signing up")

Fundamental Patterns

Pattern 1: Return Job ID Immediately

For operations exceeding a few seconds, return a job ID and process asynchronously.

python

from uuid import uuid4from dataclasses import dataclassfrom enum import Enumfrom datetime import datetime
class JobStatus(Enum):    PENDING = "pending"    RUNNING = "running"    SUCCEEDED = "succeeded"    FAILED = "failed"
@dataclassclass 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 endpointasync 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}",    )

from uuid import uuid4from dataclasses import dataclassfrom enum import Enumfrom datetime import datetime
class JobStatus(Enum):    PENDING = "pending"    RUNNING = "running"    SUCCEEDED = "succeeded"    FAILED = "failed"
@dataclassclass 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 endpointasync 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}",    )

Pattern 2: Celery Task Configuration

Configure Celery tasks with proper retry and timeout settings.

python

from celery import Celery
app = Celery("tasks", broker="redis://localhost:6379")
# Global configurationapp.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)

from celery import Celery
app = Celery("tasks", broker="redis://localhost:6379")
# Global configurationapp.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)

Pattern 3: Make Tasks Idempotent

Workers may retry on crash or timeout. Design for safe re-execution.

python

@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)

@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)

Idempotency Strategies:

Check-before-write: Verify state before action
Idempotency keys: Use unique tokens with external services
Upsert patterns: INSERT ... ON CONFLICT UPDATE
Deduplication window: Track processed IDs for N hours

Pattern 4: Job State Management

Persist job state transitions for visibility and debugging.

python

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,        )

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,        )

Detailed worked examples and patterns

Detailed sections (starting with ## Advanced Patterns) live in references/details.md. Read that file when the navigation summary above is insufficient.

Best Practices Summary

Return immediately - Don't block requests for long operations
Persist job state - Enable status polling and debugging
Make tasks idempotent - Safe to retry on any failure
Use idempotency keys - For external service calls
Set timeouts - Both soft and hard limits
Implement DLQ - Capture permanently failed tasks
Log transitions - Track job state changes
Retry appropriately - Exponential backoff for transient errors
Don't retry permanent failures - Validation errors, invalid credentials
Monitor queue depth - Alert on backlog growth

来源与署名

来源：wshobson/agents位于plugins/python-development/skills/python-background-jobs提交46891e7

许可证：无许可证

内容归原作者所有。SourceWeft 从公开仓库中收录这些内容。

举报或申请下架

Python Background Jobs

Python Background Jobs & Task Queues

When to Use This Skill

Core Concepts

1. Task Queue Pattern

2. Idempotency

3. Job State Machine

4. At-Least-Once Delivery

Quick Start

Fundamental Patterns

Pattern 1: Return Job ID Immediately

Pattern 2: Celery Task Configuration

Pattern 3: Make Tasks Idempotent

Pattern 4: Job State Management

Detailed worked examples and patterns

Best Practices Summary

来源与署名

更多来自 wshobson/agents 的技能

Uv Package Manager

Web Component Design

Python Type Safety

Visual Design Foundations

Responsive Design

Python Resource Management

更多Software Development技能

Playground

M5 Onboard

Cardputer Buddy

Testing Strategy

Tech Debt

System Design