Prefect

https://github.com/prefecthq/prefect
Admin
Prefect is a Python-based workflow orchestration framework for building resilient and dynamic data...
Tokens:1,725,248
Snippets:14,916
Trust Score:8.2
License:Apache-2.0
Update:3 weeks ago
Show doc for...
Tokens:
Raw
### Example repository structure

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/how-to-guides/deployments/customize-job-variables.mdx

Shows the directory layout for the example project used in this guide.

```text
» tree
.
|── README.md
|── requirements.txt
|── demo_project
|   |── daily_flow.py

```

--------------------------------

### Install dependencies, start Docker, and run client

Source: https://github.com/prefecthq/prefect/blob/main/scripts/proxy-test/README.md

Run these commands in order to set up and execute the proxy test. First install Python dependencies from requirements.txt, then start the Docker containers with docker compose, and finally run the client script.

```bash
$ uv pip install -r requirements.txt
$ docker compose up --build
$ python client.py
```

--------------------------------

### Run make setup

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/how-to-guides/deployment_infra/modal.mdx

Installs uv and sets up the Python environment. Run this command after creating the Makefile.

```bash
make setup
```

--------------------------------

### Start ipython with uvx and prefect

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/get-started/install.mdx

Start an `ipython` shell with Python 3.12 and Prefect installed using `uvx`.

```bash
uvx --python 3.12 --with prefect ipython
```

--------------------------------

### Makefile for environment setup

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/how-to-guides/deployment_infra/modal.mdx

Defines targets for installing uv, creating a virtual environment, and syncing dependencies. Run with `make setup`.

```makefile
export PYTHONPATH = .venv

.PHONY: uv
uv:
  pip install --upgrade 'uv>=0.5.6,<0.6'
  uv venv

setup:
  @if [ ! -d ".venv" ] || ! command -v uv > /dev/null; then \
    echo "UV not installed or .venv does not exist, running uv"; \
    make uv; \
  fi
  @if [ ! -f "uv.lock" ]; then \
    echo "Can't find lockfile. Locking"; \
    uv lock; \
  fi
  uv sync --all-extras --all-groups
  uv pip install --no-deps -e .
```

--------------------------------

### Expected output from Helm install

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/advanced/server-helm.mdx

Example output from a successful Helm install, showing the release name, deployment status, and notes.

```text
Release "prefect-worker" has been installed. Happy Helming!
NAME: prefect-worker
LAST DEPLOYED: Tue Mar  4 11:26:21 2025
NAMESPACE: prefect
STATUS: deployed
REVISION: 1
TEST SUITE: None
NOTES:
```

--------------------------------

### start()

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/api-ref/python/prefect-cli-server.mdx

Starts a Prefect server instance.

```APIDOC
## start()

### Description
Start a Prefect server instance.

### Method
CLI command

### Command
`prefect server start`

### Parameters
None

### Response
No return value.
```

--------------------------------

### Greeter flow startup output

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/advanced/interactive.mdx

Example output when starting the greeter flow with Prefect Cloud. Shows the flow is being served and provides the command to trigger a run.

```bash
______________________________________________________________________
| Your flow 'greeter' is being served and polling for scheduled runs |
|                                                                    |
| To trigger a run for this flow, use the following command:         |
|                                                                    |
|         $ prefect deployment run 'greeter/send-receive'            |
|                                                                    |
| You can also run your flow via the Prefect UI:                     |
| https://app.prefect.cloud/account/...(a URL for your account)      |
|                                                                    |
______________________________________________________________________
```

--------------------------------

### Install dependencies with uv or pip

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/examples/ai-data-analyst-with-pydantic-ai.mdx

Install the required dependencies for the AI data analyst example. Use either uv or pip.

```bash
uv add pydantic-ai[prefect] pandas
# or with pip:
pip install "pydantic-ai[prefect]" pandas
```

--------------------------------

### Start development server with hot-reloading

Source: https://github.com/prefecthq/prefect/blob/main/docs/contribute/dev-contribute.mdx

Starts all Prefect services with hot-reloading on code changes. Requires installation of UI dependencies.

```bash
prefect dev start
```

--------------------------------

### Expected output from helm install

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/advanced/server-helm.mdx

Example output from a successful 'helm install prefect-server' command. It shows the deployment status and provides the command to port-forward the UI to localhost:4200.

```text
NAME: prefect-server
LAST DEPLOYED: Tue Mar  4 09:08:07 2025
NAMESPACE: prefect
STATUS: deployed
REVISION: 1
TEST SUITE: None
NOTES:
Run the following command to port-forward the UI to your localhost:
$ kubectl --namespace prefect port-forward svc/prefect-server 4200:4200

Visit http://localhost:4200 to use Prefect!
```

--------------------------------

### Sender process startup output

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/advanced/interactive.mdx

Example output when starting the sender process. Shows the flow run creation and the prompt for user input.

```bash
11:38:41.800 | INFO    | prefect.engine - Created flow run 'gregarious-owl' for flow 'sender'
11:38:41.802 | INFO    | Flow run 'gregarious-owl' - View at https://app.prefect.cloud/account/...
What is your name?
```

--------------------------------

### Install prefect-gcp with all extras

Source: https://github.com/prefecthq/prefect/blob/main/docs/integrations/prefect-gcp/index.mdx

Install prefect-gcp with all additional capabilities. Run this after the base install command.

```bash
pip install -U "prefect-gcp[all_extras]"
```

```bash
uv pip install -U "prefect-gcp[all_extras]"
```

--------------------------------

### Run the ETL script with Python

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/examples/run-api-sourced-etl.mdx

Executes the example script from the command line. Run this from the project root to start the flow.

```bash
python 01_getting_started/03_run_api_sourced_etl.py
```

--------------------------------

### Create work pool and deploy

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/examples/per-worker-task-concurrency.mdx

Creates a process work pool named 'ml-pool' and deploys all flows. This is step 2 of the example setup.

```bash
prefect work-pool create ml-pool --type process
prefect deploy --all

```

--------------------------------

### Run the example deployment

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/get-started/github-quickstart.mdx

Runs the example deployment named github_quickstart in the hello_world flow.

```bash
uvx prefect-cloud run hello_world/github_quickstart
```

--------------------------------

### Get logs from a container with ID starting with c157

Source: https://github.com/prefecthq/prefect/blob/main/docs/integrations/prefect-docker/api-ref/prefect_docker-containers.mdx

Example flow that retrieves logs from a container with an ID that starts with "c157".

```python
from prefect import flow
from prefect_docker.containers import get_docker_container_logs

@flow
def get_docker_container_logs_flow():
    logs = get_docker_container_logs(container_id="c157")
    return logs

get_docker_container_logs_flow()
```

--------------------------------

### Clone and run the quickstart project

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/get-started/quickstart.mdx

Clone the quickstart repository and run the initial script to see the flow execute locally.

```bash
git clone https://github.com/PrefectHQ/quickstart && cd quickstart
python 01_getting_started.py
```

--------------------------------

### Start a container with ID starting with c157

Source: https://github.com/prefecthq/prefect/blob/main/docs/integrations/prefect-docker/api-ref/prefect_docker-containers.mdx

Example flow that starts a container with an ID that starts with "c157".

```python
from prefect import flow
from prefect_docker.containers import start_docker_container

@flow
def start_docker_container_flow():
    container = start_docker_container(container_id="c157")
    return container

start_docker_container_flow()
```

--------------------------------

### alist_blobs example

Source: https://github.com/prefecthq/prefect/blob/main/docs/integrations/prefect-gcp/api-ref/prefect_gcp-cloud_storage.mdx

Example usage of alist_blobs to get all blobs from a folder named "prefect".

```python
from prefect_gcp.cloud_storage import GcsBucket

gcs_bucket = GcsBucket.load("my-bucket")
await gcs_bucket.alist_blobs("prefect")
```

--------------------------------

### Clone and run the quickstart repository

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/get-started/quickstart.mdx

Run these commands in your terminal to clone the quickstart repository and execute the Python script. If successful, you'll see a link to the execution graph in the Prefect UI followed by state changes and logs.

```bash
git clone https://github.com/PrefectHQ/quickstart && cd quickstart
uv run 01_getting_started.py
```

--------------------------------

### Get bucket with aget_bucket

Source: https://github.com/prefecthq/prefect/blob/main/docs/integrations/prefect-gcp/api-ref/prefect_gcp-cloud_storage.mdx

Example of getting the bucket object asynchronously. Loads a GcsBucket and calls `aget_bucket`.

```python
from prefect_gcp.cloud_storage import GcsBucket

gcs_bucket = GcsBucket.load("my-bucket")
await gcs_bucket.aget_bucket()
```

--------------------------------

### Create a deployment with `serve`

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/how-to-guides/deployments/create-deployments.mdx

Simplest way to start with deployments. The `serve` method creates a deployment from your flow and immediately begins listening for scheduled runs. Providing `cron="* * * * *"` schedules the flow to run every minute.

```python
from prefect import flow


@flow
def my_flow():
    print("Hello, Prefect!")


if __name__ == "__main__":
    my_flow.serve(name="my-first-deployment", cron="* * * * *")
```

--------------------------------

### Start Prefect server

Source: https://github.com/prefecthq/prefect/blob/main/README.md

Fire up a Prefect server and open the UI at http://localhost:4200 to see what happened.

```bash
prefect server start
```

--------------------------------

### alist_folders example (with folder)

Source: https://github.com/prefecthq/prefect/blob/main/docs/integrations/prefect-gcp/api-ref/prefect_gcp-cloud_storage.mdx

Example usage of alist_folders to get all folders from a folder called "years".

```python
from prefect_gcp.cloud_storage import GcsBucket

gcs_bucket = GcsBucket.load("my-bucket")
await gcs_bucket.alist_folders("years")
```

--------------------------------

### Deploy example script from PrefectHQ/prefect-cloud

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/get-started/github-quickstart.mdx

Example deployment command using a sample script from the PrefectHQ/prefect-cloud repository. Use this if you don't have a script to deploy.

```bash
uvx prefect-cloud deploy examples/hello.py:hello_world \
    --from PrefectHQ/prefect-cloud \
    --name github_quickstart
```

--------------------------------

### Configure settings for a project with prefect.toml file

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/how-to-guides/configuration/manage-settings.mdx

Create a `prefect.toml` file in the project directory with settings to configure them for the project. Example sets api.url to http://127.0.0.1:4200/api.

```bash
api.url = "http://127.0.0.1:4200/api"
```

--------------------------------

### alist_folders example (no folder)

Source: https://github.com/prefecthq/prefect/blob/main/docs/integrations/prefect-gcp/api-ref/prefect_gcp-cloud_storage.mdx

Example usage of alist_folders to get all folders from a bucket named "my-bucket".

```python
from prefect_gcp.cloud_storage import GcsBucket

gcs_bucket = GcsBucket.load("my-bucket")
await gcs_bucket.alist_folders()
```

--------------------------------

### Prepare two deployments and serve them

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/api-ref/python/prefect-flows.mdx

Example showing how to create two deployments using `to_deployment` and serve them together with `serve`. This is the example provided for the `ato_deployment` method.

```python
from prefect import flow, serve

@flow
def my_flow(name):
    print(f"hello {name}")

@flow
def my_other_flow(name):
    print(f"goodbye {name}")

if __name__ == "__main__":
    hello_deploy = my_flow.to_deployment("hello", tags=["dev"])
    bye_deploy = my_other_flow.to_deployment("goodbye", tags=["dev"])
    serve(hello_deploy, bye_deploy)
```

--------------------------------

### Install dependencies with npm ci

Source: https://github.com/prefecthq/prefect/blob/main/ui/README.md

Installs project dependencies exactly as specified in the package-lock.json file. Run this before starting development.

```bash
npm ci
```

--------------------------------

### setup

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/api-ref/python/prefect-workers-base.mdx

Prepares the worker to run.

```APIDOC
## setup

### Description
Prepares the worker to run.

### Method
setup

### Signature
```python
setup(self) -> None
```
```

--------------------------------

### CLI output for prefect work-queue create

Source: https://github.com/prefecthq/prefect/blob/main/docs/contribute/styles-practices.mdx

Example CLI output for `prefect work-queue create testing`. Shows the created work queue properties, a next-step command to start an agent, and a command to inspect the queue. The output includes a blank line before the first line and after the last line.

```js
$ prefect work-queue create testing

Created work queue with properties:
    name - 'abcde'
    uuid - 940f9828-c820-4148-9526-ea8107082bda
    tags - None
    deployment_ids - None

Start an agent to pick up flows from the created work queue:
    prefect agent start -q 'abcde'

Inspect the created work queue:
    prefect work-queue inspect 'abcde'


```

--------------------------------

### Set up Coiled infrastructure interactively

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/how-to-guides/deployment_infra/coiled.mdx

Run the interactive setup CLI to connect Coiled to your cloud account.

```bash
coiled setup
```

--------------------------------

### setup method signature

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/api-ref/python/prefect-workers-base.mdx

Prepares the worker to run.

```python
setup(self) -> None
```

--------------------------------

### Install prefect-azure

Source: https://github.com/prefecthq/prefect/blob/main/docs/integrations/prefect-azure/index.mdx

Installs a version of prefect-azure compatible with your installed prefect version. If prefect is not installed, it installs the newest version as well.

```bash
pip install "prefect[azure]"
```

--------------------------------

### Run custom setup in each worker subprocess

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/api-ref/python/prefect-task_runners.mdx

Example of running custom setup in each worker subprocess using the initializer and initargs parameters of ProcessPoolTaskRunner.

```python
import sys
from prefect.task_runners import ProcessPoolTaskRunner

def add_to_path(*paths):
    sys.path[:0] = paths

runner = ProcessPoolTaskRunner(
    initializer=add_to_path,
    initargs=("/opt/my_app/libs",),
)
```

--------------------------------

### Install prefect-dask with uv

Source: https://github.com/prefecthq/prefect/blob/main/docs/integrations/prefect-dask/index.mdx

Alternative installation command using uv instead of pip. Installs a compatible version of prefect-dask and prefect if needed.

```bash
uv pip install "prefect[dask]"
```

--------------------------------

### Load flow from repository and run

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/how-to-guides/deployment_infra/run-flows-in-local-processes.mdx

Loads the 'hello' flow from the PrefectHQ/examples repository and runs it. The second block shows the expected console output.

```python
from prefect import flow


my_flow = flow.from_source(
    source="https://github.com/PrefectHQ/examples.git",
    entrypoint="flows/hello_world.py:hello"
)


if __name__ == "__main__":
    my_flow()
```

```bash
16:40:33.818 | INFO    | prefect.engine - Created flow run 'muscular-perch' for flow 'hello'
16:40:34.048 | INFO    | Flow run 'muscular-perch' - Hello world!
16:40:34.706 | INFO    | Flow run 'muscular-perch' - Finished in state Completed()
```

--------------------------------

### trigger example: sleep and echo

Source: https://github.com/prefecthq/prefect/blob/main/docs/integrations/prefect-shell/api-ref/prefect_shell-commands.mdx

Example usage of the sync trigger method. It starts the command, waits for completion, and fetches the result. Demonstrates the sync context manager pattern.

```python
from prefect_shell import ShellOperation

with ShellOperation(
    commands=["sleep 5", "echo 'Hello, world!'"]
) as shell_operation:
    shell_process = shell_operation.trigger()
    shell_process.wait_for_completion()
    shell_output = shell_process.fetch_result()
```

--------------------------------

### atrigger example: sleep and echo

Source: https://github.com/prefecthq/prefect/blob/main/docs/integrations/prefect-shell/api-ref/prefect_shell-commands.mdx

Example usage of the async atrigger method. It starts the command, waits for completion, and fetches the result. Demonstrates the async context manager pattern.

```python
from prefect_shell import ShellOperation

async with ShellOperation(
    commands=["sleep 5", "echo 'Hello, world!'"]
) as shell_operation:
    shell_process = await shell_operation.atrigger()
    await shell_process.await_for_completion()
    shell_output = await shell_process.afetch_result()
```

--------------------------------

### Get result from a flow state

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/api-ref/python/prefect-client-schemas-objects.mdx

Example of retrieving the result from a successful flow state.

```python
@flow
def my_flow():
    return "hello"
my_flow(return_state=True).result()
# hello
```

--------------------------------

### Start the serve process

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/how-to-guides/automations/chaining-deployments-with-events.mdx

Starts the serve process to listen for scheduled deployment runs. Run this command first to enable the event-driven chaining.

```bash
python event_driven_deployments.py
```

--------------------------------

### Clone the atproto-dashboard repository

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/examples/atproto-dashboard-with-prefect-assets.mdx

Clone the example repository and follow the README for setup and configuration.

```bash
git clone https://github.com/zzstoatzz/atproto-dashboard
cd atproto-dashboard
# Follow README for setup and configuration
```

--------------------------------

### Install prefect-azure with all extras

Source: https://github.com/prefecthq/prefect/blob/main/docs/integrations/prefect-azure/index.mdx

Installs prefect-azure with all additional capabilities, such as Blob Storage, Cosmos DB, and ML Datastore.

```bash
pip install "prefect-azure[all_extras]"
```

--------------------------------

### Install development dependencies with uv

Source: https://github.com/prefecthq/prefect/blob/main/docs/contribute/dev-contribute.mdx

Use `uv sync` to set up a virtual environment and install the development version of Prefect with all dependencies.

```bash
uv sync
```

--------------------------------

### Start greeter flow runner

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/advanced/interactive.mdx

Command to start the greeter flow runner in one terminal. Requires a Python environment with Prefect installed and pointed at Prefect Cloud or a self-hosted server.

```bash
python my_file_name greeter
```

--------------------------------

### Windows Scripts folder path example

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/get-started/install.mdx

Example path to the Python local packages `Scripts` folder on Windows. You may need to add this folder to your `Path` environment variable after installation.

```bash
C:\Users\MyUserNameHere\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.11_qbz5n2kfra8p0\LocalCache\local-packages\Python311\Scripts
```

--------------------------------

### Example output for resume --all

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/how-to-guides/deployments/manage-schedules.mdx

Example output from resuming all schedules, showing each affected deployment and the total count.

```text
Resumed schedule for deployment my-flow/daily-sync
Resumed schedule for deployment data-pipeline/hourly-update
Resumed schedule for deployment ml-model/weekly-training
Resumed 3 deployment schedule(s).
```

--------------------------------

### setup_logging

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/api-ref/python/prefect-logging-configuration.mdx

Sets up logging and returns the config used.

```APIDOC
## `setup_logging`

### Description
Sets up logging. Returns the config used.

### Signature
```python
setup_logging(incremental: bool | None = None) -> dict[str, Any]
```

### Parameters
- **incremental** (bool | None) - Optional - If True, applies incremental configuration changes.

### Returns
- **dict[str, Any]** - The logging configuration dictionary.
```

--------------------------------

### Install and configure Prefect server as a Windows service with NSSM

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/how-to-guides/self-hosted/server-windows.mdx

Use this PowerShell script to install Prefect server as a Windows service using NSSM. Run as Administrator. It sets the application path, parameters, working directory, and optional environment variables, then starts the service. This ensures the server starts automatically on boot and provides service management.

```powershell
# Run as Administrator
nssm install PrefectServer
nssm set PrefectServer Application "C:\Path\To\Python\python.exe"
nssm set PrefectServer AppParameters "-m prefect server start"
nssm set PrefectServer AppDirectory "C:\Path\To\Your\Project"

# Set environment variables if needed
nssm set PrefectServer AppEnvironmentExtra "PREFECT_API_URL=http://localhost:4200/api"

nssm start PrefectServer
```

--------------------------------

### setup_run_context method

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/api-ref/python/prefect-flow_engine.mdx

Sets up the run context, optionally using a provided PrefectClient.

```python
setup_run_context(self, client: Optional[PrefectClient] = None)
```

--------------------------------

### Get result from a flow state in an async context

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/api-ref/python/prefect-client-schemas-objects.mdx

Example of retrieving the result from an async flow state.

```python
@flow
async def my_flow():
    return "hello"
state = await my_flow(return_state=True)
await state.result()
# hello
```

--------------------------------

### Example output of prefect version

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/get-started/install.mdx

Example output from running `prefect version` or `uv run prefect version`. The exact values depend on your environment.

```bash
Version:              3.4.24
API version:          0.8.4
Python version:       3.12.8
Git commit:           2428894e
Built:                Mon, Oct 13, 2025 07:16 PM
OS/Arch:              darwin/arm64
Profile:              local
Server type:          server
Pydantic version:     2.11.7
Server:
  Database:           sqlite
  SQLite version:     3.47.1
```

--------------------------------

### Start Prefect server with basic authentication

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/release-notes/oss/version-3-1.mdx

Starts a protected Prefect server using the PREFECT_SERVER_API_AUTH_STRING environment variable.

```bash
PREFECT_SERVER_API_AUTH_STRING="admin:admin" prefect server start
```

--------------------------------

### get_cloud_storage_client example from service account file

Source: https://github.com/prefecthq/prefect/blob/main/docs/integrations/prefect-gcp/api-ref/prefect_gcp-credentials.mdx

Example flow that gets a GCP Cloud Storage client using a service account file path. The client is created by passing 'service_account_file' to GcpCredentials and calling get_cloud_storage_client().

```python
from prefect import flow
from prefect_gcp.credentials import GcpCredentials

@flow()
def example_get_client_flow():
    service_account_file = "~/.secrets/prefect-service-account.json"
    client = GcpCredentials(
        service_account_file=service_account_file
    ).get_cloud_storage_client()
example_get_client_flow()

```

--------------------------------

### Get an artifact of a dbt Cloud job run and write it to a file

Source: https://github.com/prefecthq/prefect/blob/main/docs/integrations/prefect-dbt/api-ref/prefect_dbt-cloud-runs.mdx

Example flow that retrieves the `manifest.json` artifact and writes it to a local file using `json.dump`. Note that this example imports `get_dbt_cloud_run_artifact` from `prefect_dbt.cloud.jobs` (not `prefect_dbt.cloud.runs`).

```python
import json

from prefect import flow

from prefect_dbt.cloud import DbtCloudCredentials
from prefect_dbt.cloud.jobs import get_dbt_cloud_run_artifact

@flow
def get_artifact_flow():
    credentials = DbtCloudCredentials(api_key="my_api_key", account_id=123456789)

    get_run_artifact_result = get_dbt_cloud_run_artifact(
        dbt_cloud_credentials=credentials,
        run_id=42,
        path="manifest.json"
    )

    with open("manifest.json", "w") as file:
        json.dump(get_run_artifact_result, file)

get_artifact_flow()
```

--------------------------------

### Complete send_input and receive_input example

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/advanced/interactive.mdx

Complete example of two Prefect flows using send_input and receive_input. The greeter flow receives string inputs with metadata, responds with a greeting, and tracks seen keys in a Variable. The sender flow triggers the greeter deployment, sends names, and prints the greeting responses.

```python
import asyncio
import sys
from prefect import flow, get_client
from prefect.variables import Variable
from prefect.context import get_run_context
from prefect.deployments import run_deployment
from prefect.input.run_input import receive_input, send_input


EXIT_SIGNAL = "__EXIT__"


@flow
async def greeter():
    run_context = get_run_context()
    assert run_context.flow_run, "Could not see my flow run ID"

    variable_name = f"{run_context.flow_run.id}-seen-ids"

    try:
        seen_keys = await Variable.get(variable_name)
    except (ValueError, TypeError):
        seen_keys = []

    async for name_input in receive_input(
        str,
        with_metadata=True,
        poll_interval=0.1,
        timeout=None
    ):
        if name_input.value == EXIT_SIGNAL:
            print("Goodbye!")
            return
        await name_input.respond(f"Hello, {name_input.value}!")

        seen_keys.append(name_input.metadata.key)
        await Variable.set(
            variable_name,
            seen_keys,
            overwrite=True
        )


@flow
async def sender():
    greeter_flow_run = await run_deployment(
        "greeter/send-receive", timeout=0, as_subflow=False
    )
    receiver = receive_input(str, timeout=None, poll_interval=0.1)
    client = get_client()

    while True:
        flow_run = await client.read_flow_run(greeter_flow_run.id)

        if not flow_run.state or not flow_run.state.is_running():
            continue

        name = input("What is your name? ")
        if not name:
            continue

        if name == "q" or name == "quit":
            await send_input(
                EXIT_SIGNAL,
                flow_run_id=greeter_flow_run.id
            )
            print("Goodbye!")
            break

        await send_input(name, flow_run_id=greeter_flow_run.id)
        greeting = await receiver.next()
        print(greeting)


if __name__ == "__main__":
    if sys.argv[1] == "greeter":
        asyncio.run(greeter.serve(name="send-receive"))
    elif sys.argv[1] == "sender":
        asyncio.run(sender())
```

--------------------------------

### get_default_base_job_template - Get default base job template

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/api-ref/python/prefect-cli-work_pool.mdx

Gets the default base job template for a given work pool type. The example uses the --type flag to specify the work pool type.

```python
get_default_base_job_template()
```

```bash
$ prefect work-pool get-default-base-job-template --type kubernetes
```

--------------------------------

### Serve a deployment with aserve

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/api-ref/python/prefect-flows.mdx

Example of preparing a deployment and an asynchronous initialization function, then serving the deployment with aserve. The init function sets a concurrency limit, and the deployment runs once a day.

```python
import asyncio
import datetime

from prefect import flow, aserve, get_client


async def init():
    await set_concurrency_limit()


async def set_concurrency_limit():
    async with get_client() as client:
        await client.create_concurrency_limit(tag='dev', concurrency_limit=3)


@flow
async def my_flow(name):
    print(f"hello {name}")


async def main():
    # Initialization function
    await init()

    # Run once a day
    hello_deploy = await my_flow.to_deployment(
        "hello", tags=["dev"], interval=datetime.timedelta(days=1)
    )

    await aserve(hello_deploy)


if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Get result from a failed state (raises)

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/api-ref/python/prefect-client-schemas-objects.mdx

Example of retrieving the result from a failed state, which raises the underlying exception.

```python
@flow
def my_flow():
    raise ValueError("oh no!")
state = my_flow(return_state=True)  # Error is wrapped in FAILED state
state.result()  # Raises `ValueError`
```

--------------------------------

### Example prefect.yaml file

Source: https://github.com/prefecthq/prefect/blob/main/docs/integrations/prefect-gcp/gcp-worker-guide.mdx

Copy this prefect.yaml file to the root of your project. It includes build, push, pull, and deployments sections for a CI/CD process. Replace placeholders like <WORKING-DIRECTORY> and <PATH-TO-ARTIFACT-REGISTRY>.

```yaml
# Welcome to your prefect.yaml file! You can you this file for storing and managing
# configuration for deploying your flows. We recommend committing this file to source
# control along with your flow code.

# Generic metadata about this project
name: <WORKING-DIRECTORY>
prefect-version: 3.0.0

# build section allows you to manage and build docker image
build:
- prefect_docker.deployments.steps.build_docker_image:
    id: build_image
    requires: prefect-docker>=0.3.1
    image_name: <PATH-TO-ARTIFACT-REGISTRY>/gcp-weather-image
    tag: latest
    dockerfile: auto
    platform: linux/amd64

# push section allows you to manage if and how this project is uploaded to remote locations
push:
- prefect_docker.deployments.steps.push_docker_image:
    requires: prefect-docker>=0.3.1
    image_name: '{{ build_image.image_name }}'
    tag: '{{ build_image.tag }}'

# pull section allows you to provide instructions for cloning this project in remote locations
pull:
- prefect.deployments.steps.set_working_directory:
    directory: /opt/prefect/<WORKING-DIRECTORY>

# the deployments section allows you to provide configuration for deploying flows
deployments:
- name: gcp-weather-deploy
  version: null
  tags: []
  description: null
  schedule: {}
  flow_name: null
  entrypoint: flows/weather_flow.py:fetch_weather
  parameters:
    lat: 14.5994
    lon: 28.6731
  work_pool:
    name: my-cloud-run-pool
    work_queue_name: default
    job_variables:
      image: '{{ build_image.image }}'
```

--------------------------------

### get_secret_manager_client example with service account dictionary

Source: https://github.com/prefecthq/prefect/blob/main/docs/integrations/prefect-gcp/api-ref/prefect_gcp-credentials.mdx

Gets a GCP Secret Manager client from a service account dictionary.

```python
from prefect import flow
from prefect_gcp.credentials import GcpCredentials

@flow()
def example_get_client_flow():
    service_account_info = {
        "type": "service_account",
        "project_id": "project_id",
        "private_key_id": "private_key_id",
        "private_key": "private_key",
        "client_email": "client_email",
        "client_id": "client_id",
        "auth_uri": "auth_uri",
        "token_uri": "token_uri",
        "auth_provider_x509_cert_url": "auth_provider_x509_cert_url",
        "client_x509_cert_url": "client_x509_cert_url"
    }
    client = GcpCredentials(
        service_account_info=service_account_info
    ).get_secret_manager_client()
example_get_client_flow()
```

--------------------------------

### Build static UI with prefect dev build-ui --include-v2

Source: https://github.com/prefecthq/prefect/blob/main/docs/contribute/dev-contribute.mdx

Build a static UI that will be served when running `prefect server start`. Use this when you need a static build of the UI.

```bash
prefect dev build-ui --include-v2
```

--------------------------------

### get_job_service_client example with service account dictionary

Source: https://github.com/prefecthq/prefect/blob/main/docs/integrations/prefect-gcp/api-ref/prefect_gcp-credentials.mdx

Gets a GCP Job Service client from a service account dictionary.

```python
from prefect import flow
from prefect_gcp.credentials import GcpCredentials

@flow()
def example_get_client_flow():
    service_account_info = {
        "type": "service_account",
        "project_id": "project_id",
        "private_key_id": "private_key_id",
        "private_key": "private_key",
        "client_email": "client_email",
        "client_id": "client_id",
        "auth_uri": "auth_uri",
        "token_uri": "token_uri",
        "auth_provider_x509_cert_url": "auth_provider_x509_cert_url",
        "client_x509_cert_url": "client_x509_cert_url"
    }
    client = GcpCredentials(
        service_account_info=service_account_info
    ).get_job_service_client()

example_get_client_flow()
```

--------------------------------

### start_services()

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/api-ref/python/prefect-cli-server.mdx

Starts all enabled Prefect services in one process.

```APIDOC
## start_services()

### Description
Start all enabled Prefect services in one process.

### Method
CLI command

### Command
`prefect server start-services`

### Parameters
None

### Response
No return value.
```

--------------------------------

### Airflow DAG example (simplified ETL)

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/how-to-guides/migrate/airflow.mdx

Airflow DAG example for a simplified ETL pipeline. It defines three PythonOperator tasks and sets sequential dependencies using the >> operator. This is the starting point for conversion to Prefect flows.

```python
# Airflow DAG example (simplified ETL)
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

# Airflow task functions (to be used by PythonOperator)
def extract_fn():
    # ... (extract data, e.g., query an API or database)
    return data

def transform_fn(data):
    # ... (transform the data)
    return processed_data

def load_fn(processed_data):
    # ... (load data to target, e.g., save to DB or file)

with DAG("etl_pipeline", start_date=datetime(2023,1,1), schedule_interval="@daily", catchup=False) as dag:
    extract = PythonOperator(task_id='extract_data', python_callable=extract_fn)
    transform = PythonOperator(task_id='transform_data', python_callable=transform_fn)
    load = PythonOperator(task_id='load_data', python_callable=load_fn)

    # Set task dependencies
    extract >> transform >> load
```

--------------------------------

### prefect server start

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/api-ref/cli/server.mdx

Starts a Prefect server instance. This command launches the server with configurable options for host, port, logging, scheduler, analytics, and more.

```APIDOC
## `prefect server start`

### Description
Start a Prefect server instance.

### Command
```
prefect server start [OPTIONS]
```

### Options
- **--host** (string) - Server host address. [from PREFECT_SERVER_API_HOST]
- **--port** (integer) - Server port. [from PREFECT_SERVER_API_PORT]
- **--keep-alive-timeout** (integer) - Keep-alive timeout seconds. [from PREFECT_SERVER_API_KEEPALIVE_TIMEOUT]
- **--log-level** (string) - Server logging level. [from PREFECT_SERVER_LOGGING_LEVEL]
- **--scheduler** (boolean) - Enable scheduler. [from PREFECT_SERVER_SERVICES_SCHEDULER_ENABLED]
- **--analytics-on** (boolean) - Toggle analytics. [from PREFECT_SERVER_ANALYTICS_ENABLED]
- **--late-runs** (boolean) - Enable late runs. [from PREFECT_SERVER_SERVICES_LATE_RUNS_ENABLED]
- **--ui** (boolean) - Enable the UI. [from PREFECT_SERVER_UI_ENABLED]
- **--no-services** (boolean) - Only run the webserver API and UI.
- **--background** (boolean) - Run in the background.
- **--workers** (integer) - Number of worker processes.
```

--------------------------------

### Install prefect into a uv virtual environment

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/get-started/install.mdx

Create a `uv` virtual environment with Python 3.12, activate it, and install Prefect.

```bash
uv venv --python 3.12
source .venv/bin/activate
uv add prefect
```

--------------------------------

### get_job_service_async_client example from service account file

Source: https://github.com/prefecthq/prefect/blob/main/docs/integrations/prefect-gcp/api-ref/prefect_gcp-credentials.mdx

Example flow that gets a GCP Job Service async client using a service account file path. The client is created by passing 'service_account_file' to GcpCredentials and calling get_job_service_async_client().

```python
from prefect import flow
from prefect_gcp.credentials import GcpCredentials

@flow()
def example_get_client_flow():
    service_account_file = "~/.secrets/prefect-service-account.json"
    client = GcpCredentials(
        service_account_file=service_account_file
    ).get_job_service_async_client()

example_get_client_flow()

```

--------------------------------

### get_cloud_storage_client example from service account info dictionary

Source: https://github.com/prefecthq/prefect/blob/main/docs/integrations/prefect-gcp/api-ref/prefect_gcp-credentials.mdx

Example flow that gets a GCP Cloud Storage client using a service account info dictionary. The client is created by passing 'service_account_info' to GcpCredentials and calling get_cloud_storage_client().

```python
from prefect import flow
from prefect_gcp.credentials import GcpCredentials

@flow()
def example_get_client_flow():
    service_account_info = {
        "type": "service_account",
        "project_id": "project_id",
        "private_key_id": "private_key_id",
        "private_key": "private_key",
        "client_email": "client_email",
        "client_id": "client_id",
        "auth_uri": "auth_uri",
        "token_uri": "token_uri",
        "auth_provider_x509_cert_url": "auth_provider_x509_cert_url",
        "client_x509_cert_url": "client_x509_cert_url"
    }
    client = GcpCredentials(
        service_account_info=service_account_info
    ).get_cloud_storage_client()
example_get_client_flow()

```

--------------------------------

### PrefectClient hello example

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/api-ref/python/prefect-client-orchestration-__init__.mdx

Example from the PrefectClient class docstring showing how to say hello to a Prefect REST API and print the response.

```python
async with get_client() as client:
    response = await client.hello()

print(response.json())
👋
```

--------------------------------

### get_job_service_async_client example from service account info dictionary

Source: https://github.com/prefecthq/prefect/blob/main/docs/integrations/prefect-gcp/api-ref/prefect_gcp-credentials.mdx

Example flow that gets a GCP Job Service async client using a service account info dictionary. The client is created by passing 'service_account_info' to GcpCredentials and calling get_job_service_async_client().

```python
from prefect import flow
from prefect_gcp.credentials import GcpCredentials

@flow()
def example_get_client_flow():
    service_account_info = {
        "type": "service_account",
        "project_id": "project_id",
        "private_key_id": "private_key_id",
        "private_key": "private_key",
        "client_email": "client_email",
        "client_id": "client_id",
        "auth_uri": "auth_uri",
        "token_uri": "token_uri",
        "auth_provider_x509_cert_url": "auth_provider_x509_cert_url",
        "client_x509_cert_url": "client_x509_cert_url"
    }
    client = GcpCredentials(
        service_account_info=service_account_info
    ).get_job_service_async_client()

example_get_client_flow()

```

--------------------------------

### setup_run_context method signature

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/api-ref/python/prefect-flow_engine.mdx

Sets up the run context, optionally using a provided synchronous Prefect client.

```python
setup_run_context(self, client: Optional[SyncPrefectClient] = None)
```

--------------------------------

### Create process work pool and start worker

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/how-to-guides/migrate/airflow.mdx

Create a process work pool and start a worker on each node to replicate Airflow's CeleryExecutor setup. Workers poll the work pool and execute assigned flow runs.

```bash
prefect work-pool create prod-work-pool --type process
```

```bash
prefect worker start -p prod-work-pool
```

--------------------------------

### Configure settings for a project with .env file

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/how-to-guides/configuration/manage-settings.mdx

Create a `.env` file in the project directory with settings to configure them for the project. Example sets PREFECT_API_URL to http://127.0.0.1:4200/api.

```bash
PREFECT_API_URL=http://127.0.0.1:4200/api
```

--------------------------------

### Example: provision infrastructure for a work pool

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/api-ref/cli/work-pools.mdx

Example commands for provisioning infrastructure for a work pool named "my-pool". Both the full and abbreviated command forms are shown.

```bash
$ prefect work-pool provision-infrastructure "my-pool"
$ prefect work-pool provision-infra "my-pool"
```

--------------------------------

### get_secret_manager_client example with service account file

Source: https://github.com/prefecthq/prefect/blob/main/docs/integrations/prefect-gcp/api-ref/prefect_gcp-credentials.mdx

Gets a GCP Secret Manager client from a service account file path.

```python
from prefect import flow
from prefect_gcp.credentials import GcpCredentials

@flow()
def example_get_client_flow():
    service_account_file = "~/.secrets/prefect-service-account.json"
    client = GcpCredentials(
        service_account_file=service_account_file
    ).get_secret_manager_client()
example_get_client_flow()
```

--------------------------------

### prefect.yaml templating example

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/how-to-guides/deployments/prefect-yaml.mdx

Example prefect.yaml file demonstrating templating options: step outputs ({{ build-image.tag }}), block references ({{ prefect.blocks.docker-registry-credentials.dev-registry }}), variables ({{ prefect.variables.some_common_tag }}), and environment variables ({{ $my_deployment_tag }}). The build step must produce image_name and tag fields for the deployment to populate dynamically.

```yaml
build:
- prefect_docker.deployments.steps.build_docker_image:
    id: build-image
    requires: prefect-docker>=0.6.0
    image_name: my-repo/my-image
    tag: my-tag
    dockerfile: auto

push:
- prefect_docker.deployments.steps.push_docker_image:
    requires: prefect-docker>=0.6.0
    image_name: my-repo/my-image
    tag: my-tag
    credentials: "{{ prefect.blocks.docker-registry-credentials.dev-registry }}"

deployments:
- # base metadata
  name: null
  version: "{{ build-image.tag }}"
  tags:
  - "{{ $my_deployment_tag }}"
  - "{{ prefect.variables.some_common_tag }}"
  description: null
  concurrency_limit: null

  # flow-specific fields
  entrypoint: null
  parameters: {}

  # infra-specific fields
  work_pool:
    name: "my-k8s-work-pool"
    work_queue_name: null
    job_variables:
      image: "{{ build-image.image }}"
      cluster_config: "{{ prefect.blocks.kubernetes-cluster-config.my-favorite-config }}"
```

--------------------------------

### get_job_service_client example with service account file

Source: https://github.com/prefecthq/prefect/blob/main/docs/integrations/prefect-gcp/api-ref/prefect_gcp-credentials.mdx

Gets a GCP Job Service client from a service account file path.

```python
from prefect import flow
from prefect_gcp.credentials import GcpCredentials

@flow()
def example_get_client_flow():
    service_account_file = "~/.secrets/prefect-service-account.json"
    client = GcpCredentials(
        service_account_file=service_account_file
    ).get_job_service_client()

example_get_client_flow()
```

--------------------------------

### Verify Prefect installation with uv

Source: https://github.com/prefecthq/prefect/blob/main/docs/contribute/dev-contribute.mdx

Run `uv run prefect --version` to confirm Prefect is installed correctly when using uv.

```bash
uv run prefect --version
```

--------------------------------

### Complete reschedule_late_flows.py example

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/advanced/api-client.mdx

Complete runnable example (`reschedule_late_flows.py`) that defines the rescheduling function and executes it for the `healthcheck-storage-test` deployment with a six-hour delay. It prints the number of rescheduled runs and asserts that all new runs are scheduled with future expected start times.

```python
from __future__ import annotations

import asyncio
from datetime import datetime, timedelta, timezone

from prefect import get_client
from prefect.client.schemas.filters import DeploymentFilter, FlowRunFilter
from prefect.client.schemas.objects import FlowRun
from prefect.client.schemas.sorting import FlowRunSort
from prefect.states import Scheduled

async def reschedule_late_flow_runs(
    deployment_name: str,
    delay: timedelta,
    most_recent_n: int,
    delete_remaining: bool = True,
    states: list[str] | None = None
) -> list[FlowRun]:
    states = states or ["Late"]

    async with get_client() as client:
        flow_runs = await client.read_flow_runs(
            flow_run_filter=FlowRunFilter(
                state=dict(name=dict(any_=states)),
                expected_start_time=dict(before_=datetime.now(timezone.utc)),
            ),
            deployment_filter=DeploymentFilter(name={'like_': deployment_name}),
            sort=FlowRunSort.START_TIME_DESC,
            limit=most_recent_n if not delete_remaining else None
        )

        if not flow_runs:
            print(f"No flow runs found in states: {states!r}")
            return []
        
        rescheduled_flow_runs: list[FlowRun] = []
        for i, run in enumerate(flow_runs):
            await client.delete_flow_run(flow_run_id=run.id)
            if i < most_recent_n:
                new_run = await client.create_flow_run_from_deployment(
                    deployment_id=run.deployment_id,
                    state=Scheduled(scheduled_time=run.expected_start_time + delay),
                )
                rescheduled_flow_runs.append(new_run)
            
        return rescheduled_flow_runs


if __name__ == "__main__":
    rescheduled_flow_runs = asyncio.run(
        reschedule_late_flow_runs(
            deployment_name="healthcheck-storage-test",
            delay=timedelta(hours=6),
            most_recent_n=3,
        )
    )
    
    print(f"Rescheduled {len(rescheduled_flow_runs)} flow runs")
    
    assert all(run.state.is_scheduled() for run in rescheduled_flow_runs)
    assert all(
        run.expected_start_time > datetime.now(timezone.utc)
        for run in rescheduled_flow_runs
    )
```

--------------------------------

### Get result from a failed state without erroring

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/api-ref/python/prefect-client-schemas-objects.mdx

Example of retrieving the result from a failed state without raising, using raise_on_failure=False.

```python
@flow
def my_flow():
    raise ValueError("oh no!")
state = my_flow(return_state=True)
result = state.result(raise_on_failure=False)
print(result)
# ValueError("oh no!")
```

--------------------------------

### Switch to a profile

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/how-to-guides/configuration/manage-settings.mdx

Run `prefect profile use <profile>` to switch to a new profile.

```bash
prefect profile use <profile>
```

--------------------------------

### Initialize Docker recipe with interactive prompts

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/how-to-guides/deployments/prefect-yaml.mdx

Initializes a Prefect project with the Docker recipe and prompts for the required image name and tag fields.

```bash
prefect init --recipe docker
>> image_name: < insert image name here >
>> tag: < insert image tag here >
```

--------------------------------

### Get BigQuery client from a dictionary

Source: https://github.com/prefecthq/prefect/blob/main/docs/integrations/prefect-gcp/api-ref/prefect_gcp-credentials.mdx

Example flow that creates a GcpCredentials block with service account info as a dictionary and retrieves a BigQuery client.

```python
from prefect import flow
from prefect_gcp.credentials import GcpCredentials

@flow()
def example_get_client_flow():
    service_account_info = {
        "type": "service_account",
        "project_id": "project_id",
        "private_key_id": "private_key_id",
        "private_key": "private_key",
        "client_email": "client_email",
        "client_id": "client_id",
        "auth_uri": "auth_uri",
        "token_uri": "token_uri",
        "auth_provider_x509_cert_url": "auth_provider_x509_cert_url",
        "client_x509_cert_url": "client_x509_cert_url"
    }
    client = GcpCredentials(
        service_account_info=service_account_info
    ).get_bigquery_client()

example_get_client_flow()
```

--------------------------------

### Get BigQuery client from a path

Source: https://github.com/prefecthq/prefect/blob/main/docs/integrations/prefect-gcp/api-ref/prefect_gcp-credentials.mdx

Example flow that creates a GcpCredentials block with a service account file path and retrieves a BigQuery client.

```python
from prefect import flow
from prefect_gcp.credentials import GcpCredentials

@flow()
def example_get_client_flow():
    service_account_file = "~/.secrets/prefect-service-account.json"
    client = GcpCredentials(
        service_account_file=service_account_file
    ).get_bigquery_client()
example_get_client_flow()
```

--------------------------------

### get_client connection example

Source: https://github.com/prefecthq/prefect/blob/main/docs/integrations/prefect-sqlalchemy/api-ref/prefect_sqlalchemy-database.mdx

Example for get_client: creates a context-managed connection by calling get_client with client_type='connection'.

```python
from prefect_sqlalchemy import SqlAlchemyConnector

sqlalchemy_connector = SqlAlchemyConnector.load("BLOCK_NAME")
with sqlalchemy_connector.get_client(client_type="connection") as conn:
    ...
```

--------------------------------

### Import Prefect flow and tags

Source: https://github.com/prefecthq/prefect/blob/main/docs/v3/examples/hello-world.mdx

Start by importing the essential `flow` decorator (and the `tags` context manager) from Prefect. This is the required setup for defining flows.

```python
from prefect import flow, tags

```

--------------------------------

### Create an ECS work pool

Source: https://github.com/prefecthq/prefect/blob/main/docs/integrations/prefect-aws/api-ref/prefect_aws-workers-ecs_worker.mdx

Creates an ECS work pool named 'my-ecs-pool' with the 'ecs' type. This is the first step to get started with the ECS worker.

```bash
$ prefect work-pool create --type ecs my-ecs-pool
```
Tokens:
Raw