ykuc commented on code in PR #69697:
URL: https://github.com/apache/airflow/pull/69697#discussion_r3561795098
##########
airflow-core/src/airflow/dag_processing/bundles/manager.py:
##########
@@ -396,22 +396,31 @@ def _extract_template_params(bundle_instance:
BaseDagBundle) -> dict:
return params
def get_bundle(
- self, name: str, version: str | None = None, version_data: dict[str,
Any] | None = None
+ self,
+ name: str,
+ version: str | None = None,
+ version_data: dict[str, Any] | None = None,
+ dag_id: str | None = None,
) -> BaseDagBundle:
"""
Get a DAG bundle by name.
:param name: The name of the DAG bundle.
:param version: The version of the DAG bundle you need (optional). If
not provided, ``tracking_ref`` will be used instead.
:param version_data: Optional structured data associated with this
version (e.g., S3 manifest).
+ :param dag_id: Dag ID when the bundle is constructed for a specific
task run on a worker (optional).
:return: The DAG bundle.
"""
cfg_bundle = self._bundle_config.get(name)
if not cfg_bundle:
raise ValueError(f"Requested bundle '{name}' is not configured.")
return cfg_bundle.bundle_class(
- name=name, version=version, version_data=version_data,
**cfg_bundle.kwargs
+ name=name,
+ version=version,
+ version_data=version_data,
+ dag_id=dag_id,
Review Comment:
we can make
```
class BaseDagBundle:
def set_parsing_context(self, *, dag_id: str | None) -> None:
self.dag_id = dag_id
bundle = bundle_class(
name=name,
version=version,
version_data=version_data,
**bundle_kwargs,
)
bundle.set_parsing_context(dag_id=dag_id)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
