Add variable to disable removing SQL source files for ingestion workflows

[AetherUnbound]

Description

The iNaturalist DAG uses the ingestion workflow's sql_rm_source_data_after_ingesting parameter to determine whether it should remove or retain the source files used for ingestion:

openverse/catalog/dags/providers/provider_dag_factory.py

Lines 422 to 430 in 2cffcb9

	"sql_rm_source_data_after_ingesting": Param(
	default=True,
	type="boolean",
	description=(
	"Whether to delete source data from airflow and DB once ingestion"
	" is complete. This is used to support data quality testing in"
	" SQL-only DAGs like iNaturalist."
	),
	),

While this is useful for specific runs, the iNaturalist DAG is scheduled, which means that the default run that gets kicked off locally when the DAG is enabled will remove the source files. Since these can be quite large, it's tedious and time consuming to have to manage triggering each run with the sql_rm_source_data_after_ingesting box unchecked.

We should also have an Airflow Variable which will also determine whether the files should be removed or not. The value could potentially be SQL_RM_SOURCE_DATA_AFTER_INGESTION, meaning the name of the variable added to our env.template file would be AIRFLOW_VAR_SQL_RM_SOURCE_DATA_AFTER_INGESTION. This should be True by default in the code, but False as defined in the env.template file so by default, local runs will save source files.

We will also need to update the short-circuit task for skipping this to include checking this variable as well:

openverse/catalog/dags/providers/provider_api_scripts/inaturalist.py

Lines 347 to 355 in dca0110

	check_drop_parameter = ShortCircuitOperator(
	task_id="check_drop_parameter",
	doc_md="Skip post-ingestion if NOT sql_rm_source_data_after_ingesting.",
	op_args=["{{ params.sql_rm_source_data_after_ingesting }}"],
	python_callable=(lambda x: x),
	trigger_rule=TriggerRule.NONE_SKIPPED,
	# just skip the drop steps, not the final reporting step in the dag
	ignore_downstream_trigger_rules=False,
	)

The check should be such that if either the param or the Airflow Variable are set to False, the files are retained. We should be able to use the {{ var.json.<variable_name> }} syntax for templating this into the op_args similar to the param.

Additional context

See #3846 for the impetus

[Pqformeln]

Hi,

I would like to work on this issue!

[AetherUnbound]

Fantastic, welcome @Pqformeln! I'll assign the issue to you 😄 Please check out our welcome and quickstart documentation pages, and if you have any questions about this issue feel free to leave them here!

[madewithkode]

I'd love to take on this @AetherUnbound @obulat

[AetherUnbound]

I'll assign you @madewithkode! Let us know if you have any questions about it 😄

[madewithkode]

Thanks @AetherUnbound. I think I have basically done what the task requires(not exactly sure how to test though). Can I go ahead and sort of open a draft MR so someone can take a look and see if my understanding of the requirements is correct?

[AetherUnbound]

Absolutely!

[madewithkode]

Hi @AetherUnbound, I have now made a draft PR here so we can confirm if I'm in the right direction and also provide ways to test.