@Marvin Hi Marvin! I was wondering if there was a way to automate the creation of a prefect flow run based on the results from an API request. For example, I need to request a new authentication token whenever the current token is set to expire. That time limit is contained within a JSON object, which is the result of a prior flow run. This can be weeks apart.

It seems the example I fetched does not directly address your query about automating flow runs based on API results. Let me refine my search and try again. Please hold on for a moment. Based on the information I've obtained, you might consider taking advantage of Prefect's Automations feature to programmatically create a trigger that will run a deployment of a flow based on certain conditions met within the trigger criteria. One useful example for your case could be:

python
import requests

api_url = "<https://api.prefect.io/{workspace_id}/automations/>"

data = {
    "name": "Event Driven Redeploy",
    "description": "Programmatically created an automation to redeploy a flow based on an event",
    "enabled": "true",
    "trigger": {
        "after": ["prefect.flow-run.Running"],
        "expect": ["prefect.flow-run.Completed"],
        "for_each": ["prefect.resource.id"],
        "posture": "Proactive",
        "threshold": 1,
        "within": "60"  # time interval to check for the event
    },
    "actions": [
        {
            "type": "run-deployment",
            "source": "selected",
            "deployment_id": "YOUR-DEPLOYMENT-ID",
            "parameters": {}  # any parameters you want to pass to the deployment
        }
    ],
    "owner_resource": "string"
}

headers = {"Authorization": f"Bearer {PREFECT_API_KEY}"}
response = <http://requests.post|requests.post>(api_url, headers=headers, json=data)
print(response.json())

Replace the

"YOUR-DEPLOYMENT-ID"

,

"prefect.flow-run.Running"

,

"prefect.flow-run.Completed"

, and time within accordingly with your specific use case. For authentication, use your

PREFECT_API_KEY

. This code will create an automation that checks if a flow run has completed within a certain time frame. Additionally, you may write Python code to check the results from an API request and then decide whether to initiate a Prefect flow run depending on the token expiry information. Prefect doesn't have built-in mechanisms for these kinds of long-interval checks, so you might need to use an external scheduler or a continuously running process that periodically checks the token's status. Also, note the Async Basic Example which showcases the structure of how your flow might look with async capabilities, should your API interaction require it. However, it does not specifically detail the triggering based on an API request. Remember to tailor your implementation to your specific requirements, considering the different intervals at which you'll need to renew the token.

Hey Nick! Just my 2 cents but I think you could use `run_deployment` here. It accepts a

schedule_time

to schedule a flow run (which you can set depending on the token expiry date)

Ahh this looks like it could do the trick! Will take a look and test it out, thank you for the quick reply Bianca. And thank you as always @Marvin!

You're welcome! It's what I'm here for, after all. If you run into any issues or have further questions while implementing the solution, don't hesitate to reach out. Good luck with your automations, and may your flows run (more or less) as expected.