flow.from_source(...)

prefect.yaml

EXTRA_PIP_PACKAGES

prefect-shell

python
# flows/run_script.py
from prefect import flow
from prefect_shell import shell_run_command  # pip install prefect-shell

@flow(log_prints=True)
async def run_script():
    # Execute a Python script in your repo
    out = await shell_run_command("python scripts/my_job.py")
    print(out)
    return out

if __name__ == "__main__":
    # Deploy from a Git repo (public or private with credentials, see notes below)
    flow.from_source(
        source="<https://github.com/your-org/your-repo.git>",
        entrypoint="flows/run_script.py:run_script",
    ).deploy(
        name="run-script",
        work_pool_name="my-work-pool",   # process, docker, or k8s
        # For small adds on Process/Docker/K8s pools you can do:
        # job_variables={"env": {"EXTRA_PIP_PACKAGES": "prefect-shell"}},
        # For Docker/K8s in prod, prefer a custom image with all deps preinstalled.
    )

yaml
# prefect.yaml
pull:
  - prefect.deployments.steps.git_clone:
      id: repo
      repository: <https://github.com/your-org/your-repo.git>
      # For private repos:
      # credentials: "{{ prefect.blocks.github-credentials.my-creds }}"
  - prefect.deployments.steps.pip_install_requirements:
      directory: "{{ repo.directory }}"
      requirements_file: requirements.txt
  - prefect.deployments.steps.set_working_directory:
      directory: "{{ repo.directory }}"

deployments:
  - name: run-script
    entrypoint: flows/run_script.py:run_script
    work_pool:
      name: my-process-pool

python
from prefect import flow
from prefect_shell import shell_run_command

@flow
async def run_script():
    return await shell_run_command("python scripts/my_job.py")

flow.from_source("<https://github.com/org/repo.git>", entrypoint="file.py:flow")

git_clone

GitHubCredentials

git_clone

pip install

pip_install_requirements

requirements.txt

job_variables.env.EXTRA_PIP_PACKAGES="pandas requests"

prefect_shell.shell_run_command("...")

ShellOperation(commands=[...]).run()

export PREFECT_LOGGING_SETTINGS_PATH="/opt/prefect/logging.yml"
    prefect worker start --pool my-process-pool --type process

deployments:
      - name: my-deployment
        work_pool:
          name: my-docker-pool
          job_variables:
            image: my-registry/my-image:latest
            env:
              PREFECT_LOGGING_SETTINGS_PATH: "/etc/prefect/logging.yml"

version: 1
disable_existing_loggers: False

formatters:
  json:
    '()': prefect.logging.formatters.JsonFormatter
    fmt: default

handlers:
  console:
    level: 0
    class: prefect.logging.handlers.PrefectConsoleHandler
    formatter: json
    stream: <ext://sys.stderr>
  api:
    level: 0
    class: prefect.logging.handlers.APILogHandler

loggers:
  prefect:
    level: "${PREFECT_LOGGING_LEVEL}"
    handlers: [console, api]

root:
  level: INFO
  handlers: [console]

# at the top of scripts/my_job.py
    import logging
    logging.basicConfig(level=<http://logging.INFO|logging.INFO>)  # so INFO lines actually emit

shell_run_command

shell_run_command

ShellOperation(stream_output=True)

from prefect import flow
from prefect_shell import shell_run_command

@flow
async def run_script():
    # streams stdout lines at INFO (adjust with stream_level)
    return await shell_run_command("python -u scripts/my_job.py")

from prefect import flow
from prefect_shell import ShellOperation

@flow
async def run_job():
    async with ShellOperation(
        commands=[
            "pip install -r requirements.txt",
            "python -u scripts/my_job.py",
        ],
        working_dir="/app",
        stream_output=True,  # will prefix lines (PID … stream output / stderr)
    ) as op:
        result = await op.arun()
    return result

from prefect import flow
from prefect_shell import shell_run_command

@flow
def run():
    out = shell_run_command("python -u scripts/my_job.py", return_all=True)
    return out

from prefect import flow
from prefect_shell import shell_run_command

@flow
async def run():
    out = await shell_run_command("python -u scripts/my_job.py", return_all=True)
    return out

version: 1
disable_existing_loggers: False

formatters:
  json:
    '()': prefect.logging.formatters.JsonFormatter
    fmt: default        # or "pretty"
    dmft: null          # unused but required by the constructor
    style: '%'          # unused but required by the constructor

handlers:
  console:
    level: 0
    class: prefect.logging.handlers.PrefectConsoleHandler
    formatter: json
    stream: <ext://sys.stderr>
  api:
    level: 0
    class: prefect.logging.handlers.APILogHandler

loggers:
  prefect:
    level: "${PREFECT_LOGGING_LEVEL}"
    handlers: [console, api]
    propagate: False

root:
  level: INFO
  handlers: [console]

await shell_run_command("python -u scripts/my_job.py 2>&1")

await shell_run_command(
      "python -u scripts/my_job.py 2>&1",
      shell="bash"
    )

await shell_run_command("python -u scripts/my_job.py 2>&1", shell="powershell")

# inside scripts/my_job.py
  import logging, sys
  logging.basicConfig(level=<http://logging.INFO|logging.INFO>, stream=sys.stdout)

prefect

prefect

prefect

prefect.flow_runs

prefect.task_runs

version: 1
disable_existing_loggers: False

formatters:
  json:
    '()': prefect.logging.formatters.JsonFormatter
    fmt: default     # or "pretty"
    dmft: null       # required by constructor; unused
    style: '%'       # required by constructor; unused

handlers:
  console:
    class: prefect.logging.handlers.PrefectConsoleHandler
    level: 0
    formatter: json
    stream: <ext://sys.stderr>
  api:
    class: prefect.logging.handlers.APILogHandler
    level: 0

loggers:
  # Override Prefect's default logger so it uses JSON on console
  prefect:
    level: "${PREFECT_LOGGING_LEVEL}"
    handlers: [console, api]
    propagate: False

  # Make sure flow/task run child loggers also use JSON, not inherited defaults
  prefect.flow_runs:
    level: NOTSET
    handlers: [console, api]
    propagate: False

  prefect.task_runs:
    level: NOTSET
    handlers: [console, api]
    propagate: False

  # Optional: align uvicorn/fastapi if you run server bits in the same process
  uvicorn:
    level: INFO
    handlers: [console]
    propagate: False
  fastapi:
    level: INFO
    handlers: [console]
    propagate: False

root:
  level: INFO
  handlers: [console]

prefect

prefect.flow_runs

prefect.task_runs

propagate: False

shell_run_command

ShellOperation

logging.StreamHandler

shell_run_command

await shell_run_command("python your_script.py 2>&1", shell="bash")

import logging, sys

  handler = logging.StreamHandler(sys.stdout)  # stdout instead of default stderr
  formatter = logging.Formatter("%(asctime)s.%(msecs)03d %(levelname)-7s | %(message)s")
  handler.setFormatter(formatter)

  root = logging.getLogger()
  root.setLevel(<http://logging.INFO|logging.INFO>)
  root.handlers = [handler]

json

handlers:
    api:
      level: INFO
      class: prefect.logging.handlers.APILogHandler

  loggers:
    prefect.task_runs:
      level: NOTSET
      handlers: [json, api]
      propagate: False
    prefect.flow_runs:
      level: NOTSET
      handlers: [json, api]
      propagate: False

  root:
    level: INFO
    handlers: [json, api]

PREFECT_LOGGING_TO_API_ENABLED=true

shell_run_command

from prefect import flow
  from prefect_shell import shell_run_command

  @flow
  async def demo():
      await shell_run_command("python your_script.py 2>&1", shell="bash")

2>&1

APILogHandler

prefect

prefect-shell

shell_run_command

shell_run_command

@task

shell_run_command

shell_run_command

from prefect import task
from prefect.logging import get_run_logger
from prefect_shell import shell_run_command
import logging

@task(
    name="Script",
    task_run_name="{script}",
    retries=2,
    retry_delay_seconds=10,  # single value; remove the list
    timeout_seconds=14400,
)
async def extract_and_load_task(script: str):
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Running script - {script}")

    # Redirect stderr to stdout so both appear in the parent logs
    cmd = f"poetry run python -m scripts.{script} 2>&1"

    # Use a shell that understands redirection
    result = await shell_run_command(
        cmd,
        shell="bash",
        stream_level=<http://logging.INFO|logging.INFO>,  # explicit
    )
    return result

import logging, sys, logging.config, yaml

with open(".prefect/logging.yml", "rt") as f:
    config = yaml.safe_load(f.read())
logging.config.dictConfig(config)

# Force stdout for your script’s root logger if needed
for h in logging.getLogger().handlers:
    if isinstance(h, logging.StreamHandler):
        h.stream = sys.stdout

from prefect import flow
from prefect_shell import shell_run_command
import logging

@flow
async def test():
    await shell_run_command("echo hello 2>&1", shell="bash", stream_level=<http://logging.INFO|logging.INFO>)

if __name__ == "__main__":
    import anyio
    anyio.run(test)

await shell_run_command("echo hello", shell="bash", stream_level=logging.DEBUG)

from prefect import flow
from prefect_shell import shell_run_command

@flow
def f():
    return shell_run_command("echo hello")

from prefect import flow
from prefect_shell import shell_run_command

@flow
async def f():
    return await shell_run_command("echo hello")

shell_run_command("poetry run python -m scripts.<name> 2>&1", shell="bash")

from functools import partial
from prefect import flow
from prefect_shell import ShellOperation

@flow(
    name="Split Data Source Chunk",
    task_runner=ConcurrentTaskRunner(),
    flow_run_name="{data_sources_chunk}",
)
def chunk_subflow(data_sources_chunk: list[str], env: str):
    futures = []
    for x in data_sources_chunk:
        if not x:
            continue
        t = extract_and_load_task.with_options(
            name=f"'{x}' data loader",
            on_failure=[partial(data_loader_failure_notify_slack, env=env, data_source_name=x)],
        )
        futures.append((x, t.submit(x)))  # submit => Prefect manages state + hooks

    results = []
    for x, fut in futures:
        try:
            results.append(fut.result())  # raises on final failure; hook already ran
        except Exception as e:
            print(f"Data loader '{x}' failed: {e}, continuing run")
    return results

from prefect.states import Failed

  @task(on_failure=[...])
  def extract_and_load_task(data_source_name):
      logger = get_run_logger()
      <http://logger.info|logger.info>(f"Running data loader - {data_source_name}")
      try:
          return ShellOperation(commands=[f"poetry run python -m data_loaders.{data_source_name}"]).run()
      except Exception as e:
          # Emit a terminal Failed state so the hook runs
          return Failed(message=str(e))

from prefect.logging import get_run_logger
  logger = get_run_logger()
  <http://logger.info|logger.info>(f"Running data loader - {data_source_name}")

retry_delay_seconds

.submit()

from functools import partial
from prefect import flow
from prefect.logging import get_run_logger
from prefect_shell import shell_run_command

def data_loader_failure_notify_slack(task, task_run, state, env, data_source_name):
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Sending Slack message for data loader '{data_source_name}' failure")
    # ... your Slack code here ...

@flow(name="Data loader", flow_run_name="{data_source_name}")
def extract_and_load_task(data_source_name: str, env: str):
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Running data loader - {data_source_name}")

    shell_with_hook = shell_run_command.with_options(
        name=f"load-{data_source_name}",
        on_failure=[partial(data_loader_failure_notify_slack, env=env, data_source_name=data_source_name)],
        retries=2,                # move retries to the task, if desired
        retry_delay_seconds=15,
    )

    # sequential execution; no concurrency
    return shell_with_hook(f"poetry run python -m data_loaders.{data_source_name}")

@flow(name="Split Data Source Chunk", flow_run_name="{data_sources_chunk}")
def chunk_subflow(data_sources_chunk: list[str], env: str):
    results = []
    for x in data_sources_chunk:
        if not x:
            continue
        try:
            results.append(extract_and_load_task(x, env))
        except Exception as e:
            get_run_logger().error(f"Data loader '{x}' failed: {e}, continuing run", exc_info=True)
    return results

from prefect.logging import get_run_logger
  logger = get_run_logger()

<http://logger.info|logger.info>

from prefect.blocks.notifications import SlackWebhook
  from prefect.blocks.abstract import NotificationError
  from prefect.logging import get_run_logger
  from zoneinfo import ZoneInfo
  from datetime import datetime
  from prefect.runtime import flow_run  # if you use get_flow_run_ui_url()

  def data_loader_failure_notify_slack(task, task_run, state, env: str, data_source_name: str):
      logger = get_run_logger()
      <http://logger.info|logger.info>(f"Sending Slack message for data loader '{data_source_name}' failure")

      # IMPORTANT: use .load (sync), not .aload
      webhook = SlackWebhook.load("slack-integration" if env == "prod" else "slack-staging")

      ts = datetime.now(ZoneInfo('America/New_York'))
      try:
          with webhook.raise_on_failure():
              webhook.notify(
                  body=(
                      f"Data loader `{data_source_name}` failed at {ts}\n"
                      f"Flow run URL: {flow_run.get_flow_run_ui_url()}"
                  ),
                  subject=f"Data loader failure - {data_source_name}",
              )
          <http://logger.info|logger.info>(f"Sent Slack message for data loader '{data_source_name}' failure")
      except NotificationError as e:
          logger.error(f"Slack notification failed: {e.log}", exc_info=True)
      except Exception as e:
          logger.error(f"Error sending Slack notification: {e}", exc_info=True)

webhook = SlackWebhook.load("...")

webhook.notify(...)

webhook = await SlackWebhook.aload("...")

from prefect.blocks.notifications import SlackWebhook

from prefect_slack import SlackWebhook

.load(...)

await .aload(...)

.aload(...)

print(type(webhook))

<class 'prefect.blocks.notifications.SlackWebhook'>

<class 'coroutine'>

f"... {datetime.now(ZoneInfo('America/New_York'))} ..."

from datetime import datetime
from zoneinfo import ZoneInfo
from functools import partial

from prefect.blocks.notifications import SlackWebhook
from prefect.blocks.abstract import NotificationError
from prefect.logging import get_run_logger
from prefect.runtime import flow_run

def data_loader_failure_notify_slack(task, task_run, state, env: str, data_source_name: str):
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Sending Slack message for data loader '{data_source_name}' failure")

    # IMPORTANT: use .load (sync), not .aload
    slack_webhook_block: SlackWebhook = SlackWebhook.load(
        "slack-integration" if env == "prod" else "slack-staging"
    )

    ts = datetime.now(ZoneInfo('America/New_York'))
    ui_link = flow_run.ui_url  # attribute in 3.x; may be None if no UI configured

    try:
        with slack_webhook_block.raise_on_failure():
            slack_webhook_block.notify(
                body=(
                    f"Data loader `{data_source_name}` failed at {ts}\n"
                    f"Flow run URL: {ui_link or 'N/A'}"
                ),
                subject=f"Data loader failure - {data_source_name}",
            )
            <http://logger.info|logger.info>(f"Sent Slack message for data loader '{data_source_name}' failure")
    except NotificationError as e:
        logger.error(f"Slack NotificationError: {e.log}", exc_info=True)
    except Exception as e:
        logger.error(f"Error sending Slack notification: {e}", exc_info=True)

slack_webhook_block = await SlackWebhook.aload("slack-integration" if env == "prod" else "slack-staging")

print(type(slack_webhook_block))

from datetime import datetime
from zoneinfo import ZoneInfo
from prefect.blocks.notifications import SlackWebhook
from prefect.blocks.abstract import NotificationError
from prefect.logging import get_run_logger
from prefect.runtime import flow_run

def data_loader_failure_notify_slack(task, task_run, state, env: str, data_source_name: str):
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Sending Slack message for data loader '{data_source_name}' failure")

    # Force sync even in async context
    slack_webhook_block: SlackWebhook = SlackWebhook.load(
        "slack-integration" if env == "prod" else "slack-staging",
        _sync=True,
    )

    ts = datetime.now(ZoneInfo('America/New_York'))
    ui_link = flow_run.ui_url  # Prefect 3.x attribute

    try:
        with slack_webhook_block.raise_on_failure():
            slack_webhook_block.notify(
                body=(
                    f"Data loader `{data_source_name}` failed at {ts}\n"
                    f"Flow run URL: {ui_link or 'N/A'}"
                ),
                subject=f"Data loader failure - {data_source_name}",
                _sync=True,  # also force sync here
            )
            <http://logger.info|logger.info>(f"Sent Slack message for data loader '{data_source_name}' failure")
    except NotificationError as e:
        logger.error(f"Slack NotificationError: {e.log}", exc_info=True)
    except Exception as e:
        logger.error(f"Error sending Slack notification: {e}", exc_info=True)

from datetime import datetime
from zoneinfo import ZoneInfo
from prefect.blocks.notifications import SlackWebhook
from prefect.blocks.abstract import NotificationError
from prefect.logging import get_run_logger
from prefect.runtime import flow_run

async def data_loader_failure_notify_slack(task, task_run, state, env: str, data_source_name: str):
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Sending Slack message for data loader '{data_source_name}' failure")

    slack_webhook_block: SlackWebhook = await SlackWebhook.aload(
        "slack-integration" if env == "prod" else "slack-staging"
    )

    ts = datetime.now(ZoneInfo('America/New_York'))
    ui_link = flow_run.ui_url

    try:
        with slack_webhook_block.raise_on_failure():
            await slack_webhook_block.notify(
                body=(
                    f"Data loader `{data_source_name}` failed at {ts}\n"
                    f"Flow run URL: {ui_link or 'N/A'}"
                ),
                subject=f"Data loader failure - {data_source_name}",
            )
        <http://logger.info|logger.info>(f"Sent Slack message for data loader '{data_source_name}' failure")
    except NotificationError as e:
        logger.error(f"Slack NotificationError: {e.log}", exc_info=True)
    except Exception as e:
        logger.error(f"Error sending Slack notification: {e}", exc_info=True)

from datetime import datetime
from zoneinfo import ZoneInfo
from prefect.blocks.notifications import SlackWebhook
from prefect.blocks.abstract import NotificationError
from prefect.logging import get_run_logger
from prefect.runtime import flow_run

async def data_loader_failure_notify_slack(task, task_run, state, env: str, data_source_name: str):
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Sending Slack message for data loader '{data_source_name}' failure")

    # Use the async loader
    slack_webhook_block: SlackWebhook = await SlackWebhook.aload(
        "slack-integration" if env == "prod" else "slack-staging"
    )

    ts = datetime.now(ZoneInfo('America/New_York'))
    ui_link = flow_run.ui_url  # Prefect 3.x attribute

    try:
        with slack_webhook_block.raise_on_failure():
            # Await the async-dispatched notify
            await slack_webhook_block.notify(
                body=(
                    f"Data loader `{data_source_name}` failed at {ts}\n"
                    f"Flow run URL: {ui_link or 'N/A'}"
                ),
                subject=f"Data loader failure - {data_source_name}",
            )
        <http://logger.info|logger.info>(f"Sent Slack message for data loader '{data_source_name}' failure")
    except NotificationError as e:
        logger.error(f"Slack NotificationError: {e.log}", exc_info=True)
    except Exception as e:
        logger.error(f"Error sending Slack notification: {e}", exc_info=True)

from datetime import datetime
from zoneinfo import ZoneInfo
from functools import partial

from prefect.blocks.notifications import SlackWebhook
from prefect.blocks.abstract import NotificationError
from prefect.logging import get_run_logger
from prefect.runtime import flow_run

async def data_loader_failure_notify_slack(task, task_run, state, env: str, data_source_name: str):
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Sending Slack message for data loader '{data_source_name}' failure")

    # Async load in async hook
    slack_webhook_block: SlackWebhook = await SlackWebhook.aload(
        "slack-integration" if env == "prod" else "slack-staging"
    )

    ts = datetime.now(ZoneInfo('America/New_York'))
    ui_link = flow_run.ui_url  # attribute in Prefect 3.x

    try:
        with slack_webhook_block.raise_on_failure():
            # Await notify in async context
            await slack_webhook_block.notify(
                body=(
                    f"Data loader `{data_source_name}` failed at {ts}\n"
                    f"Flow run URL: {ui_link or 'N/A'}"
                ),
                subject=f"Data loader failure - {data_source_name}",
            )
        <http://logger.info|logger.info>(f"Sent Slack message for data loader '{data_source_name}' failure")
    except NotificationError as e:
        logger.error(f"Slack NotificationError: {e.log}", exc_info=True)
    except Exception as e:
        logger.error(f"Error sending Slack notification: {e}", exc_info=True)

with_options(on_failure=[partial(...)]);

shell_run_command

return_all=True

from prefect import flow
from prefect.logging import get_run_logger
from prefect_shell import shell_run_command

@flow
async def run():
    logger = get_run_logger()
    lines = await shell_run_command(
        command="bash -lc 'echo one; echo two; echo three'",
        return_all=True,
    )
    for line in lines:
        <http://logger.info|logger.info>(line)  # one JSON record per line

ShellOperation

from prefect import flow
from prefect.logging import get_run_logger
from prefect_shell import ShellOperation

@flow
def run():
    logger = get_run_logger()
    op = ShellOperation(
        commands=["echo one", "echo two", "echo three"],
        stream_output=False,  # don't stream clumped logs
    )
    result_lines = op.run()  # returns list[str]
    for line in result_lines:
        <http://logger.info|logger.info>(line)  # one JSON record per line

.readline()

subprocess.Popen(..., text=True)

for line in proc.stdout

shell_run_command

stdbuf -oL -eL

command="bash -lc 'stdbuf -oL -eL your_command_here'"

prefect_shell.shell_run_command

ShellOperation

commands.py

shell_run_command

shell_run_command

from prefect import flow
from prefect.logging import get_run_logger
from prefect_shell import shell_run_command

@flow
async def run():
    logger = get_run_logger()
    lines = await shell_run_command(
        command="bash -lc 'echo one; echo two; echo three'",
        return_all=True,  # ensures you get a list[str]
    )
    for line in lines:
        <http://logger.info|logger.info>(line)  # one JSON log entry per line

ShellOperation

stream_output=False

from prefect import flow
from prefect.logging import get_run_logger
from prefect_shell import ShellOperation

@flow
def run():
    logger = get_run_logger()
    op = ShellOperation(
        commands=["echo one", "echo two", "echo three"],
        stream_output=False,  # don't emit chunked stream logs
    )
    result_lines = op.run()  # returns list[str]
    for line in result_lines:
        <http://logger.info|logger.info>(line)  # one JSON log entry per line

subprocess.Popen(..., text=True)

from prefect import flow, task
from prefect.logging import get_run_logger
import subprocess

@task
def run_shell_line_stream(cmd: str):
    logger = get_run_logger()
    with subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1) as p:
        for line in p.stdout:  # line-by-line
            <http://logger.info|logger.info>(line.rstrip())
    return p.wait()

@flow
def run():
    run_shell_line_stream("bash -lc 'echo one; echo two; echo three'")

bash -lc 'stdbuf -oL -eL your_command_here'

PYTHONUNBUFFERED=1

env

-u

from prefect import flow
from prefect_shell import ShellOperation

@flow
def run():
    op = ShellOperation(
        commands=["bash -lc 'echo one; sleep 1; echo two; sleep 1; echo three'"],
        stream_output=True,  # live streaming
    )
    op.run()  # logs each line as it's emitted

shell_run_command

from prefect import flow
from prefect.logging import get_run_logger
import anyio
import logging
from anyio.streams.text import TextReceiveStream

async def stream_lines(cmd: str, env: dict | None = None, cwd: str | None = None) -> int:
    logger = get_run_logger()
    # Adjust shell as needed (Windows vs Linux)
    shell = ["/bin/bash", "-lc", cmd]

    async with await anyio.open_process(shell, env=env, cwd=cwd) as proc:
        async def pump(byte_stream, level: int):
            recv = TextReceiveStream(byte_stream)
            buf = ""
            async for chunk in recv:
                buf += chunk
                while "\n" in buf:
                    line, buf = buf.split("\n", 1)
                    if line:
                        logger.log(level, line)
            if buf:
                logger.log(level, buf)

        async with anyio.create_task_group() as tg:
            tg.start_soon(pump, proc.stdout, <http://logging.INFO|logging.INFO>)
            tg.start_soon(pump, proc.stderr, logging.WARNING)

        return await proc.wait()

@flow
async def run():
    rc = await stream_lines("echo one; sleep 1; echo two; sleep 1; echo three")
    return rc

bash -lc 'stdbuf -oL -eL your_command'

PYTHONUNBUFFERED=1

python -u script.py

shell_run_command

from functools import partial
from datetime import datetime
from zoneinfo import ZoneInfo

from prefect import flow, task, State, Task
from prefect.logging import get_run_logger
from prefect.runtime import flow_run
from prefect.blocks.notifications import SlackWebhook, NotificationError
from prefect_shell import ShellOperation


async def script_failure_notify_slack(
    task: Task, task_run, state: State, env: str, script_name: str
):
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Sending Slack message for script '{script_name}' failure")

    slack_webhook_block: SlackWebhook = await SlackWebhook.aload(
        "slack-integration" if env == "prod" else "slack-staging"
    )

    try:
        with slack_webhook_block.raise_on_failure():
            await slack_webhook_block.notify(
                body=(
                    f"script `{script_name}` failed at "
                    f"{datetime.now(ZoneInfo('America/New_York')).strftime('%H:%M:%S %Z on %Y-%m-%d')}\n"
                    f"Flow run URL: {flow_run.ui_url}"
                ),
                subject=f"script failure - {script_name}",
            )
            <http://logger.info|logger.info>(f"Sent Slack message for script '{script_name}' failure")
    except Exception as e:
        logger.error(f"Error sending Slack notification: {e}", exc_info=True)
        if isinstance(e, NotificationError):
            logger.error(f"Apprise NotificationError: {e.log}")


@task
def run_script_sync(script_name: str):
    # Real-time, per-line streaming
    op = ShellOperation(
        commands=[f"poetry run python -u -m scripts.{script_name}"],
        stream_output=True,
        # Encourage immediate flushing from Python children
        env={"PYTHONUNBUFFERED": "1"},
    )
    # Raises on non-zero exit -> triggers on_failure hook
    return op.run()


@flow(
    name="script",
    flow_run_name="{script_name}",
    retries=2,
    retry_delay_seconds=15,
    timeout_seconds=14400,
)
def extract_and_load_task_with_failure_hook(script_name: str, env: str):
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Running script - {script_name}")

    task_with_hook = run_script_sync.with_options(
        name=f"Script - {script_name}",
        on_failure=[
            partial(
                script_failure_notify_slack,
                env=env,
                script_name=script_name,
            )
        ],
        retries=2,
        retry_delay_seconds=15,
    )

    # sequential execution; no concurrency
    return task_with_hook(script_name)


@flow(
    name="Split Scripts Chunk",
    flow_run_name="{scripts_chunk}",
)
def chunk_subflow(scripts_chunk: list[str], env: str):
    logger = get_run_logger()
    results = []
    for x in scripts_chunk:
        if x:
            try:
                result = extract_and_load_task_with_failure_hook(x, env)
                results.append(result)
            except Exception as e:
                logger.error(
                    f"script '{x}' failed: {e}, continuing run", exc_info=True
                )
    return results

stdbuf -oL -eL

commands=[...]

prefect.yaml

prefect-github

pip install prefect-github

gh-creds

prefect block register -m prefect_github

GitHubCredentials

gh-creds

git_clone

GitHubCredentials

pull_with_block

poetry install

pull

entrypoint

pip install -e .

include_submodules: true

git_clone

commit_sha

branch

git_clone

---
prefect-version: "3.6.25"
name: multi-pool-project

# DRY definitions for work pools
definitions:
  process_pool: &process_pool
    name: process-pool
    work_queue_name: default
    job_variables:
      env:
        LOG_LEVEL: INFO

  docker_pool: &docker_pool
    name: docker-pool
    work_queue_name: default
    job_variables:
      image: "your-registry/your-image:py3.11"
      env:
        PIP_INDEX_URL: "<https://pypi.org/simple>"

  k8s_pool: &k8s_pool
    name: k8s-pool
    work_queue_name: default
    job_variables:
      image: "your-registry/your-image:py3.11"
      namespace: "prefect"
      env:
        LOG_LEVEL: DEBUG

# Each deployment picks the pool it should run on
deployments:
  - name: flow-on-process
    entrypoint: flows/my_flow.py:my_flow
    work_pool: *process_pool

  - name: flow-on-docker
    entrypoint: flows/my_flow.py:my_flow
    work_pool: *docker_pool

  - name: flow-on-k8s
    entrypoint: flows/my_flow.py:my_flow
    work_pool: *k8s_pool

prefect work-pool create "process-pool" --type process
prefect work-pool create "docker-pool" --type docker
prefect work-pool create "k8s-pool" --type kubernetes

prefect deploy

prefect deploy

pull_with_block

prefect.yaml

--entrypoint

deployments:
  - name: "Kal Testing With prefect.yaml"
    entrypoint: "data_loaders/prefect_jobs_entrypoint.py:kal_testing_with_prefect_yaml"
    work_pool: *kal_testing
    pull:
      - prefect.deployments.steps.pull_with_block:
          block: "{{ prefect.blocks.github-repository.data-platform-github-kal }}"
      - prefect.deployments.steps.set_working_directory:
          directory: "{{ pull_with_block.directory }}"
      - prefect.deployments.steps.pip_install_requirements:
          directory: "{{ pull_with_block.directory }}"
          requirements_file: "requirements.txt"

data_loaders/prefect_jobs_entrypoint.py

prefect deploy

prefect deploy data_loaders/prefect_jobs_entrypoint.py:kal_testing_with_prefect_yaml

entrypoint

prefect deploy

flow.from_source(...).deploy(...)

&kal_testing

&kal_testing

pull_with_block

entrypoint

---
  prefect-version: "3.6.25"
  name: data-platform

  build: null
  push: null

  deployments:
    - name: my-deploy
      entrypoint: flows/my_flow.py:my_flow
      work_pool:
        name: process-pool
        work_queue_name: default
        job_variables: {}
      pull:
        - prefect.deployments.steps.git_clone:
            id: clone
            repository: "<https://github.com/your-org/your-repo.git>"
            branch: "main"
            credentials: "{{ prefect.blocks.github-credentials.gh-creds }}"
        - prefect.deployments.steps.pip_install_requirements:
            directory: "{{ clone.directory }}"
            requirements_file: "requirements.txt"

from prefect import flow
  from prefect.docker import DockerImage

  @flow
  def my_flow(): ...

  if __name__ == "__main__":
      my_flow.from_source(
          source="<https://github.com/your-org/your-repo.git>",
          entrypoint="flows/my_flow.py:my_flow",
      ).deploy(
          name="my-deploy",
          work_pool_name="docker-pool",
          image=DockerImage(name="your-registry/your-image", tag="py3.11"),
          build=True,   # build the image at deploy time
          push=True,    # push it to your registry
      )

prefect work-pool create "docker-pool" --type docker --base-job-template ./base-job-template.json

block=...

pull_with_block

block_type_slug

block_document_name

data-platform-github-kal

github-repository

prefect-version: "3.6.25"
name: data-platform
description: "Deploy flows that import and run scripts directly, with per-branch git pulls and pre-import dependency installs."

build: null
push: null

definitions:
  kal_testing_pool: &kal_testing
    name: "kal-testing"
    work_queue_name: default
    job_variables: {}

deployments:
  - name: "Kal Testing With prefect.yaml"
    entrypoint: "data_loaders/prefect_jobs_entrypoint.py:kal_testing_with_prefect_yaml"
    description: "Testing using prefect.yaml deploy, direct load scripts"
    tags: ["staging"]
    work_pool: *kal_testing
    pull:
      - prefect.deployments.steps.pull_with_block:
          block_type_slug: "github-repository"
          block_document_name: "data-platform-github-kal"
      - prefect.deployments.steps.set_working_directory:
          directory: "{{ pull_with_block.directory }}"
      - prefect.deployments.steps.pip_install_requirements:
          directory: "{{ pull_with_block.directory }}"
          requirements_file: "requirements.txt"

data-platform-github-kal

github-repository

pip install prefect-github

requirements_file

&kal_testing

&kal_testing

prefect-version: "3.6.25"
name: data-platform

build: null
push: null

definitions:
  kal_testing_pool: &kal_testing
    name: "kal-testing"
    work_queue_name: default
    job_variables: {}

deployments:
  - name: "Kal Testing With prefect.yaml"
    entrypoint: "data_loaders/prefect_jobs_entrypoint.py:kal_testing_with_prefect_yaml"
    description: "Testing using prefect.yaml deploy, direct load scripts"
    tags: ["staging"]
    work_pool: *kal_testing
    pull:
      - prefect.deployments.steps.pull_with_block:
          id: pull_code
          block_type_slug: "github-repository"
          block_document_name: "data-platform-github-kal"
      - prefect.deployments.steps.set_working_directory:
          id: set_cwd
          directory: "{{ pull_code.directory }}"
      - prefect.deployments.steps.pip_install_requirements:
          directory: "{{ pull_code.directory }}"
          requirements_file: "requirements.txt"

deployments:
  - name: "Kal Testing With prefect.yaml"
    entrypoint: "data_loaders/prefect_jobs_entrypoint.py:kal_testing_with_prefect_yaml"
    work_pool: *kal_testing
    pull:
      - prefect.deployments.steps.pull_with_block:
          id: pull_code
          block_type_slug: "github-repository"
          block_document_name: "data-platform-github-kal"
      - prefect.deployments.steps.pip_install_requirements:
          directory: "{{ pull_code.directory }}"
          requirements_file: "requirements.txt"

directory

{{ pull_code.directory }}

pip install prefect-github

formatters:
  json:
    class: prefect.logging.formatters.JsonFormatter
    format: default   # or "pretty" for indented

handlers:
  console_json:
    class: prefect.logging.handlers.PrefectConsoleHandler
    level: 0
    formatter: json
    stream: <ext://sys.stderr>

loggers:
  # Prefect’s deployment steps use this logger
  deployment:
    level: INFO
    handlers: [console_json]
    propagate: false

root:
  level: WARNING
  handlers: [console_json]

export PREFECT_LOGGING_SETTINGS_PATH=/path/to/logging.yml

export PREFECT_LOGGING_HANDLERS_CONSOLE_FORMATTER=json

export PREFECT_LOGGING_FORMATTERS_JSON_CLASS=prefect.logging.formatters.JsonFormatter

export PREFECT_LOGGING_FORMATTERS_JSON_FORMAT=default

stream_output: false

- prefect.deployments.steps.utility.run_shell_script:
    id: pip_install
    # Run pip but do not stream raw output
    script: |
      python - <<'PY'
      import subprocess, sys
      from prefect.logging import get_logger
      logger = get_logger("deployment")  # picked up by your JSON config
      p = subprocess.Popen(
          [sys.executable, "-m", "pip", "install", "-r", "requirements.txt"],
          cwd="{{ pull_code.directory }}",
          stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1
      )
      for line in p.stdout:
          <http://logger.info|logger.info>(line.rstrip())
      sys.exit(p.wait())
      PY
    directory: "{{ pull_code.directory }}"

loggers:
  deployment:
    level: INFO
    handlers: [console_json]
    propagate: false
  prefect.deployments.steps.core:
    level: INFO
    handlers: [console_json]
    propagate: false

- prefect.deployments.steps.pip_install_requirements:
    directory: "{{ pull_code.directory }}"
    requirements_file: "requirements.txt"
    stream_output: false

# my_steps/pip_json.py
import subprocess, sys, shlex
from prefect.logging import get_logger

def pip_install_json(directory=None, requirements_file="requirements.txt"):
    logger = get_logger("deployment")  # picked up by your JSON config
    cmd = [sys.executable, "-m", "pip", "install", "-r", requirements_file]
    <http://logger.info|logger.info>("pip_start cmd=%s cwd=%s", " ".join(shlex.quote(x) for x in cmd), directory or ".")
    p = subprocess.Popen(
        cmd, cwd=directory, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1
    )
    for line in p.stdout:
        <http://logger.info|logger.info>(line.rstrip())
    code = p.wait()
    if code != 0:
        raise RuntimeError(f"pip exited with {code}")
    <http://logger.info|logger.info>("pip_done returncode=%s", code)
    return {"returncode": code}

- my_steps.pip_json:pip_install_json:
    directory: "{{ pull_code.directory }}"
    requirements_file: "requirements.txt"

prefect

deployment

version: 1
disable_existing_loggers: false

formatters:
  json:
    class: prefect.logging.formatters.JsonFormatter
    format: default

handlers:
  console_json:
    class: prefect.logging.handlers.PrefectConsoleHandler
    level: DEBUG
    formatter: json
    stream: <ext://sys.stderr>
  null:
    class: logging.NullHandler
    level: WARNING

loggers:
  apprise:
    level: WARNING
    handlers: [null]      # swallow apprise logs at WARNING+
    propagate: false      # prevent reaching root

  # Optionally cover children explicitly (belt & suspenders)
  apprise.plugins:
    level: WARNING
    handlers: [null]
    propagate: false
  apprise.plugins.slack:
    level: WARNING
    handlers: [null]
    propagate: false

root:
  level: INFO
  handlers: [console_json]

version: 1
disable_existing_loggers: false

formatters:
  json:
    class: prefect.logging.formatters.JsonFormatter
    format: default

handlers:
  console_json:
    class: prefect.logging.handlers.PrefectConsoleHandler
    level: DEBUG
    formatter: json
    stream: <ext://sys.stderr>

loggers:
  apprise:
    level: WARNING       # filter out INFO at the logger
    handlers: []         # no direct handler here
    propagate: true      # WARNING+ will go to the root handler

root:
  level: INFO
  handlers: [console_json]

loggers:
  apprise:
    level: WARNING
    handlers: [console_json]  # send WARNING+ to JSON console
    propagate: false          # prevent duplicates via root

version: 1

PREFECT_LOGGING_SETTINGS_PATH

apprise

# my_logging_filters.py
import logging

class DropAppriseInfo(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        # Drop INFO (and below) from any 'apprise' logger
        return not (record.name.startswith("apprise") and record.levelno < logging.WARNING)

version: 1
disable_existing_loggers: true

formatters:
  json:
    class: prefect.logging.formatters.JsonFormatter
    format: default

filters:
  drop_apprise_info:
    (): my_logging_filters.DropAppriseInfo

handlers:
  console_json:
    class: prefect.logging.handlers.PrefectConsoleHandler
    level: DEBUG
    formatter: json
    filters: [drop_apprise_info]
    stream: <ext://sys.stderr>

  # If you send logs to the API, attach the filter there too
  api:
    class: prefect.logging.handlers.APILogHandler
    level: DEBUG
    filters: [drop_apprise_info]

loggers:
  # You can keep this minimal; the handler-level filter already does the work
  apprise:
    level: WARNING
    handlers: []
    propagate: true

root:
  level: INFO
  handlers: [console_json, api]

import logging; print(logging.getLogger('apprise').getEffectiveLevel())

# my_formatters.py
import logging, orjson

class CleanJsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "name": record.name,
            "levelname": record.levelname,
            "message": record.getMessage(),  # final, formatted text
            "created": record.created,
        }
        if record.exc_info:
            payload["exc_info"] = self.formatException(record.exc_info)
        return orjson.dumps(payload).decode()

version: 1
disable_existing_loggers: false

formatters:
  clean_json:
    (): my_formatters.CleanJsonFormatter

handlers:
  console_json:
    class: prefect.logging.handlers.PrefectConsoleHandler
    level: DEBUG
    formatter: clean_json
    stream: <ext://sys.stderr>

loggers:
  prefect:
    level: INFO
    handlers: [console_json]
    propagate: false

root:
  level: WARNING
  handlers: [console_json]

formatters:
  json:
    (): prefect.logging.formatters.JsonFormatter
    fmt: default
    dmft: null
    style: '%'

# my_formatters.py
import logging, orjson
from prefect.logging.formatters import JsonFormatter

class RobustJsonFormatter(JsonFormatter):
    def format(self, record: logging.LogRecord) -> str:
        # Try normal JSON formatting first
        try:
            return super().format(record)
        except Exception:
            # If interpolation fails, build a safe message
            try:
                safe_message = record.getMessage()
            except Exception:
                safe_message = f"{record.msg} | args={record.args!r}"
            payload = {
                "name": record.name,
                "levelname": record.levelname,
                "message": safe_message,
                "created": record.created,
            }
            if record.exc_info:
                payload["exc_info"] = self.formatException(record.exc_info)
            return orjson.dumps(payload).decode()

formatters:
  json:
    (): my_formatters.RobustJsonFormatter
    fmt: default
    dmft: null
    style: '%'

# my_formatters.py
from prefect.logging.formatters import JsonFormatter

class JsonFormatterWithMessage(JsonFormatter):
    def format(self, record):
        # Let Prefect build the dict first via parent logic,
        # then inject the formatted 'message' before serialization.
        # Easiest is to copy their logic and add message; but to avoid
        # re-implementing internals, we call super(), parse, add, re-dump.
        s = super().format(record)               # JSON string
        import orjson
        d = orjson.loads(s)                      # to dict
        try:
            d["message"] = record.getMessage()   # add formatted message
        except Exception:
            # fall back if interpolation fails
            d["message"] = f"{record.msg} | args={record.args!r}"
        return orjson.dumps(d).decode()

formatters:
  json:
    (): my_formatters.JsonFormatterWithMessage
    fmt: default
    dmft: null
    style: '%'

# my_formatters.py
import orjson
from logging import Formatter

class CleanJsonFormatter(Formatter):
    def format(self, record):
        try:
            message = record.getMessage()
        except Exception:
            message = f"{record.msg} | args={record.args!r}"
        payload = {
            "name": record.name,
            "levelname": record.levelname,
            "message": message,
            "created": record.created,
        }
        if record.exc_info:
            payload["exc_info"] = self.formatException(record.exc_info)
        return orjson.dumps(payload).decode()

formatters:
  json:
    (): my_formatters.CleanJsonFormatter

formatters:
  json:
    (): prefect.logging.formatters.JsonFormatter
    fmt: default
    dmft: null
    style: '%'

# my_formatters.py
from prefect.logging.formatters import JsonFormatter

class JsonFormatterWithMessage(JsonFormatter):
    def format(self, record):
        # Compute the final message once
        try:
            formatted = record.getMessage()
        except Exception:
            # If interpolation fails for any reason, fall back to raw msg/args
            formatted = f"{record.msg} | args={record.args!r}"
        # Call base to build the dict and serialize
        out = super().format(record)
        # out is a JSON string; quickest path is to re-serialize with the extra field
        # but to avoid double parse/serialize, override JsonFormatter to expose the dict instead.
        return out

# my_formatters.py
import orjson
from prefect.logging.formatters import JsonFormatter, _make_record_serializable

class JsonFormatterWithMessage(JsonFormatter):
    def format(self, record):
        record_dict = record.__dict__.copy()
        # Inject 'message' field
        try:
            record_dict["message"] = record.getMessage()
        except Exception:
            record_dict["message"] = f"{record.msg} | args={record.args!r}"

        # Rest mirrors Prefect’s JsonFormatter
        record_dict["severity"] = record.levelname
        if record.exc_info:
            record_dict["exc_info"] = self.formatException(record.exc_info)

        try:
            return self.serializer.dumps(record_dict).decode()
        except Exception:
            record_dict = _make_record_serializable(record_dict)
            return self.serializer.dumps(record_dict).decode()

formatters:
  json:
    (): my_formatters.JsonFormatterWithMessage
    fmt: default
    dmft: null
    style: '%'

# my_logging_filters.py
import logging

class FormatMsgFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        try:
            # Compute the final message once and store it back in msg
            record.msg = record.getMessage()
            record.args = ()  # prevent double-formatting downstream
        except Exception:
            # If anything goes wrong, leave msg/args untouched
            pass
        return True

version: 1
disable_existing_loggers: false

formatters:
  json:
    (): prefect.logging.formatters.JsonFormatter
    fmt: default
    dmft: null
    style: '%'

filters:
  format_msg:
    (): my_logging_filters.FormatMsgFilter

handlers:
  console_json:
    class: prefect.logging.handlers.PrefectConsoleHandler
    level: DEBUG
    formatter: json
    filters: [format_msg]
    stream: <ext://sys.stderr>

  api:
    class: prefect.logging.handlers.APILogHandler
    level: DEBUG
    formatter: json
    filters: [format_msg]

loggers:
  prefect:
    level: INFO
    handlers: [console_json, api]
    propagate: false

root:
  level: WARNING
  handlers: [console_json]

# my_logging_filters.py
import logging

class DropAppriseInfo(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        # Drop INFO (and below) from any 'apprise' logger
        return not (record.name.startswith("apprise") and record.levelno < logging.WARNING)

version: 1
disable_existing_loggers: true

formatters:
  json:
    (): prefect.logging.formatters.JsonFormatter
    fmt: default
    dmft: null
    style: '%'

filters:
  drop_apprise_info:
    (): my_logging_filters.DropAppriseInfo

handlers:
  console_json:
    class: prefect.logging.handlers.PrefectConsoleHandler
    level: DEBUG
    formatter: json
    filters: [drop_apprise_info]
    stream: <ext://sys.stderr>

  api:
    class: prefect.logging.handlers.APILogHandler
    level: DEBUG
    formatter: json
    filters: [drop_apprise_info]

loggers:
  # Optional: keep this minimal; handler-level filter already does the work
  apprise:
    level: INFO         # can be NOTSET/INFO; filter will still drop < WARNING
    handlers: []        # let it propagate
    propagate: true

root:
  level: INFO
  handlers: [console_json, api]

import logging
l = logging.getLogger("apprise")
print("apprise level:", l.getEffectiveLevel(), "handlers:", l.handlers, "propagate:", l.propagate)

State

python
@flow
def inner():
    ...
    return Failed(...)

@flow
def outer():
    state = inner(return_state=True)
    return state

state

inner()

outer()

text
Failed("Flow run encountered an exception: UnfinishedRun: Run is in SCHEDULED state, its result is not available.")

inner(return_state=True)

State

outer()

UnfinishedRun: Run is in SCHEDULED state

Failed

python
from prefect import flow

@flow
def inner():
    raise ValueError("inner failed")

@flow
def outer():
    inner()

python
from prefect import flow

@flow
def inner():
    raise ValueError("inner failed")

@flow
def outer():
    state = inner(return_state=True)

    if state.is_failed():
        # convert to your own exception / logic
        raise RuntimeError(f"Inner flow failed: {state.message}")

    return state.result()

python
from prefect import flow
from prefect.states import Failed

@flow
def inner():
    raise ValueError("inner failed")

@flow
def outer():
    state = inner(return_state=True)

    if state.is_failed():
        return Failed(message=f"Inner flow failed: {state.message}")

    return state.result()

return state_from_inner_flow

return Failed(message="...")

python
from prefect import flow

@flow
def inner():
    raise ValueError("inner failed")

@flow
def outer():
    state = inner(return_state=True)

    if state.is_failed():
        raise RuntimeError(state.message or "Inner flow failed")

    return state.result()

inner

outer

retries=...

retry_condition_fn

retry_condition_fn

retry_condition_fn

python
from prefect import flow, task
import httpx

class NonRetryableError(Exception):
    pass

@flow
def inner_flow(x: int):
    if x == 0:
        raise NonRetryableError("This will never succeed")
    if x < 0:
        raise ValueError("Temporary/transient example")
    return x * 2

def should_retry(task, task_run, state) -> bool:
    try:
        state.result()
    except NonRetryableError:
        return False
    except Exception:
        return True
    return False

@task(retries=3, retry_delay_seconds=10, retry_condition_fn=should_retry)
def run_inner_flow(x: int):
    return inner_flow(x)

@flow
def outer_flow(x: int):
    return run_inner_flow(x)

NonRetryableError

Failed(...)

Failed

UnfinishedRun

@flow(..., retries=...)

retry_condition_fn

inner_flow

retry_condition_fn

inner flow

Failed

successes

failures

Failed

python
from prefect import flow, task
import time

class RetryableItemError(Exception):
    pass

class NonRetryableItemError(Exception):
    pass

def build_items(job_config):
    # your job file logic
    return ["a", "b", "c"]

def process_item(item):
    if item == "b":
        raise RetryableItemError("temporary problem")
    if item == "c":
        raise NonRetryableItemError("bad input")
    return f"processed-{item}"

@task
def run_job(job_config, items_to_process=None, max_internal_retries=3):
    items = items_to_process or build_items(job_config)

    successes = []
    retryable_failures = []
    non_retryable_failures = []

    remaining = items
    attempt = 0

    while remaining and attempt < max_internal_retries:
        attempt += 1
        next_remaining = []

        for item in remaining:
            try:
                result = process_item(item)
                successes.append({"item": item, "result": result})
            except NonRetryableItemError as exc:
                non_retryable_failures.append({"item": item, "error": str(exc)})
            except RetryableItemError as exc:
                next_remaining.append(item)

        remaining = next_remaining
        if remaining:
            time.sleep(5)

    for item in remaining:
        retryable_failures.append(
            {"item": item, "error": "exhausted retries"}
        )

    return {
        "successes": successes,
        "retryable_failures": retryable_failures,
        "non_retryable_failures": non_retryable_failures,
    }

@flow
def outer_flow(job_config):
    result = run_job(job_config)

    if result["retryable_failures"] or result["non_retryable_failures"]:
        # decide whether the flow should fail or complete-with-report
        return result

    return result

But there’s an important limitation:

*Prefect task retries rerun the whole task call with the same inputs unless you explicitly persist and pass updated state yourself.*

So if you want attempt N+1 to process only the remaining failed items, you need some way to persist that intermediate payload between attempts. That usually makes *Option A* easier.

*My recommendation for your case*

Given:
- the work list is only known inside the job
- you want partial success reporting
- you want to retry only failed items

I’d use *manual retry inside a single Prefect task*.

That means:
- no nested flow-state tricks
- no fighting flow retry semantics
- easy to distinguish retryable vs non-retryable item failures
- easy to return a structured report

A good return shape would be:

```python
{
    "successes": [...],
    "retryable_failures": [...],
    "non_retryable_failures": [...],
    "total": 10,
    "succeeded_count": 7,
    "failed_count": 3
}

python
from prefect import flow

@flow
def outer_flow(job_config):
    result = run_job(job_config)

    if result["retryable_failures"] or result["non_retryable_failures"]:
        raise RuntimeError(
            f"Job completed with failures: "
            f"{len(result['retryable_failures'])} retryable, "
            f"{len(result['non_retryable_failures'])} non-retryable"
        )

    return result

build_items(job_config)

process_item(item)

run_job(...)

outer_flow(...)

parameters

parameters

json
[
  {
    "schedule": {
      "cron": "45 16 * * *",
      "timezone": "America/New_York",
      "day_or": true
    },
    "active": true,
    "max_active_runs": null,
    "catchup": true,
    "slug": "test_scheduler_via_api_2",
    "parameters": {
      "job_keys": [
        "staging_job_1"
      ]
    }
  }
]

prefect version

prefect version

Read Deployment Schedules

job_keys=["staging_job_1"]

parameters

parameters

parameters

text
Creating a deployment schedule with `parameters` via POST /deployments/{deployment_id}/schedules succeeds.
Scheduled runs correctly use the provided parameters.
However, the `parameters` field is missing from:
- the 201 create response
- the Read Deployment Schedules API response
- the Prefect UI

Read Deployment Schedules

#18701

#18701

text
I’m seeing related behavior:

- Setting schedule parameters in the UI fails
- Setting schedule parameters via API succeeds
- Scheduled runs do receive the expected parameters
- However, the parameters are not visible in the UI or in the Read Deployment Schedules API response

So the write path via API appears to work, but the read/display path seems broken.

prefect.flow-run.Failed

tag

staging

json
{
  "name": "Staging deployment failures",
  "description": "Trigger when flow runs associated with the staging tag fail",
  "enabled": true,
  "trigger": {
    "type": "event",
    "posture": "Reactive",
    "expect": [
      "prefect.flow-run.Failed"
    ],
    "match_related": {
      "prefect.resource.id": [
        "prefect.tag.staging"
      ],
      "prefect.resource.role": "tag"
    },
    "threshold": 1
  },
  "actions": [
    {
      "type": "send-notification"
    }
  ]
}

prefect.tag.<tag>

staging

json
{
  "name": "Specific staging deployment failures",
  "enabled": true,
  "trigger": {
    "type": "event",
    "posture": "Reactive",
    "expect": [
      "prefect.flow-run.Failed"
    ],
    "match_related": [
      {
        "prefect.resource.id": [
          "prefect.tag.staging"
        ],
        "prefect.resource.role": "tag"
      },
      {
        "prefect.resource.role": "deployment",
        "prefect.resource.name": [
          "my-flow/my-deployment"
        ]
      }
    ],
    "threshold": 1
  },
  "actions": [
    {
      "type": "send-notification"
    }
  ]
}

prefect.flow-run.Failed

prefect.tag.staging

curl

POST /automations