Hi all, i wanna create a api, that use to call to execute a task in prefect. How to do that? thanks all

@Anh Nguyen do you want to call an external API from a Prefect task, or do you want to kick off a Prefect flow from some API?

yes! that's right. thanks you. Could you tell me to do that ?

1. do you want to call an external API from a Prefect task? 2. do you want to kick off a Prefect flow from some API? which one is your use case, @Anh Nguyen?

yes i want to kick off a Prefect flow from some API

you can use the following GraphQL mutation:

import requests

query = """
 mutation {
  create_flow_run(input: { flow_id: "your-flow-id" }) {
    id
  }
}
"""

url = "<https://api.prefect.io>"
response = <http://requests.post|requests.post>(
    url, json={"query": query}, headers={"authorization": "Bearer XXX"}
)
print(response.status_code)
print(response.text)

You need to set an API key instead of XXX and your flow ID.

yes! In the request, can i pass parameter to flow and using the parameter in processing flow ?

Perhaps this

create_flow_run

task is what you’re looking for? Can you describe your use case?

from prefect.tasks.prefect import create_flow_run

with Flow("FLOW_NAME",) as flow:
    extract_load_id = create_flow_run(
        flow_name="your-flow-name",
        project_name="PROJECT_NAME",
        parameters={"key": "value"},
    )

i want to use the value is got from request that will be insert to db

from prefect.tasks.prefect import create_flow_run

with Flow("FLOW_NAME",) as flow:
    extract_load_id = create_flow_run(
        flow_name="your-flow-name",
        project_name="PROJECT_NAME",
        parameters={"key": "value"},
    )

I think you can start by following this getting started tutorial to get familiar with Prefect in general: https://docs.prefect.io/core/getting_started/quick-start.html Then, you can build your logic to insert data into a database e.g. via a custom function and decorate it with @task

yes, but i cant get value of parameters! how to get value of parameters in flow ?

@Anh Nguyen Like so:

from prefect import task, Flow, Parameter


@task(log_stdout=True)
def hello_world(name):
    print(f"hello {name}")


with Flow("mini.example") as flow:
    name = Parameter("name", default="Anh")
    hw = hello_world(name)

if __name__ == "__main__":
    flow.run()

sorry if my question is not clear. Step 1: Kill off the prefect flow from api and passing [key:value] Step 2: In the prefect flow i want to get value in [key:value]. This value is use to insert to db.