メニュー
Async job processing patterns for background tasks, Celery workflows, task scheduling, retry strategies, and distributed task execution. Use when implementing background job processing, task queues, or scheduled task systems.
デフォルトでは、最初にソースを確認する Prompt が選択されています。直接コマンドに切り替えるか、ローカルコピーをダウンロードすることもできます。
インストールを決める前に、SKILL.md と SkillsMP に表示されている付属ファイルをお読みください。
Codex または Claude でインストール この Prompt をコピーして Codex、Claude、または他のアシスタントに貼り付けると、Skill ページを確認してインストールできます。
直接コマンドでは確認用 Prompt が省略されます。実行前にソースを確認してください。
npx skills add https://github.com/diegosouzapw/awesome-omni-skill --skill async-jobsコマンドは1行のまま表示されます。コピー前に横へスクロールして全体を確認してください。
ローカルで確認しますか？SkillsMP が現在取得できるファイルをダウンロードできます。
Token-efficient tracking for AI orchestration. CLI-first for status updates (~50 tokens), agent fallback for complex ops (~1KB). Use when: updating task status, querying blockers, creating progress files, validating phases.
AshAi extension guidelines for integrating AI capabilities with Ash Framework. Use when implementing vectorization/embeddings, exposing Ash actions as LLM tools, creating prompt-backed actions, or setting up MCP servers. Covers semantic search, LangChain integration, and structured outputs.
This skill should be used when solving hard questions, complex architectural problems, or debugging issues that benefit from GPT-5 Pro or GPT-5.1 thinking models with large file context. Use when standard Claude analysis needs deeper reasoning or extended context windows.
SOC 職業分類に基づく
SKILL.md を表示中
| name | async-jobs |
| license | MIT |
| compatibility | Claude Code 2.1.56+. |
| description | Async job processing patterns for background tasks, Celery workflows, task scheduling, retry strategies, and distributed task execution. Use when implementing background job processing, task queues, or scheduled task systems. |
| tags | ["async","jobs","celery","background-tasks","scheduling","queues"] |
| context | fork |
| agent | python-performance-engineer |
| version | 2.0.0 |
| author | OrchestKit |
| user-invocable | false |
| complexity | medium |
| metadata | {"category":"workflow-automation"} |
Patterns for background task processing with Celery, ARQ, and Redis. Covers task queues, canvas workflows, scheduling, retry strategies, rate limiting, and production monitoring. Each category has individual rule files in references/ loaded on-demand.
| Category | Rules | Impact | When to Use |
|---|---|---|---|
| Configuration | celery-config | HIGH | Celery app setup, broker, serialization, worker tuning |
| Task Routing | task-routing | HIGH | Priority queues, multi-queue workers, dynamic routing |
| Canvas Workflows | canvas-workflows | HIGH | Chain, group, chord, nested workflows |
| Retry Strategies | retry-strategies | HIGH | Exponential backoff, idempotency, dead letter queues |
| Scheduling | scheduled-tasks | MEDIUM | Celery Beat, crontab, database-backed schedules |
| Monitoring | monitoring-health | MEDIUM | Flower, custom events, health checks, metrics |
| Result Backends | result-backends | MEDIUM | Redis results, custom states, progress tracking |
| ARQ Patterns | arq-patterns | MEDIUM | Async Redis Queue for FastAPI, lightweight jobs |
| Temporal Workflows | temporal-workflows | HIGH | Durable workflow definitions, sagas, signals, queries |
| Temporal Activities | temporal-activities | HIGH | Activity patterns, workers, heartbeats, testing |
Total: 10 rules across 9 categories
# Celery task with retry
from celery import shared_task
@shared_task()
() -> :
result = do_processing(order_id)
{: order_id, : }
# ARQ task with FastAPI
from arq import create_pool
from arq.connections import RedisSettings
async def generate_report(ctx: dict, report_id: str) -> dict:
data = await ctx["db"].fetch_report_data(report_id)
pdf = await render_pdf(data)
return {"report_id": report_id, "size": len(pdf)}
@router.post("/api/v1/reports")
async def create_report(data: ReportRequest, arq: ArqRedis = Depends(get_arq_pool)):
job = await arq.enqueue_job("generate_report", data.report_id)
return {"job_id": job.job_id}
Production Celery app configuration with secure defaults and worker tuning.
task_serializer="json" for safetytask_acks_late=True to prevent task loss on crashtask_time_limit (hard) and task_soft_time_limit (soft)worker_prefetch_multiplier=1task_reject_on_worker_lost=True| Decision | Recommendation |
|---|---|
| Serializer | JSON (never pickle) |
| Ack mode | Late ack (task_acks_late=True) |
| Prefetch | 1 for fair, 4-8 for throughput |
| Time limit | soft < hard (e.g., 540/600) |
| Timezone | UTC always |
Priority queue configuration with multi-queue workers and dynamic routing.
queue_order_strategy: "priority" and 0-9 levels| Decision | Recommendation |
|---|---|
| Queue count | 3-5 (critical/high/default/low/bulk) |
| Priority levels | 0-9 with Redis x-max-priority |
| Worker assignment | Dedicated workers per queue |
| Prefetch | 1 for critical, 4-8 for bulk |
| Routing | Router class for 5+ routing rules |
Celery canvas primitives for sequential, parallel, and fan-in/fan-out workflows.
si()) for steps that ignore input| Decision | Recommendation |
|---|---|
| Sequential | Chain with s() |
| Parallel | Group for independent tasks |
| Fan-in | Chord (all must succeed for callback) |
| Ignore input | Use si() immutable signature |
| Error in chain | Reject stops chain, retry continues |
| Partial failures | Return error dict in chord tasks |
Retry patterns with exponential backoff, idempotency, and dead letter queues.
retry_backoff=True and retry_backoff_maxretry_jitter=True to prevent thundering herd| Decision | Recommendation |
|---|---|
| Retry delay | Exponential backoff with jitter |
| Max retries | 3-5 for transient, 0 for permanent |
| Idempotency | Redis key with TTL |
| Failed tasks | DLQ for manual review |
| Singleton | Redis lock with TTL |
Celery Beat periodic task configuration with crontab, database-backed schedules, and overlap prevention.
django-celery-beat for dynamic schedules| Decision | Recommendation |
|---|---|
| Schedule type | Crontab for time-based, interval for frequency |
| Dynamic | Database scheduler (django-celery-beat) |
| Overlap | Redis lock with timeout |
| Beat process | Separate process (not embedded) |
| Timezone | UTC always |
Production monitoring with Flower, custom signals, health checks, and Prometheus metrics.
task_prerun, task_postrun, task_failure) for metrics| Decision | Recommendation |
|---|---|
| Dashboard | Flower with persistent storage |
| Metrics | Prometheus via celery signals |
| Health | Broker + worker + queue depth |
| Alerting | Signal on task_failure |
| Autoscale | Queue depth > threshold |
Task result storage, custom states, and progress tracking patterns.
update_state() for real-time progress reporting| Decision | Recommendation |
|---|---|
| Status storage | Redis result backend |
| Large results | S3 or database (never Redis) |
| Progress | Custom states with update_state() |
| Result query | AsyncResult with state checks |
Lightweight async Redis Queue for FastAPI and simple background tasks.
arq for FastAPI integrationstartup/shutdown hooks for resource managementenqueue_job()Job.status() and Job.result()_delay=timedelta() for deferred execution| Decision | Recommendation |
|---|---|
| Simple async | ARQ (native async) |
| Complex workflows | Celery (chains, chords) |
| In-process quick | FastAPI BackgroundTasks |
| LLM workflows | LangGraph (not Celery) |
| Tool | Best For | Complexity |
|---|---|---|
| ARQ | FastAPI, simple async jobs | Low |
| Celery | Complex workflows, enterprise | High |
| RQ | Simple Redis queues | Low |
| Dramatiq | Reliable messaging | Medium |
| FastAPI BackgroundTasks | In-process quick tasks | Minimal |
# NEVER run long tasks synchronously in request handlers
@router.post("/api/v1/reports")
async def create_report(data: ReportRequest):
pdf = await generate_pdf(data) # Blocks for minutes!
# NEVER block on results inside tasks (causes deadlock)
@celery_app.task
def bad_task():
result = other_task.delay()
return result.get() # Blocks worker!
# NEVER store large results in Redis
@shared_task
def process_file(file_id: str) -> bytes:
return large_file_bytes # Store in S3/DB instead!
# NEVER skip idempotency for retried tasks
@celery_app.task(max_retries=3)
def create_order(order):
Order.create(order) # Creates duplicates on retry!
# NEVER use BackgroundTasks for distributed work
background_tasks.add_task(long_running_job) # Lost if server restarts
# NEVER ignore task acknowledgment settings
celery_app.conf.task_acks_late = False # Default loses tasks on crash
# ALWAYS use immutable signatures in chords
chord([task.s(x) for x in items], callback.si()) # si() prevents arg pollution
Durable execution engine for reliable distributed applications with Temporal.io.
@workflow.defn and deterministic codeworkflow.wait_condition() for human-in-the-loopasyncio.gather inside workflows| Decision | Recommendation |
|---|---|
| Workflow ID | Business-meaningful, idempotent |
| Determinism | Use workflow.random(), workflow.now() |
| I/O | Always via activities, never directly |
Activity and worker patterns for Temporal.io I/O operations.
@activity.defn for all I/OApplicationError(non_retryable=True) for business errorsWorkflowEnvironment.start_local()| Decision | Recommendation |
|---|---|
| Activity timeout | start_to_close for most cases |
| Error handling | Non-retryable for business errors |
| Testing | WorkflowEnvironment for integration tests |
ork:python-backend - FastAPI, asyncio, SQLAlchemy patternsork:langgraph - LangGraph workflow patterns (use for LLM workflows, not Celery)ork:distributed-systems - Resilience patterns, circuit breakersork:monitoring-observability - Metrics and alertingKeywords: celery, configuration, broker, worker, setup Solves:
Keywords: priority, queue, routing, high priority, worker Solves:
Keywords: chain, group, chord, signature, canvas, workflow, pipeline Solves:
Keywords: retry, backoff, idempotency, dead letter, resilience Solves:
Keywords: periodic, scheduled, cron, celery beat, interval Solves:
Keywords: flower, monitoring, health check, metrics, alerting Solves:
Keywords: result, state, progress, AsyncResult, status Solves:
Keywords: arq, async queue, redis queue, fastapi background Solves: