Syar123 opened a new issue #21765:
URL: https://github.com/apache/airflow/issues/21765
### Apache Airflow version
2.1.4
### What happened
in airflow 1: we used to pass parameters like below to external dag trigger
through dag_run_obj.payload
```
def trigger_gcs_to_bq_dag(context, dag_run_obj):
dag_run_obj.payload = {
"gcs_bucket" : context['ti'].xcom_pull(key='gcs_bucket',
task_ids='parse_params'),"gcs_objects_list" :
context['ti'].xcom_pull(key='gcs_objects_list',
task_ids='filter_gcs_files'),"bq_auto_detect_schema" :
context['ti'].xcom_pull(key='bq_auto_detect_schema',
task_ids='parse_params'),"bq_schema" : context['ti'].xcom_pull(key='bq_schema',
task_ids='parse_params')
}
return dag_run_obj
trigger_gcs_to_bq_dag = TriggerDagRunOperator(
task_id='trigger_gcs_to_bq_dag',
trigger_dag_id="gcs_to_bq",
python_callable=trigger_gcs_to_bq_dag,
retries=1
)
```
in airflow 2.1.4 as python_callable is deprecated in TriggerDagRunOperator
we needed to pass parameters through 'conf' under TriggerDagRunOperator
instead of dag_run_obj.payload
```
trigger_gcs_to_bq_dag = TriggerDagRunOperator(
task_id='trigger_gcs_to_bq_dag',
trigger_dag_id="gcs_to_bq",
conf={"gcs_bucket" : "{{ ti.xcom_pull(key='gcs_bucket',
task_ids='parse_params') }}","gcs_objects_list" : "{{
ti.xcom_pull(key='gcs_objects_list', task_ids='filter_gcs_files')
}}","bq_auto_detect_schema" : "{{ ti.xcom_pull(key='bq_auto_detect_schema',
task_ids='parse_params') }}","bq_schema" : "{{ xcom_pull(key='bq_schema',
task_ids='parse_params') }}" },
retries=1
)
```
when passing through conf the parameter values are going as strings instead
of their own values (None is passing as 'None')
in the external dag ( gcs_to_bq) we are using logic like
```
def parse_params(ti, **context):
if context['dag_run'].conf:
params['gcs_bucket'] =
context['dag_run'].conf.get('gcs_bucket', None)
params['gcs_objects_list'] =
context['dag_run'].conf.get('gcs_objects_list', None)
params['bq_auto_detect_schema'] =
context['dag_run'].conf.get('bq_auto_detect_schema', None)
params['bq_schema'] =
context['dag_run'].conf.get('bq_schema', None)
else: # fail without retry
raise AirflowFailException("config missing. please trigger
the dag with proper config")
# set defaults:
if (params['bq_auto_detect_schema'] is None and
params['bq_schema'] is None):
params['bq_auto_detect_schema'] = True
if params['bq_schema'] is not None:
schema_fields=[
{
'name':f[0],
'type':f[1]
} for f in params['bq_schema']]
```
in our external dag (gcs_to_bq) [ as parsing parameters are strings we are
having issues while reading values like None, integers (0,1) and boolean values
(true)]
### What you expected to happen
I would like to pass parameters which aren't strings from source Dag to
target Dag and read them as is
### How to reproduce
_No response_
### Operating System
macOS Big Sur
### Versions of Apache Airflow Providers
v2
### Deployment
Composer
### Deployment details
I am using Google Composer 2.0.1
### Anything else
_No response_
### Are you willing to submit PR?
- [ ] Yes I am willing to submit a PR!
### Code of Conduct
- [X] I agree to follow this project's [Code of
Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
