Pass dag_id to Dag bundle construction on task execution

[ykuc]

Pass dag_id to Dag bundle construction when a worker runs a task.

Custom Dag bundles can now read self.dag_id on the bundle instance before initialize() runs. That lets bundle authors tailor setup per Dag (for example, per-Dag sparse checkout in Git) without changing the initialize() signature.

Changes:

Add optional dag_id to BaseDagBundle.init (None by default).
Extend DagBundlesManager.get_bundle() with dag_id and pass it into the bundle constructor.
Pass what.ti.dag_id from task_runner.parse() on the worker task execution path.
Document self.dag_id in the custom Dag bundles section of dag-bundles.rst.
Not changed: Dag processor, CLI, callbacks, and BundleInfo / Execution API — there dag_id stays None.

Tests:

test_get_bundle — asserts dag_id is passed and defaults to None.
test_parse_passes_dag_id_to_get_bundle — asserts parse() calls get_bundle(..., dag_id=...).

[ashb]

It's a little bit odd/sub-optimal, but this might be available today via https://airflow.apache.org/docs/apache-airflow/stable/howto/dynamic-dag-generation.html#optimizing-dag-parsing-delays-during-execution:

from airflow.sdk import get_parsing_context

current_dag_id = get_parsing_context().dag_id

[ykuc]

It's a little bit odd/sub-optimal, but this might be available today via

That's actually what I'm using today, but it doesn't feel very reliable because it's not part of the public contract. I think this is a useful capability, so I'd prefer to make it an explicit, supported API rather than relying on the current implementation details.

[potiuk]

@ykuc This PR has been converted to draft because it does not yet meet our Pull Request quality criteria.

Issues found:

• ❌ Provider tests: Failing: Provider distributions tests / Compat 2.11.1:P3.10:, Provider distributions tests / Compat 3.0.6:P3.10:, Provider distributions tests / Compat 3.1.8:P3.10:, Provider distributions tests / Compat 3.2.2:P3.10:, Provider distributions tests / Compat 3.3.0:P3.10:. Run breeze testing providers-tests --test-type "Providers[<provider>]" locally.
• ❌ Unresolved review comments: There are 1 unresolved review thread(s) on this PR from @ashb. For each one: either apply the suggestion in a follow-up commit, or reply in-thread with a brief explanation of why the feedback doesn't apply, or resolve the thread if the feedback is no longer relevant. Once you believe a thread is resolved — whether by pushing a fix or by explaining why the suggestion doesn't apply — it is the author's responsibility to mark it as resolved by clicking the 'Resolve conversation' button at the bottom of the thread. Reviewers don't auto-close their own threads, so unresolved threads read as 'still waiting on the author' and block the PR from moving forward.

What to do next:

• Fix each issue listed above.
• Make sure static checks pass locally (prek run --from-ref main --stage pre-commit).
• Mark the PR as "Ready for review" when you're done.

Converting a PR to draft is not a rejection — it is an invitation to bring the PR up to the project's standards so that maintainer review time is spent productively. There is no rush — take your time and work at your own pace. We appreciate your contribution and are happy to wait for updates. If you have questions, feel free to ask on the Airflow Slack.

---

Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you.