רפיד היא ספריית פייתון להרצת משימות במקביל. היא דומה ל Celery אבל פופולרית פחות ואסינכרונית, ודומה ל dramatiq הפעם אסינכרונית כמוה, חדשה יותר ופחות פופולרית.

התכונות הטובות של Repid כוללות:

1. משימות אסינכרוניות (כמובן, בשביל זה באנו).

2. תמיכה טובה בבדיקות עם רכיב TestClient.

3. תיעוד ידידותי לסוכנים עם קבצי llms.txt ו llms-full.txt.

4. חיבור מובנה ל pydantic בשביל הגדרת טיפוסים.

דוגמה? בטח בואו נכתוב תוכנית פייתון שתשמור פוסטים מהבלוג הזה לקריאה אופליין.

1. מבנה המערכת docker-compose.yml

תחנה ראשונה היא מבנה המערכת. מאחר ורפיד הוא מערכת להרצת משימות בצורה מבוזרת אשתמש ב docker-compose.yml כדי לייצר מספר קונטיינרים. זה הקובץ:

services:
  redis:
    image: redis:7-alpine
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 3s
      retries: 5

  worker-1:
    image: ghcr.io/astral-sh/uv:python3.14-bookworm
    working_dir: /app
    depends_on:
      redis:
        condition: service_healthy
    command: sh -c "uv sync --frozen --no-dev && uv run python worker_repid.py"
    volumes:
      - .:/app
      - ./data:/app/data
      - /app/.venv
    environment:
      - REPID_REDIS_URL=redis://redis:6379/0

  worker-2:
    image: ghcr.io/astral-sh/uv:python3.14-bookworm
    working_dir: /app
    depends_on:
      redis:
        condition: service_healthy
    command: sh -c "uv sync --frozen --no-dev && uv run python worker_repid.py"
    volumes:
      - .:/app
      - ./data:/app/data
      - /app/.venv
    environment:
      - REPID_REDIS_URL=redis://redis:6379/0

  dispatcher:
    image: ghcr.io/astral-sh/uv:python3.14-bookworm
    working_dir: /app
    depends_on:
      redis:
        condition: service_healthy
    command: sh -c "uv sync --frozen --no-dev && uv run python dispatcher_repid.py ${MAX_POSTS:-10}"
    volumes:
      - .:/app
      - ./data:/app/data
      - /app/.venv
    environment:
      - REPID_REDIS_URL=redis://redis:6379/0
    profiles:
      - dispatch

מתאם העבודה הוא רדיס בקונטיינר הראשון. קונטיינרים 2 ו 3 הם הפועלים וקונטיינר רביעי שולח את המשימות המתוזמנות.

2. קוד הפועלים

הסקריפט של הפועלים ממש פשוט:

import asyncio

from repid import Worker

from app_repid import app, router, QUEUE


async def run_worker():
    async with app.magic(auto_disconnect=True):
        # auto_declare=True (default) makes the worker declare the queue
        # before it starts consuming.
        # tasks_limit caps how many actor coroutines run concurrently on this
        # worker's event loop -> at most 10 concurrent downloads per worker.
        worker = Worker(routers=[router], tasks_limit=10)
        await worker.run()


if __name__ == "__main__":
    asyncio.run(run_worker())

הוא מריץ פונקציה של רפיד ומגדיר הגבלה של 10 משימות במקביל. קוד המשימה עצמה מוגדר בקובץ app_repid.py באותה תיקיה. זה תוכן הקובץ:

"""
Repid application: defines the broker connection, the router and the
two actors (download_index_page, download_post).

This module is imported by both the worker (worker_repid.py) and the
dispatcher (dispatcher_repid.py). The actors are the repid equivalent of
the Celery tasks in tasks.py.
"""

import os
import httpx
from bs4 import BeautifulSoup
from repid import Repid, Connection, RedisMessageBroker, Job, Router

BROKER_URL = os.environ.get("REPID_REDIS_URL", "redis://localhost:6379/0")

# A single Repid app wraps the connection to the message broker (Redis).
# Entering `app.magic()` makes this connection the implicit one used by
# Job().enqueue(), Queue().declare() and the Worker.
app = Repid(Connection(RedisMessageBroker(BROKER_URL)))

# Shared async HTTP client. Because the actors `await` it, many downloads can
# be in flight on the worker's event loop at once (up to the worker's
# tasks_limit). follow_redirects mirrors requests' default behavior.
client = httpx.AsyncClient(timeout=30, follow_redirects=True)

router = Router()

QUEUE = "crawler"
BASE_URL = "https://www.tocode.co.il/blog"
DATA_DIR = "/app/data"


def to_markdown_url(post_url: str) -> str:
    """Convert a blog post URL to its markdown version by stripping
    the query string and appending .md."""
    base = post_url.split("?")[0]
    return f"{base}.md"


def extract_post_urls(soup: BeautifulSoup, limit: int):
    """
    Generator that yields up to `limit` unique blog post URLs
    from a blog index page BeautifulSoup parse tree.

    Blog post URLs are like /blog/<date-slug>?page=1
    Index page is /blog?page=N (no trailing slash) — excluded by startswith.
    """
    seen = set()
    count = 0
    for a_tag in soup.select("a[href]"):
        href = a_tag["href"]
        if href.startswith("/blog/") and href != "/blog/":
            full_url = f"https://www.tocode.co.il{href}"
            if full_url not in seen:
                seen.add(full_url)
                yield full_url
                count += 1
                if count >= limit:
                    return


@router.actor(queue=QUEUE)
async def download_index_page(page_num: int, posts_limit: int) -> dict:
    """
    Download a blog index page, extract post links, and enqueue a
    download_post job for each post found.

    posts_limit: how many posts to download from this page (max 10).
    """
    url = f"{BASE_URL}?page={page_num}"
    resp = await client.get(url)
    resp.raise_for_status()

    soup = BeautifulSoup(resp.text, "html.parser")

    for post_url in extract_post_urls(soup, posts_limit):
        md_url = to_markdown_url(post_url)
        # The worker holds the active (magic) connection, so jobs enqueued
        # from inside an actor reuse it automatically.
        await Job(
            name="download_post",
            queue=QUEUE,
            args={"md_url": md_url},
            retries=3,
        ).enqueue()

    return {"page": page_num}


@router.actor(queue=QUEUE)
async def download_post(md_url: str) -> dict:
    """
    Download a single blog post markdown file and save it to the data folder.
    The URL is already the .md version (converted by the dispatcher).
    """
    resp = await client.get(md_url)
    resp.raise_for_status()

    # Derive filename from the URL slug: /blog/slug.md → slug
    slug = md_url.rstrip("/").split("/")[-1].removesuffix(".md")
    filename = f"{slug}.md"
    filepath = os.path.join(DATA_DIR, filename)

    os.makedirs(DATA_DIR, exist_ok=True)
    with open(filepath, "w", encoding="utf-8") as f:
        f.write(resp.text)

    return {"file": filename, "size_bytes": len(resp.text)}

כל פונקציה שמסומת בתור actor מגדירה משימה. אצלנו בתוכנית אלה הפונקציות download_index_page ו download_post. משימות מוגדרות על ה Router וזכרו שכל Worker מקבל ביצירה את ה Router-ים לקחת מהם משימות:

worker = Worker(routers=[router], tasks_limit=10)

המשימות עצמן הן פשוט קוד שמוריד דף אינדקס (רשימה של פוסטים) או פוסט ספציפי ומה שחשוב לגביהן שיהיו פונקציות אסינכרוניות.

מעניין לשים לב בקוד המשימה שמורידה את דף האינדקס איך היא מייצרת משימות חדשות כדי להוריד את הפוסטים:

    for post_url in extract_post_urls(soup, posts_limit):
        md_url = to_markdown_url(post_url)
        # The worker holds the active (magic) connection, so jobs enqueued
        # from inside an actor reuse it automatically.
        await Job(
            name="download_post",
            queue=QUEUE,
            args={"md_url": md_url},
            retries=3,
        ).enqueue()

    return {"page": page_num}

3. שיגור המשימות

החלק האחרון של המערכת הוא הקונטיינר שמשגר את המשימות. למעשה המשימה היחידה שאנחנו צריכים להפעיל היא הורדת דף האינדקס כי הפונקציה שלו כבר מפעילה את משימת הורדת הפוסטים כמו שקודם ראינו.

זה הקוד של dispatcher_repid.py שמתחיל הורדה של דפי האינדקס במקביל:

"""
Dispatcher (repid version): calculates how many blog index pages are needed
and enqueues download_index_page jobs via repid.

Usage:
    uv run python dispatcher_repid.py <max_posts>

Example:
    uv run python dispatcher_repid.py 25   # downloads up to 25 posts
"""

import asyncio
import math
import sys

from repid import Job, Queue

from app_repid import app, QUEUE


async def main():
    if len(sys.argv) < 2:
        print("Usage: python dispatcher_repid.py <max_posts>")
        sys.exit(1)

    max_posts = int(sys.argv[1])
    pages_needed = math.ceil(max_posts / 10)

    print(f"Downloading up to {max_posts} posts across {pages_needed} index page(s)")

    async with app.magic(auto_disconnect=True):
        # Declare the queue so jobs can be enqueued before any worker starts.
        await Queue(QUEUE).declare()

        for page_num in range(1, pages_needed + 1):
            # Each page has 10 posts; the last page may have fewer
            already_allocated = (page_num - 1) * 10
            posts_limit = min(10, max_posts - already_allocated)

            await Job(
                name="download_index_page",
                queue=QUEUE,
                args={"page_num": page_num, "posts_limit": posts_limit},
                retries=3,
            ).enqueue()
            print(f"  Dispatched page {page_num} (limit={posts_limit})")

    print("All index pages dispatched. Workers will now process the jobs.")


if __name__ == "__main__":
    asyncio.run(main())

4. סיכום - מה מוצלח בדוגמה. מה עוד צריך לשפר.

הקוד מראה איך להוריד במקביל פוסטים מהבלוג. הקוד מקבל מספר פוסטים להורדה, מחשב כמה דפים צריך לשמור ואז מוציא משימות במקביל:

1. משימה לכל דף שצריך להוריד.

2. בזמן שדפי האינדקס יורדים, כל דף שסיים יוצר משימות חדשות להורדת הפוסטים שהופיעו בו.

הכל אסינכרוני והכל מוגבל בעד 10 משימות לפועל ושני פועלים, סך הכל עד 20 הורדות במקביל מהאתר.

הדוגמה לא כוללת Type Safety על המשימות עצמן. לרפיד יש אינטגרציה עם pydantic אבל עדיין לא ניסיתי אותה אז זה יהיה נושא לפוסט המשך.