Move DagBag to SDK and make it return SDK DAG objects

[uranusjr]

• Remove DagBag usages where read_dags_from_db=True
• Remove db access from DagBag
  • Everything except auto expiring check in get_dag
• Move DagBag to SDK (for use in dag parser and task runner) I give up

[uranusjr]

The Databricks operator links are a mess… Not dealing with them right now, I believe they are already broken in 3.0 so we can just keep breaking it more.

[uranusjr]

Initially I planned to remove all db access, and move the DagBag class to SDK. However, after spending almost a week trying to debug some weird Kubernetes integration test failures, I give up. This no longer moves the class.

I removed all database access in DagBag except get_dag, where it checks whether a dag has expired and auto-refreshes it. I think we no longer needs this check since dagbag uses are ephemeral in Airflow now, but we will need to fix test cases to reflect that. This will be done in a later PR.

[jason810496]

Nice! LGTM overall.

I removed all database access in DagBag except get_dag, where it checks whether a dag has expired and auto-refreshes it. I think we no longer needs this check since dagbag uses are ephemeral in Airflow now, but we will need to fix test cases to reflect that. This will be done in a later PR.

Does it mean if we want to remove db access for DagBag, we need to add another execution API route for checking whether the dag for given dag_id is expired?

The only db access in DagBag is orm_dag := DagModel.get_current, and orm_dag is used for checking is_expired.

[uranusjr]

I think we ultimately should remove the functionality. Scheduler and API server (the long running processes) already do not use this class now. The worker and dag parsing process are short-lived and this refreshing logic should not kick in at all. The main issue is to fix the tests so they reflect this, and implement necessary refreshing when a test expect it to happen automatically in-process (but now only happens in the dag processor in production circumstances).