()
| 218 | |
| 219 | @dag(schedule=None, start_date=DEFAULT_DATE, catchup=False, default_args=default_args) |
| 220 | def dag_decomposed(): |
| 221 | nonlocal tasks_list |
| 222 | tasks = PipelineTasksGroup( |
| 223 | "pipeline_dag_decomposed", |
| 224 | local_data_folder=get_test_storage_root(), |
| 225 | wipe_local_data=False, |
| 226 | ) |
| 227 | |
| 228 | # set duckdb to be outside of pipeline folder which is dropped on each task |
| 229 | pipeline_dag_decomposed = dlt.pipeline( |
| 230 | pipeline_name="pipeline_dag_decomposed", |
| 231 | dataset_name="mock_data_" + uniq_id(), |
| 232 | destination="duckdb", |
| 233 | ) |
| 234 | tasks_list = tasks.add_run( |
| 235 | pipeline_dag_decomposed, |
| 236 | mock_data_source(), |
| 237 | decompose="serialize", |
| 238 | trigger_rule="all_done", |
| 239 | retries=0, |
| 240 | ) |
| 241 | |
| 242 | dag_def = dag_decomposed() |
| 243 | assert len(tasks_list) == 3 |
no test coverage detected