FrankYang0529 opened a new pull request, #73031:
URL: https://github.com/apache/airflow/pull/73031
## Why
- `serialize_template_field()` writes a callable as `<callable
{module}.{qualname}>` and a plain object as `<{module}.{qualname} object>`. For
a function or class defined in the Dag file, the marker includes the path-based
module name, which ends up in the Dag hash.
- If the bundle path changes between deployments, the hash changes even
though the Dag code stays the same. When the current version already has task
instances, each of these deployments creates a new Dag version.
## How
- Follow #60799 to pass `exclude_module=True` to `qualname()` for both
markers.
## Verification
- Unit test:
- `uv run --frozen --project airflow-core pytest
airflow-core/tests/unit/serialization/test_helpers.py
airflow-core/tests/unit/serialization/test_dag_serialization.py
airflow-core/tests/unit/models/test_renderedtifields.py`
- `breeze run pytest
task-sdk/tests/task_sdk/execution_time/test_task_runner.py`
- Integration test:
1. Setup
```sh
mkdir -p files/template-field-demo
cat > files/template-field-demo/etl.py <<'EOF'
from __future__ import annotations
import pendulum
from airflow.providers.standard.operators.python import PythonOperator
from airflow.sdk import DAG
def clean_rows(rows):
return [row.strip() for row in rows]
class LoadSettings:
batch_size = 500
def load(transform, settings):
print(transform([" a ", " b "]), settings.batch_size)
with DAG(dag_id="etl", start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
schedule=None):
PythonOperator(
task_id="load",
python_callable=load,
op_kwargs={"transform": clean_rows, "settings": LoadSettings()},
)
EOF
cat > files/template-field-demo/show_state.py <<'EOF'
import sys
from sqlalchemy import select
from airflow.models.dag_version import DagVersion
from airflow.models.serialized_dag import SerializedDagModel
from airflow.utils.session import create_session
with create_session() as session:
versions = session.scalars(
select(DagVersion.version_number).where(DagVersion.dag_id ==
"etl").order_by(DagVersion.version_number)
).all()
op_kwargs = SerializedDagModel.get("etl",
session=session).data["dag"]["tasks"][0]["__var"]["op_kwargs"]
print(f"{sys.argv[1]}: Dag versions = {versions}")
print(f" op_kwargs = {op_kwargs}")
EOF
cat > files/template-field-demo/deploy.sh <<'EOF'
set -euo pipefail
export AIRFLOW__CORE__LOAD_EXAMPLES=False
# Write each re-serialization right away instead of waiting 30 seconds
between writes.
export AIRFLOW__CORE__MIN_SERIALIZED_DAG_UPDATE_INTERVAL=0
demo=/files/template-field-demo
quiet() { "$@" > /tmp/demo-command.log 2>&1 || { cat /tmp/demo-command.log;
exit 1; }; }
quiet airflow db migrate
rm -rf "${demo}/releases"
for release in 41 42 43; do
# Deploy the same Dag file to a new directory, like a versioned deployment
does.
bundle_path="${demo}/releases/release-${release}"
mkdir -p "${bundle_path}"
cp "${demo}/etl.py" "${bundle_path}/etl.py"
export AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST="[{\"name\":
\"etl-bundle\", \"classpath\":
\"airflow.dag_processing.bundles.local.LocalDagBundle\", \"kwargs\": {\"path\":
\"${bundle_path}\"}}]"
quiet airflow dags reserialize --bundle-name etl-bundle
python "${demo}/show_state.py" "release-${release}"
# Run the Dag once, so the next deployment cannot update this version in
place.
quiet airflow dags trigger etl --run-id "manual__release-${release}"
done
EOF
```
2. Run demo script
```sh
breeze run --backend sqlite bash /files/template-field-demo/deploy.sh
```
On main branch, it bumps dag version with different paths.
```text
release-41: Dag versions = [1]
op_kwargs = {'settings':
'<unusual_prefix_b258b7f6578d0fef91ba779c52c32ad43a0d96b1_etl.LoadSettings
object>', 'transform': '<callable
unusual_prefix_b258b7f6578d0fef91ba779c52c32ad43a0d96b1_etl.clean_rows>'}
release-42: Dag versions = [1, 2]
op_kwargs = {'settings':
'<unusual_prefix_30d5abbd19b0bdd47056ee70d929329f050eff4c_etl.LoadSettings
object>', 'transform': '<callable
unusual_prefix_30d5abbd19b0bdd47056ee70d929329f050eff4c_etl.clean_rows>'}
release-43: Dag versions = [1, 2, 3]
op_kwargs = {'settings':
'<unusual_prefix_c6581b5a63a06e6a67c32562fc5425e66cef00de_etl.LoadSettings
object>', 'transform': '<callable
unusual_prefix_c6581b5a63a06e6a67c32562fc5425e66cef00de_etl.clean_rows>'}
```
On this PR, it shows same version.
```text
release-41: Dag versions = [1]
op_kwargs = {'settings': '<LoadSettings object>', 'transform': '<callable
clean_rows>'}
release-42: Dag versions = [1]
op_kwargs = {'settings': '<LoadSettings object>', 'transform': '<callable
clean_rows>'}
release-43: Dag versions = [1]
op_kwargs = {'settings': '<LoadSettings object>', 'transform': '<callable
clean_rows>'}
```
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull
request.
Write a good git commit message following this guide:
https://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the
documentation.
For user-facing UI changes, please attach before/after screenshots (or a
short
screen recording) so reviewers can assess the visual impact.
Feel free to ping (in general) for the review if you do not see reaction for
a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we
sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR,
please
change below checkbox to `[X]` followed by the name of the tool, uncomment
the "Generated-by".
-->
- [X] Yes - Claude Code
<!--
Generated-by: [Tool Name] following [the
guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
-->
---
* Read the **[Pull Request
Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)**
for more information. Note: commit author/co-author name and email in commits
become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal
([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals))
is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License
Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment:
`{pr_number}.significant.rst`, in
[airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments).
You can add this file in a follow-up commit after the PR is created so you
know the PR number.
--
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]
