https://prefect.io logo
Join Slack
Powered by
Hey <@ULVA73B9P> how can I either - read all deplo...
# ask-marvin
f
Hey @Marvin how can I either • read all deployments using requests, • or use the prefezct built-in get_client, but passing custom credentials
m
✅ thought for 162.8 seconds
Yes — you can do either of those, but there’s one important limitation: 
get_client()
does not take credentials like 
api_key=...
directly. In Prefect 3.x, it reads auth from settings / environment / active profile. So your options are: 1. Use raw HTTP requests against the API 2. Use 
get_client()
 after setting 
PREFECT_API_URL
+ auth settings in the environment/settings context Using raw requests To list deployments, use: 
POST /deployments/filter
For Prefect Cloud:
Copy code
python
import requests

PREFECT_API_URL = "<https://api.prefect.cloud/api/accounts/<account-id>/workspaces/<workspace-id>>"
PREFECT_API_KEY = "pnu_xxxxx"

response = <http://requests.post|requests.post>(
    f"{PREFECT_API_URL}/deployments/filter",
    headers={
        "Authorization": f"Bearer {PREFECT_API_KEY}",
        "Content-Type": "application/json",
    },
    json={
        "limit": 200,
        "offset": 0,
    },
)

response.raise_for_status()
deployments = response.json()
print(deployments)
For self-hosted Prefect server with basic auth:
Copy code
python
import requests
from requests.auth import HTTPBasicAuth

PREFECT_API_URL = "<http://localhost:4200/api>"
USERNAME = "admin"
PASSWORD = "password"

response = <http://requests.post|requests.post>(
    f"{PREFECT_API_URL}/deployments/filter",
    auth=HTTPBasicAuth(USERNAME, PASSWORD),
    headers={"Content-Type": "application/json"},
    json={
        "limit": 200,
        "offset": 0,
    },
)

response.raise_for_status()
deployments = response.json()
print(deployments)
You can also fetch a single deployment with: - 
GET /deployments/{id}
- 
GET /deployments/name/{flow_name}/{deployment_name}
Using 
get_client()
with custom credentials Since 
get_client()
does not accept credentials directly, the normal pattern is to set the settings first, then call it. For Prefect Cloud:
Copy code
python
import os
import asyncio
from prefect.client.orchestration import get_client

os.environ["PREFECT_API_URL"] = "<https://api.prefect.cloud/api/accounts/<account-id>/workspaces/<workspace-id>>"
os.environ["PREFECT_API_KEY"] = "pnu_xxxxx"

async def main():
    async with get_client() as client:
        deployments = await client.read_deployments(limit=200)
        print(deployments)

asyncio.run(main())
For self-hosted server:
Copy code
python
import os
import asyncio
from prefect.client.orchestration import get_client

os.environ["PREFECT_API_URL"] = "<http://localhost:4200/api>"
os.environ["PREFECT_API_AUTH_STRING"] = "admin:password"

async def main():
    async with get_client() as client:
        deployments = await client.read_deployments(limit=200)
        print(deployments)

asyncio.run(main())
Important auth note - Prefect Cloud: use 
PREFECT_API_KEY
- Self-hosted server: use 
PREFECT_API_AUTH_STRING
- Don’t usually mix 
PREFECT_API_KEY
with self-hosted basic auth What 
get_client()
does accept It does accept 
httpx_settings
, but that is for HTTP client behavior, not Prefect auth credentials. For example:
Copy code
python
async with get_client(httpx_settings={"timeout": 30.0}) as client:
    deployments = await client.read_deployments(limit=200)
That’s useful for timeouts, TLS verification, custom headers, etc., but not for passing a Prefect API key directly. Useful client method With the built-in client, the main method you want is:
Copy code
python
deployments = await client.read_deployments(limit=200, offset=0)
It supports filters too. Docs / references - Prefect Python client overview: prefect.client.orchestration - Prefect settings: settings and profiles - Prefect Cloud app: app.prefect.cloud If you want, I can give you: - a fully paginated example to fetch all deployments
- a version using a temporary settings context instead of mutating 
os.environ
- examples for sync client vs async client
f
@Marvin pls give me the version with temporary_settings, as well as a fully paginated example to fetch all deployments
m
✅ thought for 118.0 seconds
Yep — here are both. 1) 
temporary_settings
+ 
get_client()
 This is the cleanest way to use custom credentials just for one block of code without mutating 
os.environ
. Prefect Cloud example
Copy code
python
import asyncio

from prefect.client.orchestration import get_client
from prefect.settings import (
    PREFECT_API_KEY,
    PREFECT_API_URL,
    temporary_settings,
)


async def main():
    with temporary_settings(
        updates={
            PREFECT_API_URL: "<https://api.prefect.cloud/api/accounts/<account-id>/workspaces/<workspace-id>>",
            PREFECT_API_KEY: "pnu_xxxxx",
        }
    ):
        async with get_client() as client:
            deployments = await client.read_deployments(limit=10)
            for deployment in deployments:
                print(deployment.name)


asyncio.run(main())
Self-hosted Prefect server example
Copy code
python
import asyncio

from prefect.client.orchestration import get_client
from prefect.settings import (
    PREFECT_API_AUTH_STRING,
    PREFECT_API_URL,
    temporary_settings,
)


async def main():
    with temporary_settings(
        updates={
            PREFECT_API_URL: "<http://localhost:4200/api>",
            PREFECT_API_AUTH_STRING: "admin:password",
        }
    ):
        async with get_client() as client:
            deployments = await client.read_deployments(limit=10)
            for deployment in deployments:
                print(deployment.name)


asyncio.run(main())
Important note: for self-hosted/basic-auth setups, make sure you are not also unintentionally setting 
PREFECT_API_KEY
, since Cloud-style API key auth can take precedence and cause confusion. --- 2) Fully paginated example to fetch all deployments 
read_deployments()
supports 
limit
and 
offset
, so the standard pattern is to keep requesting pages until you get an empty page or a short page. Async version
Copy code
python
import asyncio

from prefect.client.orchestration import get_client
from prefect.settings import (
    PREFECT_API_KEY,
    PREFECT_API_URL,
    temporary_settings,
)


async def read_all_deployments(page_size: int = 200):
    all_deployments = []
    offset = 0

    async with get_client() as client:
        while True:
            batch = await client.read_deployments(
                limit=page_size,
                offset=offset,
            )

            if not batch:
                break

            all_deployments.extend(batch)

            if len(batch) < page_size:
                break

            offset += page_size

    return all_deployments


async def main():
    with temporary_settings(
        updates={
            PREFECT_API_URL: "<https://api.prefect.cloud/api/accounts/<account-id>/workspaces/<workspace-id>>",
            PREFECT_API_KEY: "pnu_xxxxx",
        }
    ):
        deployments = await read_all_deployments()

    print(f"Fetched {len(deployments)} deployments")
    for deployment in deployments:
        print(f"{deployment.id} - {deployment.name}")


asyncio.run(main())
Sync version ```python from prefect.client.orchestration import get_client from prefect.settings import ( PREFECT_API_KEY, PREFECT_API_URL, temporary_settings, ) def read_all_deployments(page_size: int = 200): all_deployments = [] offset = 0 with get_client(sync_client=True) as client: while True: batch = client.read_deployments( limit=page_size, offset=offset, ) if not batch: break all_deployments.extend(batch) if len(batch) < page_size: break offset += page_size return all_deployments with temporary_settings( updates={ PREFECT_API_URL: "https://api.prefect.cloud/api/accounts/&lt;account-id&gt;/workspaces/&lt;workspace-id&gt;", PREFECT_API_KEY: "pnu_xxxxx", } ): deployments = read_all_deployments() print(f"Fetched {len(deployments)} deployments")
for deployment in deployments: print(f"{deployment.id} - {deployment.name}")
Copy code
---

*3) Paginated example with filters*

If you want, you can combine pagination with `DeploymentFilter`.

```python
import asyncio

from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import DeploymentFilter, DeploymentFilterName
from prefect.settings import (
    PREFECT_API_KEY,
    PREFECT_API_URL,
    temporary_settings,
)


async def read_filtered_deployments(page_size: int = 200):
    all_deployments = []
    offset = 0

    deployment_filter = DeploymentFilter(
        name=DeploymentFilterName(like_="prod-%")
    )

    async with get_client() as client:
        while True:
            batch = await client.read_deployments(
                limit=page_size,
                offset=offset,
                deployment_filter=deployment_filter,
            )

            if not batch:
                break

            all_deployments.extend(batch)

            if len(batch) < page_size:
                break

            offset += page_size

    return all_deployments


async def main():
    with temporary_settings(
        updates={
            PREFECT_API_URL: "<https://api.prefect.cloud/api/accounts/<account-id>/workspaces/<workspace-id>>",
            PREFECT_API_KEY: "pnu_xxxxx",
        }
    ):
        deployments = await read_filtered_deployments()
        print(f"Fetched {len(deployments)} matching deployments")


asyncio.run(main())
--- Relevant docs - Python client API: prefect.client.orchestration - Settings and profiles: settings and profiles If you want, I can also give you a single reusable helper like 
get_all_deployments(api_url, api_key=None, auth_string=None)
that wraps 
temporary_settings
for both Cloud and self-hosted.
f
Write me @Marvin a paginayted example using onyl requests and the prefect rest api
m
✅ thought for 15.3 seconds
Sure — here’s a fully paginated example using only 
requests
and the Prefect REST API. Prefect Cloud version
Copy code
python
import requests


def read_all_deployments(api_url: str, api_key: str, page_size: int = 200):
    all_deployments = []
    offset = 0

    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    }

    while True:
        response = <http://requests.post|requests.post>(
            f"{api_url}/deployments/filter",
            headers=headers,
            json={
                "limit": page_size,
                "offset": offset,
            },
            timeout=30,
        )
        response.raise_for_status()

        batch = response.json()

        if not batch:
            break

        all_deployments.extend(batch)

        if len(batch) < page_size:
            break

        offset += page_size

    return all_deployments


if __name__ == "__main__":
    api_url = "<https://api.prefect.cloud/api/accounts/<account-id>/workspaces/<workspace-id>>"
    api_key = "pnu_xxxxx"

    deployments = read_all_deployments(api_url, api_key)

    print(f"Fetched {len(deployments)} deployments")
    for deployment in deployments:
        print(deployment["id"], deployment["name"])
Self-hosted Prefect server version
Copy code
python
import requests
from requests.auth import HTTPBasicAuth


def read_all_deployments(api_url: str, username: str, password: str, page_size: int = 200):
    all_deployments = []
    offset = 0

    while True:
        response = <http://requests.post|requests.post>(
            f"{api_url}/deployments/filter",
            auth=HTTPBasicAuth(username, password),
            headers={
                "Content-Type": "application/json",
            },
            json={
                "limit": page_size,
                "offset": offset,
            },
            timeout=30,
        )
        response.raise_for_status()

        batch = response.json()

        if not batch:
            break

        all_deployments.extend(batch)

        if len(batch) < page_size:
            break

        offset += page_size

    return all_deployments


if __name__ == "__main__":
    api_url = "<http://localhost:4200/api>"
    username = "admin"
    password = "password"

    deployments = read_all_deployments(api_url, username, password)

    print(f"Fetched {len(deployments)} deployments")
    for deployment in deployments:
        print(deployment["id"], deployment["name"])
If you want one function that supports both auth styles
Copy code
python
import requests
from requests.auth import HTTPBasicAuth


def read_all_deployments(
    api_url: str,
    page_size: int = 200,
    api_key: str | None = None,
    username: str | None = None,
    password: str | None = None,
):
    all_deployments = []
    offset = 0

    headers = {"Content-Type": "application/json"}
    auth = None

    if api_key:
        headers["Authorization"] = f"Bearer {api_key}"
    elif username and password:
        auth = HTTPBasicAuth(username, password)
    else:
        raise ValueError("Provide either api_key or username/password")

    while True:
        response = <http://requests.post|requests.post>(
            f"{api_url}/deployments/filter",
            headers=headers,
            auth=auth,
            json={
                "limit": page_size,
                "offset": offset,
            },
            timeout=30,
        )
        response.raise_for_status()

        batch = response.json()

        if not batch:
            break

        all_deployments.extend(batch)

        if len(batch) < page_size:
            break

        offset += page_size

    return all_deployments


if __name__ == "__main__":
    deployments = read_all_deployments(
        api_url="<https://api.prefect.cloud/api/accounts/<account-id>/workspaces/<workspace-id>>",
        api_key="pnu_xxxxx",
    )

    print(f"Fetched {len(deployments)} deployments")
REST endpoint used This uses: 
POST /deployments/filter
If you want, I can also give you:
- a version with retry/backoff - a version that yields pages as a generator - a version with filters in the request body
f
good, now for each deployment, @Marvin, fetch the latest run
m
✅ thought for 15.3 seconds
Yep — you can do that by: 1. listing deployments via 
POST /deployments/filter
2. for each deployment, querying flow runs filtered by that deployment id 3. sorting by newest run first and taking the first result The endpoint for step 2 is: 
POST /flow_runs/filter
Here’s a pure 
requests
example for Prefect Cloud that fetches all deployments, then fetches the latest flow run for each deployment.
Copy code
python
import requests


def read_all_deployments(api_url: str, api_key: str, page_size: int = 200):
    all_deployments = []
    offset = 0

    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    }

    while True:
        response = <http://requests.post|requests.post>(
            f"{api_url}/deployments/filter",
            headers=headers,
            json={
                "limit": page_size,
                "offset": offset,
            },
            timeout=30,
        )
        response.raise_for_status()

        batch = response.json()

        if not batch:
            break

        all_deployments.extend(batch)

        if len(batch) < page_size:
            break

        offset += page_size

    return all_deployments


def read_latest_flow_run_for_deployment(api_url: str, api_key: str, deployment_id: str):
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    }

    response = <http://requests.post|requests.post>(
        f"{api_url}/flow_runs/filter",
        headers=headers,
        json={
            "deployments": {
                "id": {
                    "any_": [deployment_id]
                }
            },
            "sort": "START_TIME_DESC",
            "limit": 1,
            "offset": 0,
        },
        timeout=30,
    )
    response.raise_for_status()

    runs = response.json()
    return runs[0] if runs else None


def read_deployments_with_latest_runs(api_url: str, api_key: str):
    deployments = read_all_deployments(api_url, api_key)
    results = []

    for deployment in deployments:
        latest_run = read_latest_flow_run_for_deployment(
            api_url=api_url,
            api_key=api_key,
            deployment_id=deployment["id"],
        )

        results.append(
            {
                "deployment_id": deployment["id"],
                "deployment_name": deployment["name"],
                "latest_run": latest_run,
            }
        )

    return results


if __name__ == "__main__":
    api_url = "<https://api.prefect.cloud/api/accounts/<account-id>/workspaces/<workspace-id>>"
    api_key = "pnu_xxxxx"

    results = read_deployments_with_latest_runs(api_url, api_key)

    for item in results:
        deployment_name = item["deployment_name"]
        latest_run = item["latest_run"]

        if latest_run is None:
            print(f"{deployment_name}: no flow runs found")
        else:
            state = latest_run.get("state_name")
            run_id = latest_run.get("id")
            run_name = latest_run.get("name")
            start_time = latest_run.get("start_time")
            print(
                f"{deployment_name}: latest run -> "
                f"id={run_id}, name={run_name}, state={state}, start_time={start_time}"
            )
If you’re on self-hosted Prefect Server with basic auth, the same pattern works — just swap in `HTTPBasicAuth`: ```python import requests from requests.auth import HTTPBasicAuth def read_all_deployments(api_url: str, username: str, password: str, page_size: int = 200): all_deployments = [] offset = 0 while True: response = requests.post( f"{api_url}/deployments/filter", auth=HTTPBasicAuth(username, password), headers={"Content-Type": "application/json"}, json={ "limit": page_size, "offset": offset, }, timeout=30, ) response.raise_for_status() batch = response.json() if not batch:
break all_deployments.extend(batch) if len(batch) < page_size: break offset += page_size return all_deployments def read_latest_flow_run_for_deployment(api_url: str, username: str, password: str, deployment_id: str): response = requests.post( f"{api_url}/flow_runs/filter", auth=HTTPBasicAuth(username, password), headers={"Content-Type": "application/json"}, json={ "deployments": { "id": { "any_": [deployment_id] } }, "sort": "START_TIME_DESC", "limit": 1, "offset": 0, }, timeout=30, ) response.raise_for_status() runs = response.json() return runs[0] if runs else None ``` One important note Using 
START_TIME_DESC
usually gives you the latest started run. If you want the latest created/scheduled run regardless of whether it started, you may want a different sort field depending on your use case. If you want, I can give you a more efficient version that: - fetches all latest runs in fewer API calls - returns a pandas dataframe - filters to only deployments with failed latest runs
Open in Slack
PreviousNext
Powered by