Fetches celery-expert from martinholovsky/claude-skills-generator and configures it for Cursor.
3
Select Cursor when prompted
The CLI shows a list of agents. Use arrow keys and space to select Cursor:
◆ Which agents do you want to install to?
│
│ ── Universal (.agents/skills) ────────────────
│ · Cline · Codex · Goose · Windsurf
│ ●Cursor(selected)
│ · Cursor · Aider · Continue
4
Verify installation
Confirm successful installation by checking the skill directory location:
.cursor/skills/celery-expert
Restart Cursor to activate celery-expert. Access via /celery-expert in your agent's command palette.
⚠
Security Notice
We perform automated surface-level scans (Gen AI Scanner, Socket, Snyk) during installation. These checks detect common vulnerabilities but do not guarantee complete security. Always review skill source code and verify the publisher's reputation before production use.
Skills execute code in your environment. Always review source, verify the publisher, and test in isolation before production.
Task processing failures can impact business operations
Improper serialization (pickle) can lead to code execution vulnerabilities
Missing retries/timeouts can cause task accumulation and system degradation
Broker misconfigurations can lead to task loss or message exposure
2. Implementation Workflow (TDD)
Step 1: Write Failing Test First
# tests/test_tasks.pyimport pytest
from celery.contrib.testing.tasks import ping
from celery.result import EagerResult
@pytest.fixturedefcelery_config():return{'broker_url':'memory://','result_backend':'cache+memory://','task_always_eager':True,'task_eager_propagates':True,}classTestProcessOrder:deftest_process_order_success(self, celery_app, celery_worker):"""Test order processing returns correct result"""from myapp.tasks import process_order
# Execute task result = process_order.delay(order_id=123)# Assert expected behaviorassert result.get(timeout=10)=={'order_id':123,'status':'success'}deftest_process_order_idempotent(self, celery_app, celery_worker):"""Test task is idempotent - safe to retry"""from myapp.tasks import process_order
# Run twice result1 = process_order.delay(order_id=123).get(timeout=10) result2 = process_order.delay(order_id=123).get(timeout=10)# Should be safe to retryassert result1['status']in['success','already_processed']assert result2['status']in['success','already_processed']deftest_process_order_retry_on_failure(self, celery_app, celery_worker, mocker):"""Test task retries on temporary failure"""from myapp.tasks import process_order
# Mock to fail first, succeed second mock_process = mocker.patch('myapp.tasks.perform_order_processing') mock_process.side_effect =[TemporaryError("Timeout"),{'result':'ok'}] result = process_order.delay(order_id=123)assert result.get(timeout=10)['status']=='success'assert mock_process.call_count ==2
Step 2: Implement Minimum to Pass
# myapp/tasks.pyfrom celery import Celery
app = Celery('tasks', broker='redis://localhost:6379/0')@app.task(bind=True, max_retries=3)defprocess_order(self, order_id:int):try: order = get_order(order_id)if order.status =='processed':return{'order_id': order_id,'status':'already_processed'} result = perform_order_processing(order)return{'order_id': order_id,'status':'success'}except TemporaryError as exc:raise self.retry(exc=exc, countdown=2** self.request.retries)
Step 3: Refactor Following Patterns
Add proper error handling, time limits, and observability.
Step 4: Run Full Verification
# Run all Celery testspytest tests/test_tasks.py -v# Run with coveragepytest tests/test_tasks.py --cov=myapp.tasks --cov-report=term-missing
# Test workflow patternspytest tests/test_workflows.py -v# Integration test with real brokerpytest tests/integration/ --broker=redis://localhost:6379/0
3. Performance Patterns
Pattern 1: Task Chunking
# Bad - Individual tasks for each itemfor item_id in item_ids:# 10,000 items = 10,000 tasks process_item.delay(item_id)# Good - Process in batches@app.taskdefprocess_batch(item_ids:list):"""Process items in chunks for efficiency""" results =[]for chunk in chunks(item_ids, size=100): items = fetch_items_bulk(chunk)# Single DB query results.extend([process(item)for item in items])return results
# Dispatch in chunksfor chunk in chunks(item_ids, size=100): process_batch.delay(chunk)# 100 tasks instead of 10,000
Pattern 2: Prefetch Tuning
# Bad - Default prefetch for I/O-bound tasksapp.conf.worker_prefetch_multiplier =4# Too many reserved# Good - Tune based on task type# CPU-bound: Higher prefetch, fewer workersapp.conf.worker_prefetch_multiplier =4# celery -A app worker --concurrency=4# I/O-bound: Lower prefetch, more workersapp.conf.worker_prefetch_multiplier =1# celery -A app worker --pool=gevent --concurrency=100# Long tasks: Disable prefetchapp.conf.worker_prefetch_multiplier =1app.conf.task_acks_late =True
Pattern 3: Result Backend Optimization
# Bad - Storing results for fire-and-forget tasks@app.taskdefsend_email(to, subject, body)
✓
Make data-driven prioritization decisions faster
Stakeholder Communication
Draft PRDs, status updates, and stakeholder presentations
›Access to product documentation and roadmap tools (Jira, Notion, etc.)
›Understanding of product management frameworks (RICE, Jobs-to-be-Done, etc.)
›Stakeholder contact information and communication channels
Time Estimate
30-60 minutes to see productivity improvements
Steps
1Install product management skill
2Start with user story generation for known feature
3Progress to competitive analysis: research 2-3 competitors
4Use for roadmap prioritization: apply RICE/ICE scoring
5Draft stakeholder communications and refine based on feedback
6Build template library for recurring PM tasks
7Share effective prompts with product team
Common Pitfalls
⚠Not validating competitive research—verify facts before sharing
⚠Accepting user stories without involving engineering team
⚠Over-relying on frameworks without qualitative judgment
⚠Not customizing outputs to company culture and communication style
⚠Skipping stakeholder validation of generated requirements
Best Practices
✓ Do
+Validate research and competitive analysis with real data
+Collaborate with engineering when generating technical requirements
+Customize frameworks and templates to your company context
+Use skill for first drafts, refine with stakeholder input
+Document successful prompt patterns for PM tasks
+Combine AI efficiency with human judgment and intuition
✗ Don't
−Don't publish competitive analysis without fact-checking
−Don't finalize user stories without engineering review
−Don't make prioritization decisions solely on AI scoring
−Don't skip customer validation of generated requirements
−Don't ignore company-specific context and culture
💡 Pro Tips
★Provide context: company goals, constraints, customer feedback
★Ask for alternatives: 'Show 3 ways to prioritize this roadmap'
★Request stakeholder-specific formatting: 'Executive summary vs. engineering spec'
★Use skill for 70% generation + 30% customization to company needs
When to Use This
✓ Use when
Use for user story writing, competitive research, roadmap prioritization, stakeholder communication, and PRD drafting. Best for reducing repetitive documentation and research work.
✗ Avoid when
Avoid for strategic product vision (requires deep customer empathy), pricing decisions (needs market and financial expertise), or when face-to-face customer discovery is more valuable than speed.
Learning Path
1Basic: user stories, feature specs, status updates
