multitenany_snippet.py

VERSION='v1.0.3'

# DAG Factory
factory = DAGFactory(config.environment)

process_invoices_dag = factory.create('ProcessInvoices', config.process_invoices).build()
process_messages_dag = factory.create('ProcessMessages', config.process_messages).build()


def create_main_dag(dag_id, org_config):
  with DAG(dag_id=f'main_{VERSION}',
           default_args=default_args,
           schedule_interval=config.controller['schedule'],
           catchup=False,
           template_searchpath=['/home/airflow/gcs/dags'],
      ) as dag:

      # Reconfigure the dag object as a subdag 
      # under our main dag
      process_invoices = SubDagOperator(
          task_id='process_invoices',
          subdag=process_invoices_dag,
      )

      process_messages = SubDagOperator(
          task_id='process_messages',
          subdag=process_messages_dag, 'process_messages')
      )
      
      return dag

    
# Loop over dynamic config and auto generate 
# dags for each client. 
for client_config in client_configs: 
    parsed_client_config = config.parse_config(client_config)
    main_dag_id = f"{client_config['identifier']}_main_{VERSION}"
    globals()[main_dag_id] = create_main_dag(
        main_dag_id,
        parsed_client_config
    )