FastAPI: 后台任务与 Celery — 异步任务队列

from celery import Celery

celery_app = Celery("pricetracker", broker="redis://localhost:6379/0")

@celery_app.task(bind=True, max_retries=3)
def scrape_prices(self, product_ids: list[int]):
    # Async price scraping - runs in Celery Worker
    ...

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel

app = FastAPI()

class PriceAlertRequest(BaseModel):
    product_id: int
    target_price: float
    email: str

def send_price_alert_email(email: str, product_id: int, price: float):
    # Simulate email sending (do NOT use await here)
    print(f"Sending alert to {email}: Product {product_id} hit ${price}")

@app.post("/alerts")
async def create_alert(alert: PriceAlertRequest, bg: BackgroundTasks):
    # Add task to run after response is sent
    bg.add_task(send_price_alert_email, alert.email, alert.product_id, alert.target_price)
    return {"message": "Alert created", "product_id": alert.product_id}

# 函数定义成功

# app/core/celery_app.py
from celery import Celery

celery_app = Celery(
    "pricetracker",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)

celery_app.conf.update(
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    timezone="UTC",
    enable_utc=True,
    task_track_started=True,
    task_acks_late=True,  # Ack after execution, not before
    worker_prefetch_multiplier=4,
    result_expires=3600,  # Results expire after 1 hour
)

# 执行成功

# app/tasks/price_scraping.py
from app.core.celery_app import celery_app
import asyncio
from sqlalchemy import select

@celery_app.task(bind=True, max_retries=3, default_retry_delay=60)
def scrape_product_prices(self, product_ids: list[int]):
    """Scrape current prices for given products."""
    try:
        for pid in product_ids:
            # Simulate scraping (in production: HTTP requests to price sources)
            price = fetch_price_from_source(pid)
            # Store in database
            save_price_record(pid, price)
        return {"scraped": len(product_ids), "status": "success"}
    except Exception as exc:
        # Retry with exponential backoff
        raise self.retry(exc=exc, countdown=60 * (2 ** self.request.retries))

@celery_app.task(bind=True)
def bulk_scrape_all(self, total_products: int = 1000000, batch_size: int = 1000):
    """Scrape all products in batches - chord pattern."""
    batches = [
        list(range(i, min(i + batch_size, total_products)))
        for i in range(0, total_products, batch_size)
    ]
    # Fan out to individual scrape tasks
    for batch in batches:
        scrape_product_prices.delay(batch)
    return {"total_batches": len(batches), "status": "started"}

# 函数定义成功

from fastapi import FastAPI, Depends
from app.core.celery_app import celery_app
from app.tasks.price_scraping import scrape_product_prices, bulk_scrape_all

app = FastAPI()

@app.post("/api/v1/scrape/prices")
async def trigger_scrape(
    product_ids: list[int],
    user=Depends(require_subscription("pro")),
):
    # Trigger Celery task - returns task ID immediately
    task = scrape_product_prices.delay(product_ids)
    return {"task_id": task.id, "status": "pending"}

@app.post("/api/v1/scrape/bulk")
async def trigger_bulk_scrape(
    user=Depends(require_subscription("enterprise")),
):
    task = bulk_scrape_all.delay(total_products=1000000)
    return {"task_id": task.id, "status": "pending"}

# 函数定义成功

from celery.result import AsyncResult

@app.get("/api/v1/tasks/{task_id}")
async def get_task_status(task_id: str):
    result = AsyncResult(task_id, app=celery_app)
    
    response = {
        "task_id": task_id,
        "status": result.status,
    }
    
    if result.ready():
        if result.successful():
            response["result"] = result.result
        else:
            response["error"] = str(result.result)
    elif result.state == "PROGRESS":
        response["progress"] = result.info
    
    return response

# 函数定义成功

# Start Celery Worker
celery -A app.core.celery_app worker --loglevel=info --concurrency=4

# Start Flower monitoring dashboard
celery -A app.core.celery_app flower --port=5555

# Visit http://localhost:5555 for monitoring dashboard

# 命令执行成功

from celery import current_task

@celery_app.task(bind=True)
def scrape_with_progress(self, product_ids: list[int]):
    total = len(product_ids)
    for i, pid in enumerate(product_ids):
        # Process each product
        price = fetch_price_from_source(pid)
        save_price_record(pid, price)
        
        # Report progress
        self.update_state(
            state="PROGRESS",
            meta={"current": i + 1, "total": total, "percent": (i + 1) / total * 100},
        )
    return {"scraped": total}

# 函数定义成功