How can I list all blocks across my workspaces?

Archive
, ,
1

Here’s an example of how you would list your blocks across all your workspaces using the Prefect API Python client. There’s two interesting bits to note:

  1. The API client is async, so you need some async flavor sprinkled in.
  2. The API client generally assumes that you’re working within a specific workspace, so we have to do a bit of extra work to make the list blocks call for each workspace.
import anyio

from prefect.client import OrionClient
from prefect.client.cloud import get_cloud_client
from prefect.settings import PREFECT_API_KEY


# https://github.com/PrefectHQ/prefect/blob/b063f84/src/prefect/cli/block.py#L171-L194
async def list_blocks(api_url):
    # https://github.com/PrefectHQ/prefect/blob/b063f84/src/prefect/client/orion.py#L51-L57
    async with OrionClient(
        api=api_url,
        api_key=PREFECT_API_KEY.value()
    ) as client:
        blocks = await client.read_block_documents()
    return blocks


# https://github.com/PrefectHQ/prefect/blob/b063f84/src/prefect/cli/cloud.py#L467-L478
async def list_workspaces():
    async with get_cloud_client() as client:
        workspaces = await client.read_workspaces()
    return workspaces


async def list_all_blocks():
    workspaces = await list_workspaces()
    for workspace in workspaces:

        blocks = await list_blocks(workspace.api_url())

        print(f"\nBlocks in workspace: '{workspace.workspace_name}':")
        for block in blocks:
            print(f"{block.block_type.slug}/{block.name}")


if __name__ == "__main__":
    anyio.run(list_all_blocks)
1 Like