Let an auth manager filter dags in SQL instead of materializing every id

[1fanwang]

Closes: #71309

Why

get_authorized_dag_ids is where every list endpoint starts, and its default implementation reads
the whole dag table before any manager is consulted, groups the rows by team, then calls
filter_authorized_dag_ids once per team. A request for 50 dags loads every dag id in the
deployment, so the cost scales with the deployment rather than the page.

That is core's own path, not one provider's. FabAuthManager overrides it and still reads every
row. Keycloak inherits it and then calls out per dag, which
#69041 and
#61686 report as ten and twenty-five second
pages.

It also makes the documented answer expensive.
#23638 asked for dag permissions by tag or owner
rather than by name, and Airflow 3 answers that by writing an auth manager over
get_authorized_dag_ids. Tags, bundles and teams are all rows in this database, and
get_db_manager lets a manager add its own. The set[str] return type is what stops any of them
being answered as a query.

What changed

BaseAuthManager gains get_authorized_dag_ids_select, returning a select of dag ids or None.
None is the default and keeps today's behaviour, so managers backed by an external policy service
(Keycloak, Amazon Verified Permissions) are unaffected.

A returned select is applied as dag_id IN (subquery). The dependency graph services and the
run-state counts endpoint still need ids in memory, and still get them: PermittedDagFilter
materializes on first read.

A select replaces get_authorized_dag_ids outright, including its per-team grouping, so a
multi-team manager scopes the select itself or returns None and keeps the fan-out.

FabAuthManager implements it with the grant query it already knows how to write.

Testing Done

End to end against a running API server. A 40-line manager authorizing by tag, no FAB. Six
dags seeded, three tagged team-a.

$ curl -H "Authorization: Bearer $TOKEN" localhost:28080/api/v2/dags
total: 3
  dag_01 [team-a]
  dag_02 [team-a]
  dag_05 [team-a]

$ curl -H "Authorization: Bearer $TOKEN" 'localhost:28080/api/v2/dags?limit=2'
limit=2 -> ['dag_01', 'dag_02'] total 3

The statement that server executed, captured with a before_cursor_execute listener:

... LEFT OUTER JOIN dag_run ON dag_run.id = mrq.max_dag_run_id
WHERE dag.dag_id IN (SELECT dag_tag.dag_id FROM dag_tag WHERE dag_tag.name = ?)
  AND dag.is_stale != 1
ORDER BY dag.dag_id ASC LIMIT ? OFFSET ?

A subquery rather than bind parameters, LIMIT in the same statement, and the same subquery
reaching the latest-run join.

Measured. The same shape at scale. MySQL 8, 41,606 dags with an environment tag and a team
tag. Page of 50, median of 15 rounds.

Tag the manager authorizes on	Matching dags	Materialize (today)	Subquery
Two team tags	418	5.4 ms	1.4 ms
An environment tag	37,446	316.1 ms	0.8 ms

Measured, FAB. Same database at 41,606 dags, 1,610 roles, 66,529 per-dag edit grants.

Grant	Materialize (today)	Subquery
Per-dag edit on 228 dags	3.7 ms	2.3 ms
Authorized on all 41,606 dags	351.8 ms	0.6 ms

The expensive row in both is the ordinary one: a tag most dags carry, or a role with global
can_read, which is Viewer and up.

Regression tests. Eight, covering the None default, a select reaching the SQL as a subquery,
an empty select honoured as "nothing is permitted", a subclass filter inheriting it, a team-scoped
select, and the id set materializing once and only when read. Reverting the fix:

$ pytest airflow-core/tests/unit/api_fastapi/core_api/test_security.py \
    -k PermittedDagFilterSubquery -q
E   TypeError: Boolean value of this clause is not defined
4 failed, 2 passed, 111 deselected

Restored:

$ pytest airflow-core/tests/unit/api_fastapi/core_api/test_security.py \
    -k PermittedDagFilterSubquery -q
8 passed, 111 deselected

$ pytest airflow-core/tests/unit/api_fastapi/auth/managers/test_base_auth_manager.py -q
61 passed

$ pytest airflow-core/tests/unit/api_fastapi/common -q
162 passed, 6 skipped

[vincbeck]

I understand the motivation but I am against introducing a new method in base_auth_manager class that is only applicable to Fab. Fab auth manager is the only auth manager using the Db to manage permissions, so get_authorized_dag_ids_select is Fab specific. Ideally, we should get rid of Fab, and not add Fab specific features in Airflow core.

[1fanwang]

I understand the motivation but I am against introducing a new method in base_auth_manager class that is only applicable to Fab. Fab auth manager is the only auth manager using the Db to manage permissions, so get_authorized_dag_ids_select is Fab specific. Ideally, we should get rid of Fab, and not add Fab specific features in Airflow core.

Agreed, and thanks for the context on the direction. @vincbeck

I checked and confirmed there's also no easy way to update this in FAB only. PermittedDagFilter is bound to get_authorized_dag_ids in core, and the set[str] return is what forces the materialization, so there's no provider-side seam. Not worth a core hook for the one manager that's on the path to deprecation

Closing.