Prefect

https://docs.prefect.io/v3/
Admin
Prefect is an open-source orchestration engine that transforms Python functions into...
Tokens:2,507,010
Snippets:11,443
Trust Score:10
Update:1 week ago
Show doc for...
Tokens:
Raw
### Run make setup

Source: https://docs.prefect.io/v3/how-to-guides/deployment_infra/modal

Runs the setup target defined in the Makefile to install uv, create a virtual environment, and sync dependencies.

```bash
make setup

```

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

### Complete greeter and sender flows using send_input and receive_input

Source: https://docs.prefect.io/v3/advanced/interactive

This example shows a greeter flow that receives input and responds with a greeting, and a sender flow that sends input and receives the greeting. The greeter flow uses receive_input with metadata and responds to each input, while the sender flow uses run_deployment to start the greeter flow, sends input via send_input, and reads the greeting with receive_input. The example requires a Python environment with Prefect installed and pointed at a Prefect Cloud account or self-hosted server.

```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())

```

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

### Example repo structure (tree)

Source: https://docs.prefect.io/v3/how-to-guides/deployments/customize-job-variables

Shows the repository layout used in the guide, including the demo_project directory and daily_flow.py file.

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

```

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

### Setup: Install dependencies and set API key

Source: https://docs.prefect.io/v3/examples/ai-database-cleanup-with-approval

Install prefect for human approval, or pydantic-ai with prefect extras for AI approval. Set the ANTHROPIC_API_KEY environment variable for AI features.

```bash
# For human approval only
uv add prefect

# For AI approval, add pydantic-ai
uv add 'pydantic-ai[prefect]'
export ANTHROPIC_API_KEY='your-key'

```

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

### Start the serve process

Source: https://docs.prefect.io/v3/how-to-guides/automations/passing-event-payloads-to-flows

Run the Python script to start the serve process. This command starts the deployment and outputs the deployment ID.

```bash
python event_payload_example.py
```

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

### Start an ipython shell with uvx

Source: https://docs.prefect.io/v3/get-started/install

Launches an interactive Python shell with Python 3.12 and Prefect installed, using uvx.

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

```

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

### Makefile for environment setup

Source: https://docs.prefect.io/v3/how-to-guides/deployment_infra/modal

Creates a Makefile with targets for installing uv, setting up a virtual environment, and syncing dependencies. Used to simplify environment management.

```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 .

```

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

### Start a Runner with two flows

Source: https://docs.prefect.io/v3/api-ref/python/prefect-runner-runner

Example from the start method docs: initialize a Runner, add two flows (one API-runnable, one with a cron schedule), and serve them by calling asyncio.run(runner.start()).

```python
import asyncio
from prefect import flow, Runner

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

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

    if __name__ == "__main__":
        runner = Runner(name="my-runner")

        # Will be runnable via the API
        runner.add_flow(hello_flow)

        # Run on a cron schedule
        runner.add_flow(goodbye_flow, schedule={"cron": "0 * * * *"})

        asyncio.run(runner.start())
```

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

### Clone the atproto-dashboard repository

Source: https://docs.prefect.io/v3/examples/atproto-dashboard-with-prefect-assets

Clone the example repository and follow its README for setup and configuration. This is a simplified demonstration; the full implementation is in the linked repository.

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

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

### prefect api command examples

Source: https://docs.prefect.io/v3/api-ref/python/prefect-cli-api

Examples of using the `prefect api` command to make direct requests to the Prefect API. Includes GET and POST requests, sending data with --data, custom headers with -H, verbose output with --verbose, account-level operation with --account, and API root level with --root.

```bash
# GET request
$ prefect api GET /flows/abc-123
```

```bash
# POST request with data
$ prefect api POST /flows/filter --data '{"limit": 10}'
```

```bash
# POST to filter endpoint (defaults to empty object)
$ prefect api POST /flows/filter
```

```bash
# Custom headers
$ prefect api POST /flows/filter -H "X-Custom: value" --data '{}'
```

```bash
# Verbose output
$ prefect api GET /flows --verbose
```

```bash
# Account-level operation (Cloud)
$ prefect api GET /workspaces --account
```

```bash
# API root level (Cloud only)
$ prefect api GET /me --root
```

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

### Start the systemd service

Source: https://docs.prefect.io/v3/advanced/daemonize-processes

Start the service immediately.

```bash
sudo systemctl start my-prefect-service

```

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

### Clone quickstart repo and run flow locally

Source: https://docs.prefect.io/v3/get-started/quickstart

Clones the Prefect quickstart repository and runs the 01_getting_started.py script. Expect a link to the execution graph and state changes/logs.

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

```

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

### NGINX load balancer configuration for Prefect API

Source: https://docs.prefect.io/v3/advanced/self-hosted

Example NGINX configuration that load balances across multiple Prefect API servers and defines a health check endpoint at /api/health. Use this as a starting point for your load balancer setup.

```nginx
upstream prefect_api {
    least_conn;
    server prefect-api-1:4200 max_fails=3 fail_timeout=30s;
    server prefect-api-2:4200 max_fails=3 fail_timeout=30s;
    server prefect-api-3:4200 max_fails=3 fail_timeout=30s;
}

server {
    listen 4200;

    location /api/health {
        proxy_pass http://prefect_api;
        proxy_connect_timeout 1s;
        proxy_read_timeout 1s;
    }

    location / {
        proxy_pass http://prefect_api;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
}
```

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

### Get a value from the transaction

Source: https://docs.prefect.io/v3/api-ref/python/prefect-transactions

Example showing how to set and get a value within a transaction.

```python
with transaction() as txn:
    txn.set("key", "value")
    ...
    assert txn.get("key") == "value"

```

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

### Run example_flow.py

Source: https://docs.prefect.io/v3/how-to-guides/deployment_infra/coiled

Executes the Python script to create the deployment.

```bash
python example_flow.py

```

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

### GET /variables/name/{name} request examples in multiple languages

Source: https://docs.prefect.io/v3/api-ref/rest-api/server/variables/read-variable-by-name

Example requests for the GET /variables/name/{name} endpoint in cURL, Python, JavaScript, PHP, Go, Java (Unirest), and Ruby. All use the same URL and HTTP GET method. The {name} path parameter must be replaced with the variable name.

```bash
curl --request GET \
  --url https://api.example.com/variables/name/{name}
```

```python
import requests

url = "https://api.example.com/variables/name/{name}"

response = requests.get(url)

print(response.text)
```

```javascript
const options = {method: 'GET'};

fetch('https://api.example.com/variables/name/{name}', options)
  .then(res => res.json())
  .then(res => console.log(res))
  .catch(err => console.error(err));
```

```php
<?php

$curl = curl_init();

curl_setopt_array($curl, [
  CURLOPT_URL => "https://api.example.com/variables/name/{name}",
  CURLOPT_RETURNTRANSFER => true,
  CURLOPT_ENCODING => "",
  CURLOPT_MAXREDIRS => 10,
  CURLOPT_TIMEOUT => 30,
  CURLOPT_HTTP_VERSION => CURL_HTTP_VERSION_1_1,
  CURLOPT_CUSTOMREQUEST => "GET",
]);

$response = curl_exec($curl);
$err = curl_error($curl);

curl_close($curl);

if ($err) {
  echo "cURL Error #:" . $err;
} else {
  echo $response;
}
```

```go
package main

import (
	"fmt"
	"net/http"
	"io"
)

func main() {

	url := "https://api.example.com/variables/name/{name}"

	req, _ := http.NewRequest("GET", url, nil)

	res, _ := http.DefaultClient.Do(req)

	defer res.Body.Close()
	body, _ := io.ReadAll(res.Body)

	fmt.Println(string(body))

}
```

```java
HttpResponse<String> response = Unirest.get("https://api.example.com/variables/name/{name}")
  .asString();
```

```ruby
require 'uri'
require 'net/http'

url = URI("https://api.example.com/variables/name/{name}")

http = Net::HTTP.new(url.host, url.port)
http.use_ssl = true

request = Net::HTTP::Get.new(url)

response = http.request(request)
puts response.read_body
```

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

### Updated flow with .serve() (01_getting_started.py)

Source: https://docs.prefect.io/v3/get-started/quickstart

Modifies the flow to use main.serve(name="my-first-deployment") to set up a simple deployment. Run the updated file to start the deployment.

```python
from prefect import flow, task
import random

@task
def get_customer_ids() -> list[str]:
    # Fetch customer IDs from a database or API
    return [f"customer{n}" for n in random.choices(range(100), k=10)]

@task
def process_customer(customer_id: str) -> str:
    # Process a single customer
    return f"Processed {customer_id}"

@flow
def main() -> list[str]:
    customer_ids = get_customer_ids()
    # Map the process_customer task across all customer IDs
    results = process_customer.map(customer_ids)
    return results


if __name__ == "__main__":
    main.serve(name="my-first-deployment")

```

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

### Get result from a flow state

Source: https://docs.prefect.io/v3/api-ref/python/prefect-client-schemas-objects

Example showing how to get the result from a flow state using the result() method.

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

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

### setup

Source: https://docs.prefect.io/v3/api-ref/python/prefect-workers-base

Prepares the worker to run.

```APIDOC
## setup

### Description
Prepares the worker to run.

### Method
setup

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

```

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

### Java Unirest GET saved search

Source: https://docs.prefect.io/v3/api-ref/rest-api/server/savedsearches/read-saved-search

Java Unirest example to GET a saved search. Replace {id} with the UUID.

```java
HttpResponse<String> response = Unirest.get("https://api.example.com/saved_searches/{id}")
  .asString();
```

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

### Complete custom worker implementation example

Source: https://docs.prefect.io/v3/advanced/developing-a-custom-worker

A full example of a custom worker implementation, including configuration and template variables classes, and the run method with typical operations: create and start job, mark started, watch job, and return result.

```python
import anyio
import anyio.abc
from prefect.client.schemas.objects import FlowRun
from prefect.workers.base import BaseWorker, BaseWorkerResult, BaseJobConfiguration, BaseVariables
from pydantic import Field

class MyWorkerConfiguration(BaseJobConfiguration):
    memory: str = Field(
        default="1024Mi",
        description="Memory allocation for the execution environment.",
        json_schema_extra=dict(template="{{ memory_request }}Mi")
    )
    cpu: str = Field(
        default="500m",
        description="CPU allocation for the execution environment.",
        json_schema_extra=dict(template="{{ cpu_request }}m")
    )

class MyWorkerTemplateVariables(BaseVariables):
    memory_request: int = Field(
        default=1024,
        description="Memory allocation for the execution environment."
    )
    cpu_request: int = Field(
        default=500,
        description="CPU allocation for the execution environment."
    )

class MyWorkerResult(BaseWorkerResult):
    """Result returned by the MyWorker."""

class MyWorker(BaseWorker):
    type = "my-worker"
    job_configuration = MyWorkerConfiguration
    job_configuration_variables = MyWorkerTemplateVariables
    _documentation_url = "https://example.com/docs"
    _logo_url = "https://example.com/logo"
    _description = "My worker description."

    async def run(
        self, flow_run: FlowRun, configuration: BaseJobConfiguration, task_status: anyio.abc.TaskStatus = anyio.TASK_STATUS_IGNORED,
    ) -> BaseWorkerResult:
        # Create the execution environment and start execution
        job = await self._create_and_start_job(configuration)

        # Use a unique ID to mark the run as started. This ID is later used to tear down infrastructure
        # if the flow run is cancelled.
        task_status.started(job.id)

        # Monitor the execution
        job_status = await self._watch_job(job, configuration)

        exit_code = job_status.exit_code if job_status else -1 # Get result of execution for reporting
        return MyWorkerResult(
            status_code=exit_code,
            identifier=job.id,
        )
```

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

### Clone quickstart repository and run the flow

Source: https://docs.prefect.io/v3/get-started/quickstart

Run these commands to clone the quickstart repository and execute the flow locally. If successful, you'll see a link to the execution graph in the Prefect UI.

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

```

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

### EventPersister.started_event (getter)

Source: https://docs.prefect.io/v3/api-ref/python/prefect-server-events-services-event_persister

Instance method to get the started event.

```APIDOC
## EventPersister.started_event (getter)

### Description
Returns the started event.

### Signature
```python
started_event(self) -> asyncio.Event
```

### Returns
asyncio.Event - The started event.
```

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

### Basic flow entrypoint for serve process

Source: https://docs.prefect.io/v3/advanced/daemonize-processes

Example flow file that uses .serve() to create a deployment. This file should be placed in /home or your chosen directory.

```python
from prefect import flow


@flow(log_prints=True)
def say_hi():
    print("Hello!")

if __name__=="__main__":
    say_hi.serve(name="served and daemonized deployment")

```

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

### prefect deployment inspect example output

Source: https://docs.prefect.io/v3/api-ref/cli/deployment

Example output from `prefect deployment inspect "hello-world/my-deployment"`, showing deployment details including id, name, flow_id, parameters, tags, and infrastructure configuration.

```json
{
    'id': '610df9c3-0fb4-4856-b330-67f588d20201',
    'created': '2022-08-01T18:36:25.192102+00:00',
    'updated': '2022-08-01T18:36:25.188166+00:00',
    'name': 'my-deployment',
    'description': None,
    'flow_id': 'b57b0aa2-ef3a-479e-be49-381fb0483b4e',
    'schedules': None,
    'parameters': {'name': 'Marvin'},
    'tags': ['test'],
    'parameter_openapi_schema': {
        'title': 'Parameters',
        'type': 'object',
        'properties': {
            'name': {
                'title': 'name',
                'type': 'string'
            }
        },
        'required': ['name']
    },
    'storage_document_id': '63ef008f-1e5d-4e07-a0d4-4535731adb32',
    'infrastructure_document_id': '6702c598-7094-42c8-9785-338d2ec3a028',
    'infrastructure': {
        'type': 'process',
        'env': {},
        'labels': {},
        'name': None,
        'command': ['python', '-m', 'prefect.engine'],
        'stream_output': True
    }
}
```

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

### Python requests example for paginate task runs

Source: https://docs.prefect.io/v3/api-ref/rest-api/server/task-runs/paginate-task-runs

Starts a Python requests call to the same endpoint, but the example is incomplete as it only shows the import statement.

```python
import requests
```

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

### Ruby Net::HTTP GET saved search

Source: https://docs.prefect.io/v3/api-ref/rest-api/server/savedsearches/read-saved-search

Ruby Net::HTTP example to GET a saved search. Replace {id} with the UUID.

```ruby
require 'uri'
require 'net/http'

url = URI("https://api.example.com/saved_searches/{id}")

http = Net::HTTP.new(url.host, url.port)
http.use_ssl = true

request = Net::HTTP::Get.new(url)

response = http.request(request)
puts response.read_body
```

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

### Bootstrap a new project with uv

Source: https://docs.prefect.io/v3/advanced/background-tasks

Commands to initialize a new Python library project and add the prefect and marvin dependencies using uv.

```bash
uv init --lib foo
uv add prefect marvin

```

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

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

Source: https://docs.prefect.io/v3/api-ref/python/prefect-client-schemas-objects

Example showing how to get the result from a flow state in an async context using await.

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

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

### Start a worker for multiple functions with serve()

Source: https://docs.prefect.io/v3/how-to-guides/workflows/run-background-tasks

Call `serve()` with the tasks you want to run to start a worker for multiple functions. The example includes two tasks and their `.delay()` invocations, which run once the worker is started.

```python
from prefect import task
from prefect.task_worker import serve

@task(log_prints=True)
def add(a: int, b: int):
    print(f"{a} + {b} = {a + b}")

@task(log_prints=True)
def multiply(a: int, b: int):
    print(f"{a} * {b} = {a * b}")

# Both these invocations will run once the worker is started
add.delay(1, 2)
multiply.delay(1, 4)

serve(add, multiply) # start a task worker listening for both `add` and `multiply`

```

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

### Output of starting a worker

Source: https://docs.prefect.io/v3/concepts/workers

Example output when starting a worker with a work pool named 'my-pool'. Shows the discovered worker type and the worker ID.

```text
Discovered worker type 'process' for work pool 'my-pool'.
Worker 'ProcessWorker 65716280-96f8-420b-9300-7e94417f2673' started!

```

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

### Create a deployment with `serve` (Python)

Source: https://docs.prefect.io/v3/how-to-guides/deployments/create-deployments

Creates a deployment from a flow using the `serve` method)Skip and immediately starts listening for scheduled runs. Passing `cron="* * * * *"` associates a schedule that runs 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="* * * * *")

```

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

### Run updated flow file

Source: https://docs.prefect.io/v3/get-started/quickstart

Runs the updated 01_getting_started.py file to start the deployment created with .serve().

```bash
python 01_getting_started.py

```

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

### Get result from a failed state without erroring

Source: https://docs.prefect.io/v3/api-ref/python/prefect-client-schemas-objects

Example showing how to get the result from a failed state without raising an exception by setting 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!")
```

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

### Repository structure for example deployment

Source: https://docs.prefect.io/v3/deploy/infrastructure-concepts/customize

Shows the repository structure with a demo project folder containing daily_flow.py.

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

```

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

### Start a worker with a specific work pool name

Source: https://docs.prefect.io/v3/concepts/workers

Example of starting a worker with a work pool named 'my-pool'. Prefect automatically discovers the worker type from the work pool.

```bash
prefect worker start -p "my-pool"

```

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

### AsyncFlowRunEngine.setup_run_context

Source: https://docs.prefect.io/v3/api-ref/python/prefect-flow_engine

Sets up the run context.

```APIDOC
## AsyncFlowRunEngine.setup_run_context

### Description
Sets up the run context.

### Method
setup_run_context

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

### Parameters
- **client** (Optional[PrefectClient]) - Optional - The Prefect client to use.

### Returns
None
```

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

### Serve a flow with hello_world.serve

Source: https://docs.prefect.io/v3/how-to-guides/deployment_infra/run-flows-in-local-processes

Defines a Prefect flow with log_prints=True and serves it with the serve method, creating a deployment and starting a long-running process that listens for scheduled work. The serve call includes a name, tags, default parameters, and an interval schedule.

```python
from prefect import flow


@flow(log_prints=True)
def hello_world(name: str = "world", goodbye: bool = False):
    print(f"Hello {name} from Prefect! 🤗")

    if goodbye:
        print(f"Goodbye {name}!")


if __name__ == "__main__":
    # creates a deployment and starts a long-running
    # process that listens for scheduled work
    hello_world.serve(name="my-first-deployment",
        tags=["onboarding"],
        parameters={"goodbye": True},
        interval=60
    )

```

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

### 200 Response for GET saved search

Source: https://docs.prefect.io/v3/api-ref/rest-api/server/savedsearches/read-saved-search

Example 200 response body for a successful GET saved search request. Contains the search ID, creation/update timestamps, name, and filters.

```json
{
  "id": "3c90c3cc-0d44-4b50-8888-8dd25736052a",
  "created": "2023-11-07T05:31:56Z",
  "updated": "2023-11-07T05:31:56Z",
  "name": "<string>",
  "filters": [
    {
      "object": "<string>",
      "property": "<string>",
      "type": "<string>",
      "operation": "<string>",
      "value": "<unknown>"
    }
  ]
}
```

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

### Create virtualenv and authenticate to Prefect

Source: https://docs.prefect.io/v3/how-to-guides/deployment_infra/kubernetes

Creates a virtual environment, installs dependencies, and logs in to Prefect Cloud.

```bash
# Create a virtualenv & activate it
virtualenv prefect-demo
source prefect-demo/bin/activate

# Install dependencies of your flow
prefect-demo/bin/pip install -r requirements.txt

# Authenticate to Prefect & select the appropriate
# workspace to deploy your flows to
prefect-demo/bin/prefect cloud login

```

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

### PHP cURL GET saved search

Source: https://docs.prefect.io/v3/api-ref/rest-api/server/savedsearches/read-saved-search

PHP cURL example to GET a saved search. Replace the {id} with the UUID. The script prints the response or a cURL error message.

```php
<?php

$curl = curl_init();

curl_setopt_array($curl, [
  CURLOPT_URL => "https://api.example.com/saved_searches/{id}",
  CURLOPT_RETURNTRANSFER => true,
  CURLOPT_ENCODING => "",
  CURLOPT_MAXREDIRS => 10,
  CURLOPT_TIMEOUT => 30,
  CURLOPT_HTTP_VERSION => CURL_HTTP_VERSION_1_1,
  CURLOPT_CUSTOMREQUEST => "GET",
]);

$response = curl_exec($curl);
$err = curl_error($curl);

curl_close($curl);

if ($err) {
  echo "cURL Error #:" . $err;
} else {
  echo $response;
}
```

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

### Get result with raise_on_failure from a flow run in a different memory space

Source: https://docs.prefect.io/v3/api-ref/python/prefect-client-schemas-objects

Example showing how to get the result with raise_on_failure from a flow run in a different memory space using run_deployment.

```python
@flow
async def my_flow():
    raise ValueError("oh no!")
my_flow.deploy("my_deployment/my_flow")
flow_run = run_deployment("my_deployment/my_flow")
await flow_run.state.result(raise_on_failure=True) # Raises `ValueError("oh no!")`
```

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

### prefect.yaml with templating options

Source: https://docs.prefect.io/v3/deploy/infrastructure-concepts/prefect-yaml

Example prefect.yaml file showing how to reference step outputs ({{ build-image.tag }}), blocks ({{ prefect.blocks.docker-registry-credentials.dev-registry }}), variables ({{ prefect.variables.some_common_tag }}), and environment variables ({{ $my_deployment_tag }}) in build, push, and deployments sections.

```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 }}"
```

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

### Deployment naming examples for environments, schedules, and regions

Source: https://docs.prefect.io/v3/deploy/index

Shows how to name deployments using .serve() and .deploy() to reflect environment, schedule, or region. Use these patterns to create clear, scannable deployment names.

```python
# Environment-based names
my_flow.serve(name="staging")
my_flow.serve(name="production")

# Schedule-based names
my_flow.deploy(name="hourly-refresh", work_pool_name="my-pool")
my_flow.deploy(name="nightly-full-sync", work_pool_name="my-pool")

# Region or team-based names
my_flow.deploy(name="us-east-analytics", work_pool_name="my-pool")
my_flow.deploy(name="eu-west-compliance", work_pool_name="my-pool")
```

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

### Read Flow Run Input - GET Request Examples

Source: https://docs.prefect.io/v3/api-ref/rest-api/server/flow-runs/read-flow-run-input

These examples show how to call the Read Flow Run Input endpoint using a GET request to the URL https://api.example.com/flow_runs/{id}/input/{key}. The endpoint requires the flow run id and input key as path parameters. The response is a JSON object.

```bash
curl --request GET \
  --url https://api.example.com/flow_runs/{id}/input/{key}
```

```python
import requests

url = "https://api.example.com/flow_runs/{id}/input/{key}"

response = requests.get(url)

print(response.text)
```

```javascript
const options = {method: 'GET'};

fetch('https://api.example.com/flow_runs/{id}/input/{key}', options)
  .then(res => res.json())
  .then(res => console.log(res))
  .catch(err => console.error(err));
```

```php
<?php

$curl = curl_init();

curl_setopt_array($curl, [
  CURLOPT_URL => "https://api.example.com/flow_runs/{id}/input/{key}",
  CURLOPT_RETURNTRANSFER => true,
  CURLOPT_ENCODING => "",
  CURLOPT_MAXREDIRS => 10,
  CURLOPT_TIMEOUT => 30,
  CURLOPT_HTTP_VERSION => CURL_HTTP_VERSION_1_1,
  CURLOPT_CUSTOMREQUEST => "GET",
]);

$response = curl_exec($curl);
$err = curl_error($curl);

curl_close($curl);

if ($err) {
  echo "cURL Error #:" . $err;
} else {
  echo $response;
}
```

```go
package main

import (
	"fmt"
	"net/http"
	"io"
)

func main() {

	url := "https://api.example.com/flow_runs/{id}/input/{key}"

	req, _ := http.NewRequest("GET", url, nil)

	res, _ := http.DefaultClient.Do(req)

	defer res.Body.Close()
	body, _ := io.ReadAll(res.Body)

	fmt.Println(string(body))

}
```

```java
HttpResponse<String> response = Unirest.get("https://api.example.com/flow_runs/{id}/input/{key}")
  .asString();
```

```ruby
require 'uri'
require 'net/http'

url = URI("https://api.example.com/flow_runs/{id}/input/{key}")

http = Net::HTTP.new(url.host, url.port)
http.use_ssl = true

request = Net::HTTP::Get.new(url)

response = http.request(request)
puts response.read_body
```

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

### Run the serve process (python event_driven_deployments.py)

Source: https://docs.prefect.io/v3/how-to-guides/automations/chaining-deployments-with-events

Starts the `serve` process to listen for scheduled deployments. Run this command to test the event-driven deployments. The process must be running for the downstream deployment to be triggered.

```bash
python event_driven_deployments.py

```

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

### Run Prefect server with uvx

Source: https://docs.prefect.io/v3/get-started/install

Starts a Prefect server in an ephemeral Python environment using uvx.

```bash
uvx prefect server start

```

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

### Get a value from a parent transaction

Source: https://docs.prefect.io/v3/api-ref/python/prefect-transactions

Example showing that a child transaction can access values from its parent.

```python
with transaction() as parent:
    parent.set("key", "parent_value")
    with transaction() as child:
        assert child.get("key") == "parent_value"

```

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

### Project structure for background tasks example

Source: https://docs.prefect.io/v3/advanced/background-tasks

Shows the directory layout of the example project, including Dockerfile, compose.yaml, pyproject.toml, and the src/foo package with api.py and task.py.

```text
├── Dockerfile
├── README.md
├── compose.yaml
├── pyproject.toml
├── src
│   └── foo
│       ├── __init__.py
│       ├── _internal/*.py
│       ├── api.py
│       └── task.py

```

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

### Command to start the greeter flow runner

Source: https://docs.prefect.io/v3/advanced/interactive

Run this command in a terminal to start the greeter flow runner. It requires a Python environment with Prefect installed and pointed at a Prefect Cloud account or self-hosted server.

```bash
python my_file_name greeter

```

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

### Create a deployment with a YAML file using `prefect deploy`

Source: https://docs.prefect.io/v3/how-to-guides/deployments/create-deployments

Starts an interactive CLI that walks through creating a `prefect.yaml` file for declarative deployment definitions.

```bash
prefect deploy

```

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

### Install Prefect, Coiled, and Prefect-Coiled

Source: https://docs.prefect.io/v3/how-to-guides/deployment_infra/coiled

Install the required packages for using Prefect with Coiled.

```bash
pip install prefect coiled prefect-coiled

```

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

### 200 Response for GET /admin/storage

Source: https://docs.prefect.io/v3/api-ref/rest-api/server/admin/read-server-default-result-storage

Example successful response containing the default result storage block ID.

```json
{
  "default_result_storage_block_id": "3c90c3cc-0d44-4b50-8888-8dd25736052a"
}
```

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

### Create a work pool and start a worker with --type flag

Source: https://docs.prefect.io/v3/concepts/workers

Use the --type flag to create a work pool and start a worker in one command. The example uses the 'process' worker type.

```bash
prefect worker start -p "my-pool" --type "process"

```

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

### Run the example locally

Source: https://docs.prefect.io/v3/examples/run-dbt-with-prefect

Command to execute the example script that orchestrates a dbt lifecycle with Prefect, creating a local DuckDB file.

```bash
python examples/run_dbt_with_prefect.py

```

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

### Load and run a flow from a remote repository with flow.from_source

Source: https://docs.prefect.io/v3/how-to-guides/deployment_infra/run-flows-in-local-processes

Loads the 'hello' flow from the PrefectHQ/examples repository using flow.from_source and runs it. The output shows the flow run logs, including the created run ID, the 'Hello world!' message, and the Completed state.

```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()

```

```text
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()

```

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

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

Source: https://docs.prefect.io/v3/how-to-guides/self-hosted/server-windows

Install and configure Prefect server as a Windows service using NSSM. Run as Administrator. This ensures the server starts automatically on boot and provides automatic restarts on failure.

```bash
# 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

```

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

### Run Coiled Setup CLI

Source: https://docs.prefect.io/v3/how-to-guides/deployment_infra/coiled

Connect Coiled to your cloud account (AWS, Google Cloud, or Azure) using the interactive setup CLI.

```bash
coiled setup

```

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

### Output of creating a work pool and starting a worker

Source: https://docs.prefect.io/v3/concepts/workers

Example output when creating a work pool and starting a worker with the --type flag. Shows the worker ID and a log message indicating the work pool was created.

```text
Worker 'ProcessWorker d24f3768-62a9-4141-9480-a056b9539a25' started!
06:57:53.289 | INFO    | prefect.worker.process.processworker d24f3768-62a9-4141-9480-a056b9539a25 - Worker pool 'my-pool' created.

```

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

### Example: inspect storage configuration

Source: https://docs.prefect.io/v3/api-ref/cli/work-pool

Example commands to inspect storage configuration for a work pool, with and without JSON output.

```bash
$ prefect work-pool storage inspect "my-pool"
$ prefect work-pool storage inspect "my-pool" --output json

```

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

### Read Deployment with cURL and Python (GET)

Source: https://docs.prefect.io/v3/api-ref/rest-api/server/deployments/read-deployment

Retrieve a specific deployment using its ID. The cURL command works for any client; the Python example uses the requests library to make a GET request and prints the response body.

```cURL
curl --request GET \
  --url https://api.example.com/deployments/{id}

```

```Python
import requests

url = "https://api.example.com/deployments/{id}"

response = requests.get(url)

print(response.text)

```

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

### prefect sdk generate examples

Source: https://docs.prefect.io/v3/api-ref/cli/sdk

Examples of using `prefect sdk generate` with various options: output file, flow filter, and deployment filter.

```bash
$ prefect sdk generate --output ./my_sdk.py
$ prefect sdk generate --output ./my_sdk.py --flow my-etl-flow
$ prefect sdk generate --output ./my_sdk.py --deployment my-flow/production

```

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

### PHP cURL example for counting flow runs

Source: https://docs.prefect.io/v3/api-ref/rest-api/server/flow-runs/count-flow-runs

Uses cURL in PHP to send a POST request to the /flow_runs/count endpoint with a JSON body containing filter criteria. The example is incomplete in the source, showing only the beginning of the request setup.

```php
<?php

$curl = curl_init();

curl_setopt_array($curl, [
  CURLOPT_URL => "https://api.example.com/flow_runs/count",
  CURLOPT_RETURNTRANSFER => true,
  CURLOPT_ENCODING => "",
  CURLOPT_MAXREDIRS => 10,
  CURLOPT_TIMEOUT => 30,
  CURLOPT_HTTP_VERSION => CURL_HTTP_VERSION_1_1,
  CURLOPT_CUSTOMREQUEST => "POST",
  CURLOPT_POSTFIELDS => json_encode([
    'flows' => [
        'operator' => 'and_',
        'id' => [
                'any_' => [
                                '3c90c3cc-0d44-4b50-8888-8dd25736052a'
                ],
                'not_any_' => [
                                '3c90c3cc-0d44-4b50-8888-8dd25736052a'
                ]
        ],
        'deployment' => [
                'operator' => 'and_',
```

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

### prefect automation create examples

Source: https://docs.prefect.io/v3/api-ref/cli/automation

Example invocations for creating automations from a file or JSON string.

```bash
$ prefect automation create --from-file automation.yaml
$ prefect automation create -f automation.json
$ prefect automation create --from-json '{"name": "my-automation", "trigger": {...}, "actions": [...]}'
$ prefect automation create -j '[{"name": "auto1", ...}, {"name": "auto2", ...}]'

```

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

### start(self, run_once: bool = False, webserver: Optional[bool] = None) -> None

Source: https://docs.prefect.io/v3/api-ref/python/prefect-runner-runner

Starts the runner, which begins monitoring for and executing any scheduled work for all added flows. Optionally runs a single query loop and can start a webserver.

```APIDOC
## start(self, run_once: bool = False, webserver: Optional[bool] = None) -> None

### Description
Starts a runner. The runner will begin monitoring for and executing any scheduled work for all added flows.

### Method
start

### Signature
`start(self, run_once: bool = False, webserver: Optional[bool] = None) -> None`

### Parameters
- **run_once** (bool) - Optional - If True, the runner will go through one query loop and then exit.
- **webserver** (Optional[bool]) - Optional - A boolean for whether to start a webserver for this runner. If provided, overrides the default on the runner.

### Examples
Initialize a Runner, add two flows, and serve them by starting the Runner:
```python
import asyncio
from prefect import flow, Runner

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

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

if __name__ == "__main__":
    runner = Runner(name="my-runner")

    # Will be runnable via the API
    runner.add_flow(hello_flow)

    # Run on a cron schedule
    runner.add_flow(goodbye_flow, schedule={"cron": "0 * * * *"})

    asyncio.run(runner.start())
```

### Returns
None
```

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

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

Source: https://docs.prefect.io/v3/api-ref/python/prefect-client-schemas-objects

Example showing that calling result() on a failed state 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`
```

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

### Start Python REPL and use get_client to call hello()

Source: https://docs.prefect.io/v3/api-ref/python/prefect-client-__init__

Start a Python REPL with native await functionality using `python -m asyncio`, then use `get_client()` as an async context manager to call `client.hello()` and print the JSON response.

```bash
$ # start python REPL with native await functionality
$ python -m asyncio
from prefect.client.orchestration import get_client
async with get_client() as client:
    response = await client.hello()
    print(response.json())
👋
```

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

### 200 Response schema for GET /admin/settings

Source: https://docs.prefect.io/v3/api-ref/rest-api/server/admin/read-settings

Example of the 200 OK response body, showing the structure of the settings object returned by the endpoint.

```json
{
  "home": "~/.prefect",
  "profiles_path": "<string>",
  "debug_mode": false,
  "api": {
    "url": "<string>",
    "auth_string": "<string>",
    "key": "<string>",
    "tls_insecure_skip_verify": false,
    "ssl_cert_file": "<string>",
    "enable_http2": false,
    "request_timeout": 60
  },
  "cli": {
    "colors": true,
    "prompt": true,
    "wrap_lines": true
  },
  "client": {
    "max_retries": 5,
    "retry_jitter_factor": 0.2,
    "retry_extra_codes": "404,429,503",
    "csrf_support_enabled": true,
    "custom_headers": {
      "X-Custom-Header": "value"
    },
    "server_version_check_enabled": true,
    "metrics": {
      "enabled": false,
      "port": 4201
    }
  },
  "cloud": {
    "api_url": "https://api.prefect.cloud/api",
    "enable_orchestration_telemetry": true,
    "max_log_size": 25000,
    "ui_url": "<string>"
  },
  "deployments": {
    "default_work_pool_name": "<string>",
    "default_docker_build_namespace": "my-dockerhub-registry"
  },
  "events": {
    "worker_max_queue_size": 0
  },
  "experiments": {
    "warn": true
  },
  "flows": {
    "heartbeat_frequency": 180,
    "default_retries": 0,
    "default_retry_delay_seconds": 0
  },
  "internal": {
    "logging_level": "ERROR"
  },
  "logging": {
    "level": "INFO",
    "config_path": "<string>",
    "extra_loggers": "<string>",
    "log_prints": false,
    "colors": true,
    "markup": false,
    "to_api": {
      "enabled": true,
      "batch_interval": 2,
      "batch_size": 4000000,
      "max_log_size": 1000000,
      "when_missing_flow": "warn"
    }
  },
  "plugins": {
    "enabled": false,
    "allow": [
      "<string>"
    ],
    "deny": [
      "<string>"
    ],
    "setup_timeout_seconds": 20,
    "strict": false,
    "safe_mode": false
  },
  "results": {
    "default_serializer": "pickle",
    "persist_by_default": false,
    "default_storage_block": "<string>",
    "local_storage_path": "<string>"
  },
  "runner": {
    "process_limit": 5,
    "poll_frequency": 10,
    "crash_on_cancellation_failure": false,
    "auto_install_dependencies": false
  }
}
```

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

### prefect init command usage

Source: https://docs.prefect.io/v3/api-ref/cli/init

Shows the command syntax for prefect init. It initializes a Prefect project and accepts options such as --name, --recipe, and --field.

```bash
prefect init [OPTIONS]

```

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

### 200 response body for GET /events/filter/next

Source: https://docs.prefect.io/v3/api-ref/rest-api/server/events/read-account-events-page

Example JSON response for a successful request, containing an events array, total count, and next_page token.

```json
{
  "events": [
    {
      "occurred": "2023-11-07T05:31:56Z",
      "event": "<string>",
      "resource": {},
      "id": "3c90c3cc-0d44-4b50-8888-8dd25736052a",
      "related": [
        {}
      ],
      "payload": {},
      "follows": "3c90c3cc-0d44-4b50-8888-8dd25736052a",
      "received": "2023-11-07T05:31:56Z"
    }
  ],
  "total": 123,
  "next_page": "<string>"
}
```
Tokens:
Raw