AgentAlley

Skills · Coding

Python Background Jobs & Task Queues

Unverified●30/40

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.

Originally by wshobson · MIT

Claude Code◐PartialHas SKILL.md but declares no allowed-tools — Claude Code will ask for permission each time
Cursor◐PartialPlain prose you can paste in — but no Cursor rules file
Codex◐PartialPlain prose you can paste in — but no AGENTS.md
Gemini CLI◐PartialPlain prose you can paste in
Copilot◐PartialPlain prose you can paste in — but no Copilot instructions file
npx agentalley add python-background-jobs

This command does not work yet — the CLI is still being built. Until then, use Raw in the reader below to take the file.

Who is stuck, and on what

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.

The whole source

No sign-in, no blur, nothing truncated
python-background-jobs/SKILL.md242 lines7.1 KBRawView on GitHub
Frontmatter — 2 properties
namepython-background-jobs
descriptionPython 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.
1---
2name: python-background-jobs
3description: 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.
4---A5 — No allowed-tools declared — no way to tell what this skill may touch
5 
6# Python Background Jobs & Task Queues
7 
8Decouple long-running or unreliable work from request/response cycles. Return immediately to the user while background workers handle the heavy lifting asynchronously.
9 
10## When to Use This Skill
11 
12- Processing tasks that take longer than a few seconds
13- Sending emails, notifications, or webhooks
14- Generating reports or exporting data
15- Processing uploads or media transformations
16- Integrating with unreliable external services
17- Building event-driven architectures
18 
19## Core Concepts
20 
21### 1. Task Queue Pattern
22 
23API accepts request, enqueues a job, returns immediately with a job ID. Workers process jobs asynchronously.
24 
25### 2. Idempotency
26 
27Tasks may be retried on failure. Design for safe re-execution.
28 
29### 3. Job State Machine
30 
31Jobs transition through states: pending → running → succeeded/failed.
32 
33### 4. At-Least-Once Delivery
34 
35Most queues guarantee at-least-once delivery. Your code must handle duplicates.
36 
37## Quick Start
38 
39This 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.
40 
41```python
42from celery import Celery
43 
44app = Celery("tasks", broker="redis://localhost:6379")
45 
46@app.task
47def send_email(to: str, subject: str, body: str) -> None:
48 # This runs in a background worker
49 email_client.send(to, subject, body)
50 
51# In your API handler
52send_email.delay("[email protected]", "Welcome!", "Thanks for signing up")
53```
54 
55## Fundamental Patterns
56 
57### Pattern 1: Return Job ID Immediately
58 
59For operations exceeding a few seconds, return a job ID and process asynchronously.
60 
61```python
62from uuid import uuid4
63from dataclasses import dataclass
64from enum import Enum
65from datetime import datetime
66 
67class JobStatus(Enum):
68 PENDING = "pending"
69 RUNNING = "running"
70 SUCCEEDED = "succeeded"
71 FAILED = "failed"
72 
73@dataclass
74class Job:
75 id: str
76 status: JobStatus
77 created_at: datetime
78 started_at: datetime | None = None
79 completed_at: datetime | None = None
80 result: dict | None = None
81 error: str | None = None
82 
83# API endpoint
84async def start_export(request: ExportRequest) -> JobResponse:
85 """Start export job and return job ID."""
86 job_id = str(uuid4())
87 
88 # Persist job record
89 await jobs_repo.create(Job(
90 id=job_id,
91 status=JobStatus.PENDING,
92 created_at=datetime.utcnow(),
93 ))
94 
95 # Enqueue task for background processing
96 await task_queue.enqueue(
97 "export_data",
98 job_id=job_id,
99 params=request.model_dump(),
100 )
101 
102 # Return immediately with job ID
103 return JobResponse(
104 job_id=job_id,
105 status="pending",
106 poll_url=f"/jobs/{job_id}",
107 )
108```
109 
110### Pattern 2: Celery Task Configuration
111 
112Configure Celery tasks with proper retry and timeout settings.
113 
114```python
115from celery import Celery
116 
117app = Celery("tasks", broker="redis://localhost:6379")
118 
119# Global configuration
120app.conf.update(
121 task_time_limit=3600, # Hard limit: 1 hour
122 task_soft_time_limit=3000, # Soft limit: 50 minutes
123 task_acks_late=True, # Acknowledge after completion
124 task_reject_on_worker_lost=True,
125 worker_prefetch_multiplier=1, # Don't prefetch too many tasks
126)
127 
128@app.task(
129 bind=True,
130 max_retries=3,
131 default_retry_delay=60,
132 autoretry_for=(ConnectionError, TimeoutError),
133)
134def process_payment(self, payment_id: str) -> dict:
135 """Process payment with automatic retry on transient errors."""
136 try:
137 result = payment_gateway.charge(payment_id)
138 return {"status": "success", "transaction_id": result.id}
139 except PaymentDeclinedError as e:
140 # Don't retry permanent failures
141 return {"status": "declined", "reason": str(e)}
142 except TransientError as e:
143 # Retry with exponential backoff
144 raise self.retry(exc=e, countdown=2 ** self.request.retries * 60)
145```
146 
147### Pattern 3: Make Tasks Idempotent
148 
149Workers may retry on crash or timeout. Design for safe re-execution.
150 
151```python
152@app.task(bind=True)
153def process_order(self, order_id: str) -> None:
154 """Process order idempotently."""
155 order = orders_repo.get(order_id)
156 
157 # Already processed? Return early
158 if order.status == OrderStatus.COMPLETED:
159 logger.info("Order already processed", order_id=order_id)
160 return
161 
162 # Already in progress? Check if we should continue
163 if order.status == OrderStatus.PROCESSING:
164 # Use idempotency key to avoid double-charging
165 pass
166 
167 # Process with idempotency key
168 result = payment_provider.charge(
169 amount=order.total,
170 idempotency_key=f"order-{order_id}", # Critical!
171 )
172 
173 orders_repo.update(order_id, status=OrderStatus.COMPLETED)
174```
175 
176**Idempotency Strategies:**
177 
1781. **Check-before-write**: Verify state before action
1792. **Idempotency keys**: Use unique tokens with external services
1803. **Upsert patterns**: `INSERT ... ON CONFLICT UPDATE`
1814. **Deduplication window**: Track processed IDs for N hours
182 
183### Pattern 4: Job State Management
184 
185Persist job state transitions for visibility and debugging.
186 
187```python
188class JobRepository:
189 """Repository for managing job state."""
190 
191 async def create(self, job: Job) -> Job:
192 """Create new job record."""
193 await self._db.execute(
194 """INSERT INTO jobs (id, status, created_at)
195 VALUES ($1, $2, $3)""",
196 job.id, job.status.value, job.created_at,
197 )
198 return job
199 
200 async def update_status(
201 self,
202 job_id: str,
203 status: JobStatus,
204 **fields,
205 ) -> None:
206 """Update job status with timestamp."""
207 updates = {"status": status.value, **fields}
208 
209 if status == JobStatus.RUNNING:
210 updates["started_at"] = datetime.utcnow()
211 elif status in (JobStatus.SUCCEEDED, JobStatus.FAILED):
212 updates["completed_at"] = datetime.utcnow()
213 
214 await self._db.execute(
215 "UPDATE jobs SET status = $1, ... WHERE id = $2",
216 updates, job_id,
217 )
218 
219 logger.info(
220 "Job status updated",
221 job_id=job_id,
222 status=status.value,
223 )
224```
225 
226## Detailed worked examples and patterns
227 
228Detailed sections (starting with `## Advanced Patterns`) live in `references/details.md`. Read that file when the navigation summary above is insufficient.
229 
230## Best Practices Summary
231 
2321. **Return immediately** - Don't block requests for long operations
2332. **Persist job state** - Enable status polling and debugging
2343. **Make tasks idempotent** - Safe to retry on any failure
2354. **Use idempotency keys** - For external service calls
2365. **Set timeouts** - Both soft and hard limits
2376. **Implement DLQ** - Capture permanently failed tasks
2387. **Log transitions** - Track job state changes
2398. **Retry appropriately** - Exponential backoff for transient errors
2409. **Don't retry permanent failures** - Validation errors, invalid credentials
24110. **Monitor queue depth** - Alert on backlog growth
242 

Reviews

Installed this one?Write the first review and take the Trailblazer badge.

Reviews only open after a real install, so this is empty — and we leave it empty rather than invent one.

Alternatives

Also in Coding
Subagent Driven DevelopmentUse when executing implementation plans with independent tasks in the current sessionCoding · MIT◐◐◐◐◐●36/40Python Code Style & DocumentationPython code style, linting, formatting, naming conventions, and documentation standards. Use when writing new code, reviewing style, configuring linters, writing docstrings, or establishing project standards.Coding · MIT◐····●35/40Competitor Price Analysis 💲Competitor pricing strategy analysis and market positioning. Price mapping, pricing gaps identification, elasticity signals evaluation, and strategic pricing optimization. Use when the user asks about competitor pricing, price analysis, pricing strategy, or coCoding · MIT◐····●34/40Competitor Price Tracker 📊Set up competitor price tracking and monitoring workflows. Track price changes, detect promotions, analyze pricing patterns, and get alerts for competitive price movements.Coding · MIT◐····●34/40