@Marvin, how can I reference a parameter to define the branch in the below deployment step in the prefect.yml file? - prefect.deployments.steps.git_clone: id: clone-step-prefect repository: https://github.com/example/prefect.git branch: '{{$prefect_repo}}' access_token: '{{ prefect.blocks.secret.github-pat}}'

In Prefect, parameters are used to pass data between tasks in a flow. However, in the context of the

prefect.yml

file, it's used for defining deployment configurations and not for passing data between tasks. The

prefect.yml

file is a configuration file, and it doesn't support parameterized values in the way you're trying to use them. The

branch

field in the

git_clone

step should be a string representing the branch name, not a parameter reference. If you want to use a variable branch name, you might consider using environment variables. You can set an environment variable in your deployment environment and then reference it in your

prefect.yml

file like this:

yaml
- prefect.deployments.steps.git_clone:
    id: clone-step-prefect
    repository: <https://github.com/example/prefect.git>
    branch: '${PREFECT_REPO}'
    access_token: '{{ prefect.blocks.secret.github-pat}}'

In this case,

PREFECT_REPO

should be an environment variable set in your deployment environment. Please note that the syntax for referencing environment variables may vary depending on your deployment environment. Remember to replace

PREFECT_REPO

with the actual name of your environment variable.

@Marvin, could this also be a job variable?

Yes, you can use a job variable in the same way as an environment variable. The exact syntax might depend on your CI/CD system, but in general, you can reference job variables in your

prefect.yml

file like this:

yaml
- prefect.deployments.steps.git_clone:
    id: clone-step-prefect
    repository: <https://github.com/example/prefect.git>
    branch: '${CI_JOB_VARIABLE}'
    access_token: '{{ prefect.blocks.secret.github-pat}}'

In this case,

CI_JOB_VARIABLE

should be a variable set in your CI/CD job. Please note that the syntax for referencing job variables may vary depending on your CI/CD system. Remember to replace

CI_JOB_VARIABLE

with the actual name of your job variable.

@Marvin, what's best practice to trigger one deployment based on the outcome of another flow? Webhooks? Events?

In Prefect, you can use Automations and Webhooks to trigger one deployment based on the outcome of another flow. Automations API allows you to customize trigger and action policies based on arbitrary events. Triggers can be configured not only in reaction to events, but also proactively, to trigger in the absence of an event you expect to see. You can find more details here. Webhooks in Prefect allow you to send notifications to external services when certain events occur. You can configure webhooks in your workspace and toggle them on or off as needed. More information can be found here. Events in Prefect are notifications of a change. They power several features in Prefect Cloud, including flow run logs, audit logs, and automations. Events can represent API calls, state transitions, or changes in your execution environment or infrastructure. More about events can be found here.

@Marvin, i am thinking of using the automation API to create an automation that assesses the returned value from deployment X to infer whether deployment Y should be triggered. Can you give me an example of how I can achieve this using the create automation API?

Sure, here's a high-level example of how you might use the Automations API in Prefect to trigger one deployment based on the outcome of another: 1. Define your flows: Let's say you have two flows,

flow_X

and

flow_Y

.

flow_X

is the initial deployment and

flow_Y

is the deployment you want to trigger based on the outcome of

flow_X

. 2. Define your trigger condition: This is the condition that will determine whether

flow_Y

should be triggered. For example, you might want to trigger

flow_Y

if

flow_X

completes successfully. In this case, your trigger condition could be the state of

flow_X

changing to

Success

. 3. Define your action: This is what will happen when the trigger condition is met. In this case, you want to start

flow_Y

, so your action would be to start a new run of

flow_Y

. 4. Use the Automations API to create an automation with your trigger condition and action: The Automations API allows you to programmatically create automations. You would use this API to create an automation that starts a new run of

flow_Y

whenever

flow_X

completes successfully. Please note that the exact code to achieve this will depend on your specific use case and the details of your flows. Also, Automations are only available in Prefect Cloud. For more detailed information and examples, I recommend checking out the Prefect Automations documentation.

@Marvin { "match": { "prefect.resource.id": "prefect.flow-run.*" }, "match_related": { "prefect.resource.id": "prefect.deployment.70cb25fe-e33d-4f96-b1bc-74aa4e50b761", "prefect.resource.role": "deployment" }, "for_each": [ "prefect.resource.id" ], "after": [], "expect": [ "prefect.flow-run.Failed" ], "posture": "Reactive", "threshold": 2, "within": 10 } Can you fill this out for me using my example. The flow is any run within deployment abc. This flow returns a boolean through variable: "run_dbt" if this variable = True then deployment xyz should trigger a run. if variable == False then nothing should happen.

Based on your example, here's how you might fill out the automation configuration:

json
{
  "match": {
    "prefect.resource.id": "prefect.flow-run.*"
  },
  "match_related": {
    "prefect.resource.id": "prefect.deployment.abc",
    "prefect.resource.role": "deployment"
  },
  "for_each": [
    "prefect.resource.id"
  ],
  "after": [],
  "expect": [
    {
      "prefect.flow-run.Success": {
        "run_dbt": true
      }
    }
  ],
  "posture": "Reactive",
  "threshold": 1,
  "within": 10,
  "actions": [
    {
      "type": "prefect.run-flow",
      "args": {
        "flow_id": "xyz"
      }
    }
  ]
}

This configuration will match any flow run within the deployment with ID

abc

. If a flow run completes successfully and the

run_dbt

variable is

true

, it will trigger a run of the flow with ID

xyz

. Please note that this is a high-level example and the exact configuration will depend on your specific use case and the details of your flows. Also, Automations are only available in Prefect Cloud. For more detailed information and examples, I recommend checking out the Prefect Automations documentation.

@Marvin Thanks. What is the difference between "match" and "match related"?

In the context of Prefect Automations, "match" and "match_related" are used to specify the conditions for triggering an automation. - "match": This is used to specify the primary condition for triggering the automation. It defines the specific events or states that should trigger the automation. For example, if you want to trigger an automation when a specific flow run fails, you would use "match" to specify that condition. - "match_related": This is used to specify secondary or related conditions for triggering the automation. It defines additional conditions that must be met in conjunction with the primary "match" condition. For example, if you want to trigger an automation when a specific flow run fails, but only if that flow run is part of a specific deployment, you would use "match_related" to specify the deployment condition. In other words, "match" is used to define the main event that triggers the automation, while "match_related" is used to define additional, related conditions that must also be met.

@Marvin, abc and xyz should be replaced with the ID right?

Yes, that's correct. In the automation configuration,

abc

and

xyz

are placeholders for the IDs of your deployments or flows. You should replace

abc

with the ID of the deployment that contains the flow you're monitoring, and replace

xyz

with the ID of the flow you want to trigger when the condition is met. These IDs are unique identifiers assigned by Prefect to each deployment and flow. You can usually find them in the Prefect UI or via the Prefect API.

@Marvin { "match": { "prefect.resource.id": "prefect.flow-run.*" }, "match_related": { "prefect.resource.id": "prefect.deployment.317ea6fe-1f27-4cb0-902d-210138d6bb7f", "prefect.resource.role": "deployment" }, "after": [], "expect": [ { "prefect.flow-run.Success": { "run_dbt": true } } ], "for_each": [ "prefect.resource.id" ], "posture": "Reactive", "threshold": 1, "within": 0 } With the above trigger JSON, i get an error that "something went wrong"

I'm sorry to hear that you're encountering an error. The error message "something went wrong" is quite generic and it's hard to determine the exact cause without more specific details. However, here are a few things you might want to check: 1. IDs: Make sure that the IDs you're using for your deployment and flow are correct. You can usually find these in the Prefect UI or via the Prefect API. 2. Flow State: The automation is configured to trigger when a flow run within the specified deployment succeeds and the

run_dbt

variable is

true

. Make sure that your flow is actually reaching a success state and that the

run_dbt

variable is being set correctly. 3. Automation Configuration: Check your automation configuration for any syntax errors or missing fields. The configuration you've posted looks correct at a glance, but there might be some subtle issue that's causing the error. 4. Prefect Cloud: Automations are only available in Prefect Cloud. If you're trying to use them with a local Prefect server, you'll encounter errors. If you've checked all of these things and you're still encountering the error, I would recommend reaching out to Prefect support or the Prefect community for more specific help. They might be able to provide more insight into what's going wrong.

@Marvin, are you confident this works? "expect": [ { "prefect.flow-run.Success": { "run_dbt": true } } ],

Based on the information from the Prefect documentation, it seems that the

expect

field in the automation configuration is used to specify the conditions under which the automation should be triggered. However, the documentation does not provide specific examples or details about using the

expect

field to check the value of a variable like

run_dbt

. The

expect

field is typically used to specify a state change (like

prefect.flow-run.Success

) that should trigger the automation. It's possible that checking the value of a variable like

run_dbt

might not be supported directly in the automation configuration. If you need to trigger an automation based on the value of a variable, you might need to handle this within your flow logic. For example, you could add a task to your flow that checks the value of

run_dbt

and raises an error if it's not

true

. This would cause the flow to fail, which you could then use as a trigger condition in your automation. I would recommend reaching out to Prefect support or the Prefect community for more specific guidance on this. They might be able to provide more detailed information or suggest a workaround.

@Marvin it seems that the result of my flow is as follows: RESULT Result with value

False

persisted to Prefect. Nowhere does it show the variable name. Could that be the issue? Can you please update the expect statement to account for this result?

Based on your flow result, it seems like the flow is returning a boolean value (

True

or

False

) directly, rather than a variable like

run_dbt

. In this case, you might want to adjust your

expect

statement to look for a

Success

state with a result of

True

. Here's how you might do that:

json
"expect": [
  {
    "prefect.flow-run.Success": {
      "result": true
    }
  }
]

This configuration will trigger the automation when a flow run within the specified deployment succeeds and the result of the flow run is

True

. Please note that this is a high-level example and the exact configuration will depend on your specific use case and the details of your flows. Also, Automations are only available in Prefect Cloud. For more detailed information and examples, I recommend checking out the Prefect Automations documentation.

@Marvin, what is best practice to implement dependencies between tasks and flows? Occassinally my subflows duplicate within a run without an apparent reason.

To implement dependencies between tasks in Prefect, it's best to define tasks as functions that can take inputs, perform work, and return an output. Dependencies between tasks are typically established by passing the output of one task as the input to another. This makes it easy to encapsulate elements of your workflow logic in observable units that can be reused across flows and subflows (Prefect Docs - Tasks). Furthermore, for dependencies between flows, Prefect allows calling tasks or subflows within your workflows, including passing results from other tasks to your subflow. A common question often arises as to when one should use a subflow instead of a task. Generally, the recommendation is to write tasks for discrete units of work that can be observed and reused, suggesting that tasks should be the default building blocks for flow construction. For full workflows that you might want to run as part of a larger process, subflows can be useful (Prefect Docs - Flows). Regarding the issue of subflows duplicating within a run, there is no specific information provided by the initial search. However, duplication of subflows could potentially be caused by issues related to idempotency, incorrect flow configuration, or possibly race conditions where the flow might be triggered multiple times unintentionally. It might be helpful to check the scheduling and triggering mechanisms of your flows to make sure that they aren't being inadvertently started more than once. If you are still encountering issues with subflow duplication and the checks mentioned above do not resolve the problem, I would recommend providing specific examples or additional context which could be useful in diagnosing the issue further. If it seems like a bug or a complex use-case that isn't addressed by the Prefect documentation, you may consider reaching out to the Prefect community or opening an issue on the Prefect GitHub repository for more targeted help. It's crucial to note that the search for issues related to "subflows duplication in runs" on the Prefect GitHub repository returned no results, which might indicate that this is not a common or known issue within the community. However, it might still be helpful to engage with the community or explore the Prefect GitHub repository further to see if others have encountered similar issues and what solutions they might have found.

@Marvin, if a flow fails, and I want to apply a retry using e.g. retries=3, can I override a parameter?

In Prefect 2.0, flow retries can be configured similarly to task retries, using

retries

and

retry_delay_seconds

parameters (Prefect Release Notes). If both a flow and its tasks have retries configured, tasks within the flow will retry up to their specified task retry limit for each flow run. For instance, a flow with 2 retries would allow for up to 3 total runs (including the initial attempt), and if there's a task within this flow with 3 retries, this task could be attempted up to 4 times per flow run, totaling up to 12 attempts. However, the information about overriding parameters during flow retries specifically is not provided in the current excerpts. Based on Prefect's general behavior, parameters for task or flow runs are usually set when initiating the run and aren't typically designed to be overridden during retries, because retries are intended to re-execute the task or flow with the same conditions to check for transient errors. If you require a different behavior with retries and want to override parameters, you might need to implement a custom retry logic within your flow, which captures the failure and initiates a new flow run with different parameters. This custom logic could be implemented manually in your workflow definition or potentially through some advanced features of Prefect. If you need more detailed information or specific guidance on how to implement parameter override during retries—or if this feature has been introduced in a more recent Prefect release—I would recommend checking the latest Prefect documentation or reaching out to the Prefect community for the most current and in-depth practices.

@Marvin, I currently have in my prefect.yml file a job variable that is used to define the branch of prefect to clone: - prefect.deployments.steps.git_clone: branch: '{{$prefect_repo}}' I want to change this to a parameter. how do I do that?