Prefect 1.0 -> Prefect 2.0 Graphql API to REST API

Summary

Prefect supports an API that allows users the ability to provide information about and in some cases manipulate prefect objects, flows/tasks/etc.

In Prefect 1.0 this was supported through a GraphQL API while being flexible for the end user it could often be confusing making it not particularly user friendly in addition to being fairly heavy, this is why tje Prefect Server container is so large.

In Prefect 2.0 this is supported by a RestAPI. There were many reasons for this change we won’t discuss all of them here however here are a couple of notable reasons behind the change. First and foremost many users are more familiar with utilizing a RestAPI making it more accessible. Second the RestAPI is far more lightweight so less work is required to interact with it both. Our CTO Chris White explains a bit more in detail here in this video from a previous livestream.

Video Link

Audience

If you’re looking for guidance on how to get started working with 2.0’s RestAPI then this article is for you.

What is Staying the Same

Many of the same or similar Mutations are still supported with the RestAPI through various endpoints. Users are still able to utilize the prefect client to interact with the api in their code.

What is Different

Users are no longer required to construct a Graphql query to interact with the api instead they can make calls to specified endpoints each with their own supported parameters, filters, and sorting behavior. The Prefect 2.0 API is asynchronous and calls to it must be awaited. We do not currently support an interactive API within the Prefect 2.0 UI.

How to Convert from 1.0 to 2.0

Below are two examples of submitting calls to the GraphQl API and RestAPI respectively to get the last 5 flow runs sorted by time:

from prefect import Client
import os

api_key = os.getenv("api_key")

client = Client(api_key=api_key)


test =client.graphql(
            {
                "query": {
                    "flow_run(limit: 5, order_by: { created: desc })": {
                        'id',
                        "created"
                    }
                }
            }
        )
        

print(test)
from prefect import task
from prefect.client import get_client
from prefect import task, flow
from prefect.orion.schemas.sorting import FlowRunSort
import asyncio


async def get_flow_runs():
    client = get_client()
    r = await client.read_flow_runs(limit=5, sort=FlowRunSort.END_TIME_DESC)
    return r 



r = asyncio.run(get_flow_runs())

for flow in r:
    print(flow.name ,flow.flow_id, flow.created)

Note that in 2.0 the api call returns a list for flow runs schemas for each flow which can be used to easily reference the values stored in each schema. The Current API context is inferred from the currently selected profile in your environment.

Links

Troubleshooting

Since the Prefect 2.0 client is asynchronous it is necessary to await any calls made with the client you can reference pythons docs for more information on async functions Coroutines and Tasks — Python 3.10.6 documentation,

1 Like