Purge partitioned_asset_key_log and restore the db clean coverage check

[steveahnahn]

Note

Builds on #70923 and contains its commit, so the diff here will shrink to a single commit once that merges. The coverage check restored below cannot pass without it: with the check working, callback is reported as uncovered until #70923 registers it.

test_no_models_missing exists to catch a metadata table being left out of airflow db clean. It has not been able to do that for some time.

It located the models package by walking up from the test file, Path(__file__).parents[2] / "airflow/models". That resolves to airflow-core/tests/airflow/models, which stopped existing when the sources moved under airflow-core/src. pkgutil.iter_modules on a missing directory yields nothing, so all_models was always empty and both assertions held trivially:

assert set(all_models) - exclusion_list.union(config_dict) == set()   # set() - x == set()
assert exclusion_list.isdisjoint(config_dict)

Confirmed against the code prior to #70923: the check passed while callback was in neither the cleanup config nor the exclusion list. That is how the table went unpurged for several releases.

What this changes

• Walk airflow.models.__path__ instead of reconstructing the path from the test's location, so the check keeps working wherever the sources live. It now discovers 52 models, against 0 before.
• Assert that models were discovered at all, so the check cannot silently pass on an empty set again.
• Register partitioned_asset_key_log for cleanup. It is the one table the restored check reports that genuinely has no way to be purged.
• Record the remaining tables in the exclusion list with the reason each is safe.

Why partitioned_asset_key_log needs purging

It carries no foreign key at all. asset_partition_dag_run is removed by ON DELETE CASCADE when its dag_run is cleaned, but the key-log rows describing that partition run are simply left behind. The only existing delete is in SchedulerJobRunner._create_dagruns_for_partitioned_asset_dags, which removes rows for stale partition runs when a rollup definition changes. That is not a retention mechanism, so the table grows without bound.

The other tables, and why they are safe

Delete rules read from the live schema after airflow db migrate, rather than from the ORM metadata, since migrations are what the deployed database actually has:

Table	Purged by
asset_partition_dag_run	ON DELETE CASCADE from dag_run
asset_watcher	ON DELETE CASCADE from trigger
dag_favorite	ON DELETE CASCADE from dag
hitl_detail	ON DELETE CASCADE from task_instance
hitl_detail_history	ON DELETE CASCADE from task_instance_history
task_inlet_asset_reference	ON DELETE CASCADE from dag
deadline_alert	ON DELETE CASCADE from serialized_dag, which follows dag
asset_state_store	ON DELETE CASCADE from asset, itself deliberately left alone
team	team configuration, not run data

Verification

• With the repair in place, removing callback from the cleanup config makes the check flag it, so it now catches the defect that motivated Fix airflow db clean never purging the callback table #70923.
• partitioned_asset_key_log cleanup verified against a real database: an old row is purged and a recent one kept.
• The full test_db_cleanup.py suite passes.

---

Was generative AI tooling used to co-author this PR?

• Yes — Claude Code (Opus 5)

Generated-by: Claude Code (Opus 5) following the guidelines

[steveahnahn]

Folded into #70923 — the restored coverage check cannot pass without the callback fix there, and the combined change is small enough to review as one.

---

Drafted-by: Claude Code (Fable 5) (no human review before posting)