اگر هر روز اسکریپت‌های پایتون را با cron اجرا می‌کنید، لاگ‌ها را دستی چک می‌کنید و با یک خطا کل پایپ‌لاین داده‌تان می‌خوابد، Prefect دقیقاً برای همین ساخته شده است. این فریم‌ورک مدرن پایتون به شما اجازه می‌دهد هر تابع پایتون را به یک جریان کار (Workflow) قابل مشاهده، زمان‌بندی‌شونده و مقاوم در برابر خطا تبدیل کنید — بدون پیچیدگی‌های Airflow و بدون نیاز به دیتابیس جداگانه برای شروع. در این راهنما یاد می‌گیرید Prefect را در کمتر از ۱۰ دقیقه نصب و اولین Flow خود را اجرا کنید.

Prefect در پایتون چیست و چرا جایگزین Airflow شده است؟

Prefect یک فریم‌ورک ارکستریشن پایتون است که توابع عادی شما را با دکوراتورهای @flow و @task به جریان‌های کاری قابل پیگیری تبدیل می‌کند؛ برخلاف Airflow نیازی به DAGهای سنگین یا Scheduler جداگانه ندارد، به‌صورت محلی با یک دستور اجرا می‌شود، داشبورد زنده، retry خودکار و زمان‌بندی Cron دارد و برای پایپ‌لاین‌های داده، ETL و تسک‌های هوش مصنوعی ایده‌آل است. در ادامه نصب، مفاهیم و یک مثال کامل را گام‌به‌گام می‌بینید.

فهرست مطالب

پیش‌نیازها و نصب Prefect
مفاهیم کلیدی Prefect: Flow و Task
اولین Flow خود را بسازید — مثال کامل و قابل اجرا
زمان‌بندی، پارامتردهی و اجرای خودکار
مشاهده و مانیتورینگ در داشبورد Prefect
مقایسه Prefect با Airflow و Celery
اشتباهات رایج هنگام کار با Prefect
سوالات متداول
جمع‌بندی

پیش‌نیازها و نصب Prefect

برای شروع فقط به پایتون ۳.۹ به بالا و یک محیط مجازی نیاز دارید. Prefect 2.x کاملاً با پایتون استاندارد کار می‌کند و وابستگی سنگینی ندارد. اگر با پروژه‌های پایتون آشنا هستید، نصب آن شبیه هر کتابخانه دیگری است، اما یک نکته مهم وجود دارد: سرور Prefect به‌صورت محلی روی سیستم شما اجرا می‌شود و برای توسعه نیازی به Docker یا Postgres ندارید.

پیش‌نیازها:

پایتون ۳.۹، ۳.۱۰، ۳.۱۱ یا ۳.۱۲
محیط مجازی (venv یا uv)
ترمینال و ویرایشگر کد (VS Code پیشنهادی)
آشنایی اولیه با تابع و دکوراتور در پایتون

نصب در محیط مجازی با pip و uv

بهترین روش، نصب در محیط ایزوله است. اگر از uv استفاده می‌کنید که این روزها جایگزین سریع pip شده، سرعت نصب چند برابر می‌شود.

# ساخت محیط مجازی
python -m venv venv
source venv/bin/activate  # ویندوز: venv\Scripts\activate

# نصب Prefect
pip install -U prefect
# یا با uv
# uv pip install prefect

# بررسی نصب
prefect version
prefect server start

دستور prefect server start داشبورد محلی را روی http://127.0.0.1:4200 بالا می‌آورد. از این به بعد هر Flow که اجرا کنید، به‌صورت خودکار در همین داشبورد لاگ و وضعیتش ثبت می‌شود. برای توقف، کافی است ترمینال را ببندید؛ داده‌ها در پروفایل محلی ذخیره می‌ماند.

مفاهیم کلیدی Prefect: Flow و Task

تمام منطق Prefect روی دو مفهوم ساده می‌چرخد: Flow و Task. هر تابع پایتون که با @flow تزئین شود به یک جریان کاری تبدیل می‌شود و هر تابع با @task یک واحد کاری قابل retry و لاگ‌گیری است. همین سادگی باعث شده Prefect نسبت به Airflow بسیار شهودی‌تر باشد.

Flow: کانتینر اصلی کار شماست. کل پایپ‌لاین (مثلاً دانلود داده، پردازش، ذخیره) یک Flow است. Flow می‌تواند پارامتر بگیرد، زمان‌بندی شود و وضعیت کلی را گزارش دهد.
Task: قدم‌های کوچک و قابل استفاده مجدد. هر Task به‌صورت مستقل retry می‌شود، کش می‌شود و لاگ جداگانه دارد. اگر یک Task شکست بخورد، فقط همان تکرار می‌شود نه کل Flow.
State و Logging: هر اجرا یک State دارد (Completed, Failed, Retrying) و تمام print و loggerها به داشبورد می‌روند.

برای درک بهتر، اگر در پروژه‌های قبلی با اجرای تسک‌های پس‌زمینه با Celery و Redis کار کرده‌اید، Task در Prefect شبیه Task در Celery است، اما بدون نیاز به بروکر جداگانه و با قابلیت مشاهده بسیار بهتر.

تفاوت Flow و Task در عمل

قانون ساده: هر چیزی که باید زمان‌بندی یا مانیتور شود Flow است، و هر چیزی که باید retry یا کش شود Task است. مثلاً در یک پایپ‌لاین قیمت دلار:

Flow: fetch_and_store_currency
Taskها: fetch_api، clean_data، save_to_db

اگر API لحظه‌ای قطع شود، فقط fetch_api با تأخیر retry می‌شود و بقیه Flow منتظر می‌ماند — همین ویژگی خطاهای گذرا را بدون کدنویسی اضافه حل می‌کند.

اولین Flow خود را بسازید — مثال کامل و قابل اجرا

یک مثال واقعی و مینیاتوری می‌سازیم: دریافت قیمت تستی، تمیز کردن و ذخیره در فایل. این کد را در my_flow.py ذخیره کنید و اجرا کنید.

from prefect import flow, task
import random
import time

@task(retries=2, retry_delay_seconds=5, log_prints=True)
def fetch_price(symbol: str = "USD"):
    print(f"در حال دریافت قیمت {symbol}...")
    # شبیه‌سازی خطای گذرا
    if random.random() < 0.3:
        raise ValueError("خطای موقت شبکه!")
    return {"symbol": symbol, "price": 58250 + random.randint(-200, 200)}

@task(log_prints=True)
def clean_data(raw: dict):
    print(f"پردازش داده: {raw}")
    raw["price_toman"] = raw["price"] * 1  # تبدیل ساده
    raw["timestamp"] = time.time()
    return raw

@task(log_prints=True)
def save_data(data: dict):
    with open("prices.log", "a", encoding="utf-8") as f:
        f.write(str(data) + "\n")
    print("ذخیره شد!")

@flow(name="currency-pipeline", log_prints=True)
def currency_flow(symbol: str = "USD"):
    raw = fetch_price(symbol)
    cleaned = clean_data(raw)
    save_data(cleaned)
    return cleaned

if __name__ == "__main__":
    currency_flow("USD")

اجرا:

python my_flow.py
# یا با داشبورد
prefect server start
# در ترمینال دیگر
python my_flow.py

چه اتفاقی می‌افتد؟ اگر fetch_price شکست بخورد، Prefect خودکار تا ۲ بار با ۵ ثانیه فاصله retry می‌کند. در داشبورد می‌بینید هر Task چقدر طول کشیده، لاگ‌ها چیست و Flow در نهایت Completed شده یا نه. خروجی prices.log هر اجرا را ثبت می‌کند.

برای کسانی که قبلاً با ساخت RAG هوشمند با LangChain و Chroma کار کرده‌اند، می‌توانید همین Flow را توسعه دهید و یک Task برای embedding و ذخیره در Chroma اضافه کنید — Prefect به‌راحتی پایپ‌لاین هوش مصنوعی را ارکستر می‌کند.

زمان‌بندی، پارامتردهی و اجرای خودکار

قدرت اصلی Prefect در «Deploy» است. شما یک Flow را یک‌بار تعریف می‌کنید و بعد می‌گویید هر ۵ دقیقه، هر روز ساعت ۸ صبح، یا با Cron خاصی اجرا شود — بدون نوشتن cron دستی.

from prefect import flow

@flow
def daily_report(symbols: list = ["USD", "EUR"]):
    for s in symbols:
        currency_flow(s)

if __name__ == "__main__":
    # ساخت deployment با زمان‌بندی
    daily_report.serve(
        name="daily-8am",
        cron="0 8 * * *",  # هر روز ۸ صبح
        parameters={"symbols": ["USD", "EUR", "GOLD"]}
    )

دستور serve یا deploy فلو را به سرور معرفی می‌کند. همچنین می‌توانید از CLI استفاده کنید:

prefect deploy my_flow.py:currency_flow --name prod --cron "*/15 * * * *"

هر اجرا پارامترهایش در داشبورد قابل مشاهده و تغییر است؛ بدون نیاز به تغییر کد، می‌توانید همان Flow را با symbol="BTC" دوباره اجرا کنید.

زمان‌بندی با Deploy و Cron در محیط واقعی

در سرور واقعی، به جای serve از prefect worker start --pool default استفاده می‌کنید و Deploy را یک‌بار می‌سازید. Prefect Cloud هم همین API را دارد اما برای شروع، اجرای محلی کافی است. نکته امنیتی: هرگز کلید API یا رمز دیتابیس را در کد ننویسید؛ از prefect.blocks.system.Secret یا متغیر محیطی os.getenv استفاده کنید.

مشاهده و مانیتورینگ در داشبورد Prefect

بعد از prefect server start به http://127.0.0.1:4200 بروید. سه بخش کلیدی:

Flow Runs: لیست اجراها، زمان، مدت، وضعیت. روی هر Run کلیک کنید تا Taskها را ببینید.
Logs: تمام print و logger.infoها با زمان دقیق. برای دیباگ بی‌نظیر است.
Deployments & Schedules: زمان‌بندی‌های فعال و پارامترهایشان.

یک ترفند کاربردی: در Taskها از from prefect import get_run_logger استفاده کنید تا لاگ ساختاریافته داشته باشید:

from prefect import task, get_run_logger

@task
def process():
    logger = get_run_logger()
    logger.info("شروع پردازش با Prefect")
    logger.warning("این یک هشدار تستی است")

این لاگ‌ها در داشبورد قابل فیلتر و جستجو هستند و برای پایپ‌لاین‌های طولانی بسیار مفیدند.

مقایسه Prefect با Airflow و Celery

انتخاب ابزار به اندازه تیم و پیچیدگی بستگی دارد. جدول زیر تفاوت را خلاصه می‌کند:

معیار	Prefect 2.x	Apache Airflow	Celery + Redis
شروع و نصب	یک دستور، بدون دیتابیس	نیاز به Postgres, Scheduler, Webserver	نیاز به Redis/RabbitMQ
تعریف workflow	تابع پایتون با @flow	DAG پایتون با اپراتورها	Task تکی، بدون DAG
زمان‌بندی	داخلی با Cron/Interval	داخلی با DAG schedule	نیاز به Celery Beat
داشبورد	مدرن، لاگ زنده، پارامترمحور	قدیمی‌تر، DAG-centric	Flower محدود
مناسب برای	تیم کوچک تا متوسط، پایپ‌لاین داده و AI	سازمان بزرگ، DAGهای پیچیده	تسک‌های پس‌زمینه ساده
retry و کش	داخلی در سطح Task	در سطح Task اما پیچیده‌تر	داخلی اما بدون مشاهده DAG

اگر پایپ‌لاین شما ETL سبک، اسکرپینگ، یا ارکستریشن مدل‌های هوش مصنوعی است، Prefect سریع‌ترین راه رسیدن به نتیجه است. برای هزاران DAG وابسته در سازمان‌های بزرگ، Airflow هنوز انتخاب محافظه‌کارانه است.

اشتباهات رایج هنگام کار با Prefect

اجرای Flow بدون سرور: خیلی‌ها prefect server start را روشن نمی‌کنند و تعجب می‌کنند چرا داشبورد خالی است. سرور را در یک ترمینال جدا همیشه روشن نگه دارید یا از Prefect Cloud استفاده کنید.
استفاده از print به جای logger: با log_prints=True مشکل حل می‌شود، اما برای پروژه واقعی get_run_logger حرفه‌ای‌تر و قابل فیلتر است.
فراموشی retries برای APIهای ناپایدار: اگر Task شما با شبکه کار می‌کند، حتماً retries و retry_delay_seconds بگذارید؛ در غیر این صورت یک قطعی کوتاه کل Flow را Fail می‌کند.
قرار دادن کد سنگین در سطح ماژول: کدی که خارج از Flow/Task نوشته شود هنگام import اجرا می‌شود و Deploy را خراب می‌کند. همه منطق را داخل تابع ببرید.
هاردکد کردن Secrets: کلید API را مستقیم در کد نگذارید. از Blockهای Prefect یا .env و python-dotenv استفاده کنید.

سوالات متداول

آیا Prefect جایگزین کامل Airflow است؟

برای اکثر تیم‌های کوچک و متوسط بله؛ Prefect سبک‌تر، سریع‌تر و پایتونیک‌تر است. اما اگر سازمان شما صدها DAG با وابستگی پیچیده و نیاز به RBAC سازمانی دارد، Airflow بالغ‌تر است. بسیاری از تیم‌ها از Airflow به Prefect مهاجرت می‌کنند چون توسعه و دیباگ سریع‌تر می‌شود.

آیا Prefect بدون اینترنت کار می‌کند؟

بله، نسخه Open Source به‌صورت کاملاً محلی اجرا می‌شود و نیازی به Prefect Cloud ندارد. فقط prefect server start را اجرا کنید. Cloud فقط برای مانیتورینگ متمرکز، اعلان‌ها و اجرای توزیع‌شده مزیت دارد.

تفاوت Prefect با Celery چیست؟

Celery یک صف تسک توزیع‌شده است؛ هر Task مستقل است و DAG ندارد. Prefect یک ارکستریتور جریان کار است؛ Flowها و وابستگی‌ها را مدیریت، زمان‌بندی و مانیتور می‌کند. می‌توانید حتی Celery را به‌عنوان Task Runner داخل Prefect استفاده کنید، اما برای بیشتر کارها خود Prefect کافی است.

چطور Prefect را در سرور لینوکس همیشه روشن نگه دارم؟

از systemd یا tmux یا Docker استفاده کنید. ساده‌ترین روش: یک سرویس systemd برای prefect server start و یک سرویس دیگر برای prefect worker start --pool default بسازید. در Prefect 2 دستور prefect deploy فایل prefect.yaml می‌سازد که با prefect deploy --all قابل اجراست.

آیا Prefect با asyncio و تسک‌های ناهمگام سازگار است؟

بله، کاملاً. کافی است Flow یا Task را با async def تعریف کنید. Prefect خودش event loop را مدیریت می‌کند. برای ترکیب با کتابخانه‌هایی مثل HTTPX یا aiohttp این قابلیت عالی است.

جمع‌بندی

Prefect به شما اجازه می‌دهد بدون درگیری با زیرساخت سنگین، پایپ‌لاین‌های پایتون قابل اعتماد بسازید. چک‌لیست عملی برای شروع امروز:

یک محیط مجازی بسازید و pip install prefect را اجرا کنید.
سرور محلی را با prefect server start روشن کنید و داشبورد را باز کنید.
مثال currency_flow همین مقاله را کپی و اجرا کنید؛ retry را با قطع عمدی تست کنید.
یک Deployment با cron="*/15 * * * *" بسازید و اجرای زمان‌بندی‌شده را ببینید.
لاگ‌ها را از print به get_run_logger ارتقا دهید و Secrets را به Block منتقل کنید.
وقتی آماده شدید، Worker را روی سرور اصلی مستقر کنید یا به Prefect Cloud متصل شوید.

از همین امروز یک اسکریپت cron قدیمی را به Flow تبدیل کنید؛ تفاوت مشاهده‌پذیری و اطمینان را در همان هفته اول حس خواهید کرد.

آموزش Prefect در پایتون؛ مدیریت جریان کار مدرن