Apply remaining suggestions from the original review of the closed
PR apache#57320:

- bundles/manager.py: document that sync_bundles_to_db only reconciles
  bundle metadata and does not parse or write DAG rows, clarifying the
  signature confusion raised by @bhavaniravi during review.
- tests/unit/dag_processing/test_dagbag.py: stop importing the standard
  provider's example_dags module at collection time. Move the import
  into a small helper plus a pytest fixture so tests that need the
  folder request it explicitly and the module remains collectable when
  the standard provider is not yet importable. The single parametrize
  case that referenced the folder now passes a relative file name and
  resolves the absolute path inside the test.
- tests/unit/cli/commands/test_task_command.py: build a minimal DAG
  inline in test_task_states_for_dag_run instead of importing one from
  the standard provider's example_dags. The test only checks CLI
  behaviour around a known dag_id/task_id, so reproducing the name and
  a single task is enough to keep the core test decoupled from the
  standard provider's example DAGs.

Signed-off-by: André Ahlert <[email protected]>

…#57320) (#66161)

* chore: move standard examples to provider
1. loading example dags as bundles
2. fixing testcases to use bundles instead of dagbag folders
3. fixing testcases to use example_dags from standard module

Signed-off-by: André Ahlert <[email protected]>

* fix: failing testcases to load examples form dagbundle

Signed-off-by: André Ahlert <[email protected]>

* Resolve provider example DAGs via ProvidersManager

Replace the directory walk over airflow.providers.__path__ with a
lookup based on ProvidersManager. The previous approach silently
skipped:

- nested providers like apache-airflow-providers-common-sql, whose
  module path is airflow.providers.common.sql (one level deeper);
- providers installed outside the airflow.providers namespace
  package, since they are not visible via os.listdir.

The new implementation iterates over the providers registered through
the apache_airflow_provider entry point, imports each provider module
and adds its example_dags folder when present. Bundle names are now
keyed on the canonical package name to keep them unique and stable
across deployments.

Signed-off-by: André Ahlert <[email protected]>

* Tests: relax test_get_all_bundle_names assertion

The set of provider example DAG bundles depends on which providers
expose an example_dags folder, which is environment specific. Pin
only the built-in bundles and assert the prefix of any extra entry.

Signed-off-by: André Ahlert <[email protected]>

* Address review feedback on example DAG loading

Apply remaining suggestions from the original review of the closed
PR #57320:

- bundles/manager.py: document that sync_bundles_to_db only reconciles
  bundle metadata and does not parse or write DAG rows, clarifying the
  signature confusion raised by @bhavaniravi during review.
- tests/unit/dag_processing/test_dagbag.py: stop importing the standard
  provider's example_dags module at collection time. Move the import
  into a small helper plus a pytest fixture so tests that need the
  folder request it explicitly and the module remains collectable when
  the standard provider is not yet importable. The single parametrize
  case that referenced the folder now passes a relative file name and
  resolves the absolute path inside the test.
- tests/unit/cli/commands/test_task_command.py: build a minimal DAG
  inline in test_task_states_for_dag_run instead of importing one from
  the standard provider's example_dags. The test only checks CLI
  behaviour around a known dag_id/task_id, so reproducing the name and
  a single task is enough to keep the core test decoupled from the
  standard provider's example DAGs.

Signed-off-by: André Ahlert <[email protected]>

* Tests: use SQLAlchemy 2.0 delete() in trigger_dagrun teardown

The teardown deletes a single DagBundleModel row keyed by name. Use
session.execute(delete(...).where(...)) instead of the deprecated
session.query(...).filter_by(...).delete() form so prek's
prevent-deprecated-sqlalchemy-usage hook stays clean.

Signed-off-by: André Ahlert <[email protected]>

* Tests: update bundle_name for example_python_operator

After moving the standard provider's example DAGs into their own
bundle, example_python_operator no longer lives in the dags-folder
bundle. Update the 24 expected payloads in test_task_instances.py
that asserted bundle_name='dags-folder' to the new
airflow-provider-apache-airflow-providers-standard-example-dags
bundle name. The lone sync_bag_to_db('dags-folder', ...) call in this
file is unrelated; it registers a synthetic dag built by dag_maker.

Signed-off-by: André Ahlert <[email protected]>

* Tests: avoid double-loading example DAGs in parse_and_sync_to_db

On Airflow 3.1+, parse_and_sync_to_db iterates every registered DAG
bundle (including example_dags and airflow-provider-*-example-dags)
and syncs them with their own bundle name. The leading
DagBag(dag_folder=folder, include_examples=True) was also pulling
example DAGs into the dags-folder bundle, so each example DAG ended
up registered under two different bundles. The duplicated rows then
violated the (asset_id, dag_id) unique constraint on
dag_schedule_asset_reference and broke unrelated tests (notably the
fab provider tests under compat).

Force include_examples=False on the 3.1+ path; the bundle loop is
already responsible for loading example DAGs from their own bundles.
The 2.x and 3.0 paths are untouched.

Signed-off-by: André Ahlert <[email protected]>

* Trigger CI rerun

Signed-off-by: André Ahlert <[email protected]>

* Update devel-common/src/tests_common/test_utils/db.py

Co-authored-by: Jens Scheffler <[email protected]>
Signed-off-by: André Ahlert <[email protected]>

* Update airflow-core/src/airflow/dag_processing/bundles/manager.py

Co-authored-by: Jens Scheffler <[email protected]>
Signed-off-by: André Ahlert <[email protected]>

* Address review feedback on provider example DAG bundle discovery

- Hoist `importlib`, `logging`, and `ProvidersManager` to module top in
  `dag_processing/bundles/manager.py` and add a module-level logger.
- Reverse the `apache-airflow-providers-` prefix check so the canonical
  case reads first, per @jscheffl's nit.
- Broaden the exception handler around `import_module` and the
  subsequent `__path__` access to `Exception` with `log.exception`, so a
  provider with a custom `__getattr__` can no longer crash config
  loading on the scheduler/api-server.
- Switch the provider-example bundle dedup key from `bundle_name` to
  the resolved `example_dag_folder` and document the namespace-package
  scenario the guard exists for (multiple `airflow.providers.common.*`
  distributions sharing one namespace via `pkgutil.extend_path`).
- Rename the per-provider bundle from
  `airflow-provider-{package}-example-dags` to
  `{package}-example-dags`, matching the package distribution name
  surfaced to users in REST API responses and `pip list`.
- Emit a `DeprecationWarning` from `DagBag.collect_dags` and from
  `tests_common.test_utils.db.parse_and_sync_to_db` when callers pass
  `include_examples=True`, and update the docstrings to direct callers
  at the `[core] load_examples` configuration option.
- Migrate the in-tree `parse_and_sync_to_db(..., include_examples=True)`
  callers to use `conf_vars({("core", "load_examples"): "true"})`.
- Update `test_task_instances.py` bundle-name assertions and the
  `test_get_all_bundle_names` suffix check for the new format.
- Add `airflow-core/newsfragments/66161.significant.rst` covering the
  user-visible REST-API bundle-name change.

The package-name -> module-path heuristic is kept for now; replacing it
with an authoritative field on `ProviderInfo` is tracked in #66305.

Signed-off-by: André Ahlert <[email protected]>

* Remove include_examples parameter from DagBag and test helpers

Drops the include_examples knob from DagBag.__init__, BundleDagBag.__init__,
collect_dags, and parse_and_sync_to_db. Example DAGs now come in exclusively
through the per-provider example bundles registered when [core] load_examples
is enabled. Removes the deprecation warnings, updates all callers across core,
devel-common, and providers, and gates the few provider tests that still need
to read examples on older Airflow versions behind AIRFLOW_V_3_3_PLUS.

Signed-off-by: André Ahlert <[email protected]>

* Fix mypy and test assertion after DagBag include_examples removal

The previous commit removed include_examples from DagBag callers but left
two follow-ups that broke prek mypy and the config CLI test:

- devel-common test_utils/db.py: pre-3.1 compat branches still call
  DagBag(include_examples=False) for older Airflow runtimes; mypy now
  flags it because the current source no longer accepts that kwarg. Add
  call-arg type-ignore (matches the existing attr-defined pattern on the
  sync_to_db calls below).
- test_config_command.py: assertions for conf.write(...) lost the
  include_examples=False kwarg, but conf.write is AirflowConfigParser.write
  (config-file examples), unrelated to the DagBag flag, and still passes
  it. Restore the kwarg in the expected call.

Signed-off-by: André Ahlert <[email protected]>

* Drop include_examples from new DagBag callers and gate dag_maker for compat

The merge with main brought in tests in airflow-core/tests/unit/models/test_dag.py
that still passed include_examples=False to DagBag, and the dag_maker fixture in
pytest_plugin.py was loading examples by default on Airflow <3.3 because the
removed kwarg flipped the effective default to True. That broke compat 3.0.6
runs because the cleanup path then tried to roll back a session that was never
attached. Drop the kwarg from the new test_dag.py callers and gate both
dag_maker and get_test_dag DagBag construction on AIRFLOW_V_3_3_PLUS so older
Airflow versions still pass include_examples=False explicitly.

Signed-off-by: André Ahlert <[email protected]>

* Gate DagBag include_examples on Airflow 3.3+ for older compat runs

The PR drops include_examples=False from a number of DagBag callers because
DagBag in 3.3+ no longer accepts the kwarg and never loads examples by default.
On compat runs against Airflow 3.1.x and 3.2.x, parse_and_sync_to_db hit a
branch that called DagBag(folder) with the old default include_examples=True,
which let the dags-folder DagBag pull in core example DAGs that the bundle
loop also synced, producing duplicate dag_schedule_asset_reference rows and a
UNIQUE constraint failure. On Airflow 2.11.1, DagBag callers in the standard
external_task_sensor and time_delta tests defaulted to include_examples=True
and pulled in example DAGs whose required Params lack defaults, breaking
ExternalTaskMarker tests with ParamValidationError.

Restrict the new bundle-aware sync path to AIRFLOW_V_3_3_PLUS and pass
include_examples=False explicitly on older Airflow versions in the standard
provider tests so compat 2.11.1, 3.1.8, and 3.2.1 keep their original
behaviour.

Signed-off-by: André Ahlert <[email protected]>

* Trim 66161 newsfragment to user-visible facts

DagBag is not part of the Airflow public interface, so removing
include_examples from DagBag/collect_dags/BundleDagBag/parse_and_sync_to_db
does not need a release-notes migration recipe. Drop the internal removal
block and the Types of change / Migration rules scaffolding, keep the
bundle naming change and the bundle_name REST API behaviour change.

Signed-off-by: André Ahlert <[email protected]>

---------

Signed-off-by: André Ahlert <[email protected]>
Co-authored-by: Bhavani Ravi <[email protected]>
Co-authored-by: Jens Scheffler <[email protected]>