Prefect 실무 스터디 가이드

단계 1: 설치

# pip를 사용한 설치
pip install -U prefect

# 추가 패키지 (선택)
pip install prefect-aws prefect-dask prefect-sqlalchemy

# 버전 확인
prefect version

단계 2: 로컬 서버 시작 (옵션)

# Prefect 서버를 로컬에서 실행 (UI 포함)
prefect server start

# 별도의 터미널에서 서버 상태 확인
prefect config view

💡 로컬 서버 없이도 Prefect Cloud(app.prefect.cloud)를 무료로 사용할 수 있습니다.

단계 3: 첫 번째 Flow 작성

first_flow.py 파일 생성:

from prefect import flow, task
from typing import List
import httpx

@task(retries=3, retry_delay_seconds=10)
def get_repo_info(repo_owner: str, repo_name: str):
    """GitHub에서 레포지토리 정보를 가져오는 태스크"""
    url = f"https://api.github.com/repos/{repo_owner}/{repo_name}"
    response = httpx.get(url)
    response.raise_for_status()
    return response.json()

@task
def extract_stars(repo_info: dict) -> int:
    """레포지토리 정보에서 스타 수 추출"""
    return repo_info.get("stargazers_count", 0)

@flow(name="github-stars-tracker", log_prints=True)
def track_stars(repos: List[dict]):
    """여러 GitHub 레포지토리의 스타 수를 추적하는 메인 플로우"""
    total_stars = 0
    
    for repo in repos:
        print(f"Processing {repo['owner']}/{repo['name']}...")
        info = get_repo_info(repo["owner"], repo["name"])
        stars = extract_stars(info)
        total_stars += stars
        print(f"  Stars: {stars}")
    
    print(f"✅ Total stars across all repos: {total_stars}")
    return total_stars

# 로컬 실행
if __name__ == "__main__":
    # 테스트할 레포지토리 목록
    repositories = [
        {"owner": "PrefectHQ", "name": "prefect"},
        {"owner": "apache", "name": "airflow"},
        {"owner": "dagster-io", "name": "dagster"}
    ]
    
    # 플로우 실행
    result = track_stars(repositories)
    print(f"Flow 실행 완료! 반환값: {result}")

단계 4: 실행 및 관찰

# Python 스크립트로 직접 실행
python first_flow.py

# Prefect CLI를 통해 실행 (추적 가능)
prefect flow run --name "github-stars-tracker"

# UI에서 실행 결과 확인 (서버 실행 시)
# http://localhost:4200