CODEWITHSUNDEEP
pythonpython-3.xairflowdirected-acyclic-graphs
Oliver Robie
Oliver Robie

Reputation: 976

How to trigger an airflow DAG run from within a Python script?

Using apache airflow, I created some DAGS, some of which do not run on a schedule.
I'm trying to find a way that I can trigger a run for a specific DAG from within a Python script. Is this possible? How can I do?

EDIT --- The python script will be running from a different project from the project where all my DAGS are located

Upvotes: 10

Views: 26740

Answers (3)

Ramil Gataullin
Ramil Gataullin

Reputation: 141

I use this snipped for that:

import requests

def trigger_my_dag(**kwargs):
    """
    curl --location 'https://{my_airflow_url}/api/v1/dags/{my_dag_name}/dagRuns' \
        --header 'Content-Type: application/json' \
        --header 'Cache-Control: no-cache' \
        --header 'Authorization: Basic ****' \
        --data '{"conf": {}}'
    """
    url = "https://{my_airflow_url}/api/v1/dags/{my_dag_name}/dagRuns"
    data = {"conf": {}}
    headers = {
        "Authorization": "Basic ****",
        "Content-Type": "application/json",
        "Cache-Control": "no-cache",
    }
    response = requests.post(url, json=data, headers=headers)
    return response.status_code == 200

Upvotes: 0

Ihor Konovalenko
Ihor Konovalenko

Reputation: 1407

On AWS MWAA Airflow 1.10.12 I used solution based on boto3 library for Python and REST POST request:

import boto3
import requests

def TriggerAirflowDAG(mwaa_environment, dag_id):
    client = boto3.client("mwaa")
    token = client.create_cli_token(Name=mwaa_environment)
    url = "https://{0}/aws_mwaa/cli".format(token["WebServerHostname"])
    body = f"trigger_dag {dag_id}"
    headers = {
        "Authorization": "Bearer " + token["CliToken"],
        "Content-Type": "text/plain"
    }
    return requests.post(url, data=body, headers=headers)

User/role who initiates DAG run must have AmazonMWAAAirflowCliAccess policy.

Upvotes: 0

Josh
Josh

Reputation: 1976

You have a variety of options when it comes to triggering Airflow DAG runs.

Using Python

The airflow python package provides a local client you can use for triggering a dag from within a python script. For example:

from airflow.api.client.local_client import Client

c = Client(None, None)
c.trigger_dag(dag_id='test_dag_id', run_id='test_run_id', conf={})

Using the Airflow CLI

You can trigger dags in airflow manually using the Airflow CLI. More info on how to use the CLI to trigger DAGs can be found here.

Using the Airflow REST API

You can also use the Airflow REST api to trigger DAG runs. More info on that here.

The first option from within python might work for you best (it's also how I've personally done it in the past). But you could theoretically use a subprocess to interact with the CLI from python, or a library like requests to interact with the REST API from within Python.

Upvotes: 20

Related Questions