Deleted DAGs not removed from ab_view_menu table and show up in permissions (Airflow 2.7.3)

[sandeyshc]

Apache Airflow version:
Other Airflow 2 version (please specify below)

If "Other Airflow 2 version" selected, which one?
2.7.3

What happened?
When a DAG is deleted, its entry is not removed from the ab_view_menu table in the Airflow metadata database. As a result, even deleted DAG IDs continue to appear in the Airflow UI when assigning permissions to roles. This causes confusion and can lead to accidental permission assignments to non-existent DAGs.

What you think should happen instead?
When a DAG is deleted, the corresponding entry for that dag_id should also be removed from the ab_view_menu table so that only active DAGs are available for permission assignment in the Airflow UI.

How to reproduce

1. Delete a DAG from your Airflow instance.
2. Check the ab_view_menu table in the metadata database; the old DAG ID remains.
3. Go to the Airflow UI and attempt to assign permissions to a role; deleted DAG IDs are still listed.

Operating System
Please specify your operating system details (e.g., output of cat /etc/os-release).
linux

Versions of Apache Airflow Providers
Please specify provider versions if relevant, e.g. pip freeze | grep apache-airflow-providers.

Deployment
Please select your deployment type (e.g., Docker-Compose, Helm Chart, etc.).
helm deployment

Deployment details
Add any extra details about your deployment (e.g., version of Docker, Kubernetes, etc.).

Anything else?
Issue occurs consistently

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[HsiuChuanHsu]

@sandeyshc
After digging into this issue, I think the problem’s with the delete_dag.py function. It’s not fully clearing out all the related resources. You can review the function here:

airflow/airflow-core/src/airflow/api/common/delete_dag.py

Lines 42 to 91 in 30a23e5

	@provide_session
	def delete_dag(dag_id: str, keep_records_in_log: bool = True, session: Session = NEW_SESSION) -> int:
	"""
	Delete a DAG by a dag_id.
	:param dag_id: the dag_id of the DAG to delete
	:param keep_records_in_log: whether keep records of the given dag_id
	in the Log table in the backend database (for reasons like auditing).
	The default value is True.
	:param session: session used
	:return count of deleted dags
	"""
	log.info("Deleting DAG: %s", dag_id)
	running_tis = session.scalar(
	select(models.TaskInstance.state)
	.where(models.TaskInstance.dag_id == dag_id)
	.where(models.TaskInstance.state == TaskInstanceState.RUNNING)
	.limit(1)
	)
	if running_tis:
	raise AirflowException("TaskInstances still running")
	dag = session.scalar(select(DagModel).where(DagModel.dag_id == dag_id).limit(1))
	if dag is None:
	raise DagNotFound(f"Dag id {dag_id} not found")
	# To ensure the TaskInstance and DagRun model is deleted before
	# each of the model DagVersion and BackFill respectively.
	models_for_deletion = [TaskInstance, DagRun] + [
	model for model in get_sqla_model_classes() if model.__name__ not in ["TaskInstance", "DagRun"]
	]
	count = 0
	for model in models_for_deletion:
	if hasattr(model, "dag_id") and (not keep_records_in_log or model.__name__ != "Log"):
	count += session.execute(
	delete(model).where(model.dag_id == dag_id).execution_options(synchronize_session="fetch")
	).rowcount
	# Delete entries in Import Errors table for a deleted DAG
	# This handles the case when the dag_id is changed in the file
	session.execute(
	delete(ParseImportError)
	.where(
	ParseImportError.filename == dag.relative_fileloc,
	ParseImportError.bundle_name == dag.bundle_name,
	)
	.execution_options(synchronize_session="fetch")
	)
	return count

I’d suggest tossing in a new function to delete_dag.py to handle the extra cleanup.
Reference code:

airflow/devel-common/src/tests_common/test_utils/db.py

Lines 358 to 393 in e00777a

	def clear_dag_specific_permissions():
	if "FabAuthManager" not in conf.get("core", "auth_manager"):
	return
	try:
	from airflow.providers.fab.auth_manager.models import Permission, Resource, assoc_permission_role
	except ImportError:
	# Handle Pre-airflow 2.9 case where FAB was part of the core airflow
	from airflow.providers.fab.auth.managers.fab.models import (
	Permission,
	Resource,
	assoc_permission_role,
	)
	except RuntimeError as e:
	# Handle case where FAB provider is not even usable
	if "needs Apache Airflow 2.9.0" in str(e):
	from airflow.providers.fab.auth.managers.fab.models import (
	Permission,
	Resource,
	assoc_permission_role,
	)
	else:
	raise
	with create_session() as session:
	dag_resources = session.query(Resource).filter(Resource.name.like(f"{RESOURCE_DAG_PREFIX}%")).all()
	dag_resource_ids = [d.id for d in dag_resources]
	dag_permissions = session.query(Permission).filter(Permission.resource_id.in_(dag_resource_ids)).all()
	dag_permission_ids = [d.id for d in dag_permissions]
	session.query(assoc_permission_role).filter(
	assoc_permission_role.c.permission_view_id.in_(dag_permission_ids)
	).delete(synchronize_session=False)
	session.query(Permission).filter(Permission.resource_id.in_(dag_resource_ids)).delete(
	synchronize_session=False
	)
	session.query(Resource).filter(Resource.id.in_(dag_resource_ids)).delete(synchronize_session=False)

Let me know what you think!

[HsiuChuanHsu]

May I be assigned to this issue? Thanks!