<s> from datetime import datetime <EOL> import logging <EOL> from airflow . models import BaseOperator , DagRun <EOL> from airflow . utils . decorators import apply_defaults <EOL> from airflow import settings <EOL> class DagRunOrder ( object ) : <EOL> def __init__ ( self , run_id = None , payload = None ) : <EOL> self . run_id = run_id <EOL> self . payload = payload <EOL> class TriggerDagRunOperator ( BaseOperator ) : <EOL> """<STR_LIT>""" <EOL> template_fields = tuple ( ) <EOL> template_ext = tuple ( ) <EOL> ui_color = '<STR_LIT>' <EOL> @ apply_defaults <EOL> def __init__ ( <EOL> self , <EOL> trigger_dag_id , <EOL> python_callable , <EOL> * args , ** kwargs ) : <EOL> super ( TriggerDagRunOperator , self ) . __init__ ( * args , ** kwargs ) <EOL> self . python_callable = python_callable <EOL> self . trigger_dag_id = trigger_dag_id <EOL> def execute ( self , context ) : <EOL> dro = DagRunOrder ( run_id = '<STR_LIT>' + datetime . now ( ) . isoformat ( ) ) <EOL> dro = self . python_callable ( context , dro ) <EOL> if dro : <EOL> session = settings . Session ( ) <EOL> dr = DagRun ( <EOL> dag_id = self . trigger_dag_id , <EOL> run_id = dro . run_id , <EOL> conf = dro . payload , <EOL> external_trigger = True ) <EOL> logging . info ( "<STR_LIT>" . format ( dr ) ) <EOL> session . add ( dr ) <EOL> session . commit ( ) <EOL> session . close ( ) <EOL> else : <EOL> logging . info ( "<STR_LIT>" ) </s>
