<s> from airflow . exceptions import AirflowException <EOL> from airflow . models import BaseOperator , Pool <EOL> from airflow . utils . decorators import apply_defaults <EOL> from airflow . utils . db import provide_session <EOL> from airflow . executors import DEFAULT_EXECUTOR <EOL> class SubDagOperator ( BaseOperator ) : <EOL> template_fields = tuple ( ) <EOL> ui_color = '<STR_LIT>' <EOL> ui_fgcolor = '<STR_LIT>' <EOL> @ provide_session <EOL> @ apply_defaults <EOL> def __init__ ( <EOL> self , <EOL> subdag , <EOL> executor = DEFAULT_EXECUTOR , <EOL> * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if '<STR_LIT>' not in kwargs : <EOL> raise AirflowException ( "<STR_LIT>" ) <EOL> dag = kwargs [ '<STR_LIT>' ] <EOL> session = kwargs . pop ( '<STR_LIT>' ) <EOL> super ( SubDagOperator , self ) . __init__ ( * args , ** kwargs ) <EOL> if dag . dag_id + '<STR_LIT:.>' + kwargs [ '<STR_LIT>' ] != subdag . dag_id : <EOL> raise AirflowException ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" . format ( <EOL> d = dag . dag_id , t = kwargs [ '<STR_LIT>' ] , rcvd = subdag . dag_id ) ) <EOL> if self . pool : <EOL> pool = ( <EOL> session <EOL> . query ( Pool ) <EOL> . filter ( Pool . slots == <NUM_LIT:1> ) <EOL> . filter ( Pool . pool == self . pool ) <EOL> . first ( ) <EOL> ) <EOL> conflicts = [ t for t in subdag . tasks if t . pool == self . pool ] <EOL> if pool and any ( t . pool == self . pool for t in subdag . tasks ) : <EOL> raise AirflowException ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' . format ( <EOL> sd = self . task_id , <EOL> plural = len ( conflicts ) > <NUM_LIT:1> , <EOL> t = '<STR_LIT:U+002CU+0020>' . join ( t . task_id for t in conflicts ) , <EOL> p = self . pool <EOL> ) <EOL> ) <EOL> self . subdag = subdag <EOL> self . executor = executor <EOL> def execute ( self , context ) : <EOL> ed = context [ '<STR_LIT>' ] <EOL> self . subdag . run ( <EOL> start_date = ed , end_date = ed , donot_pickle = True , <EOL> executor = self . executor ) </s>
