Quickstart

Convert a Python script into your first Prefect workflow

Prefect makes it easy to deploy Python scripts, run them on a schedule, make them robust to failure, and observe them in a UI.

To do this, you need to perform the following tasks:

1. Install Prefect
2. Connect to a Prefect API (self-hosted or Prefect Cloud)
3. Add decorators to the methods in the script

Let’s start with a Python script that fetches the number of stars for a list of GitHub repositories.

Show the Python script

my_script.py

import httpx


def show_stars(github_repos: list[str]):
    """Show the number of stars that GitHub repos have"""

    for repo in github_repos:
        repo_stats = fetch_stats(repo)
        stars = get_stars(repo_stats)
        print(f"{repo}: {stars} stars")


def fetch_stats(github_repo: str):
    """Fetch the statistics for a GitHub repo"""

    return httpx.get(f"https://api.github.com/repos/{github_repo}").json()


def get_stars(repo_stats: dict):
    """Get the number of stars from GitHub repo statistics"""

    return repo_stats['stargazers_count']


if __name__ == "__main__":
    show_stars([
        "PrefectHQ/prefect",
        "pydantic/pydantic",
        "huggingface/transformers"
    ])

Install Prefect

To install Prefect with pip, run:

pip install -U prefect

See Install Prefect for more details on installation.

Connect to a Prefect API

Connect to a Prefect API:

• Self-hosted
• Prefect Cloud

1. Start a local API server:

   prefect server start
   

2. Open the Prefect dashboard in your browser at http://localhost:4200.

Convert your script to a Prefect workflow

Decorators are the easiest way to convert a Python script into a workflow.

1. Add a @flow decorator to the script’s entrypoint.
2. Add @task decorators to any methods called by the flow.

This will create a flow and corresponding tasks. Tasks receive metadata about upstream dependencies and the state of those dependencies before they run. Prefect records these dependencies and states as it orchestrates these tasks.

my_workflow.py

import httpx

from prefect import flow, task # Prefect flow and task decorators


@flow(log_prints=True)
def show_stars(github_repos: list[str]):
    """Flow: Show the number of stars that GitHub repos have"""

    for repo in github_repos:
        # Call Task 1
        repo_stats = fetch_stats(repo)

        # Call Task 2
        stars = get_stars(repo_stats)

        # Print the result
        print(f"{repo}: {stars} stars")


@task
def fetch_stats(github_repo: str):
    """Task 1: Fetch the statistics for a GitHub repo"""

    return httpx.get(f"https://api.github.com/repos/{github_repo}").json()


@task
def get_stars(repo_stats: dict):
    """Task 2: Get the number of stars from GitHub repo statistics"""

    return repo_stats['stargazers_count']


# Run the flow
if __name__ == "__main__":
    show_stars([
        "PrefectHQ/prefect",
        "pydantic/pydantic",
        "huggingface/transformers"
    ])

The log_prints=True argument provided to the @flow decorator automatically converts any print statements within the function to INFO level logs.

Run your flow

You can run your Prefect flow just as you would a Python script:

python my_workflow.py

The output in your terminal should look similar to this:

08:21:31.335 | INFO    | prefect.engine - Created flow run 'attentive-kestrel' for flow 'show-stars'
08:21:31.336 | INFO    | prefect.engine - View at http://127.0.0.1:4200/runs/flow-run/edf6866f-371d-4e51-a9e3-556a525b1146
08:21:31.731 | INFO    | Task run 'fetch_stats-dce' - Finished in state Completed()
08:21:31.775 | INFO    | Task run 'get_stars-585' - Finished in state Completed()
08:21:31.776 | INFO    | Flow run 'attentive-kestrel' - PrefectHQ/prefect: 17318 stars
08:21:32.089 | INFO    | Task run 'fetch_stats-e16' - Finished in state Completed()
08:21:32.118 | INFO    | Task run 'get_stars-756' - Finished in state Completed()
08:21:32.119 | INFO    | Flow run 'attentive-kestrel' - pydantic/pydantic: 186318 stars
08:21:32.409 | INFO    | Task run 'fetch_stats-b62' - Finished in state Completed()
08:21:32.440 | INFO    | Task run 'get_stars-8ad' - Finished in state Completed()
08:21:32.441 | INFO    | Flow run 'attentive-kestrel' - huggingface/transformers: 134848 stars
08:21:32.469 | INFO    | Flow run 'attentive-kestrel' - Finished in state Completed()

Prefect automatically tracks the state of the flow run and logs the output, which can be viewed directly in the terminal or in the UI.

Next steps

In this tutorial, you successfully converted a Python script to a deployable workflow tracked by Prefect.

Next, get this workflow off of your laptop and run it automatically on a schedule.

Need help? Book a meeting with a Prefect Product Advocate to get your questions answered.