This is an automated email from the ASF dual-hosted git repository.
vincbeck pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new e65b2384e27 Add `allow_producer_teams` to the serialization layer
(#65980)
e65b2384e27 is described below
commit e65b2384e271fa2727b63102296efb080e0510cd
Author: Vincent <[email protected]>
AuthorDate: Wed Apr 29 13:22:45 2026 -0400
Add `allow_producer_teams` to the serialization layer (#65980)
---
airflow-core/src/airflow/models/asset.py | 15 +++++-
.../airflow/serialization/definitions/assets.py | 1 +
airflow-core/tests/unit/models/test_asset.py | 63 ++++++++++++++++++++++
3 files changed, 77 insertions(+), 2 deletions(-)
diff --git a/airflow-core/src/airflow/models/asset.py
b/airflow-core/src/airflow/models/asset.py
index 5d204713b7f..7a43e056d0d 100644
--- a/airflow-core/src/airflow/models/asset.py
+++ b/airflow-core/src/airflow/models/asset.py
@@ -337,7 +337,10 @@ class AssetModel(Base):
@classmethod
def from_serialized(cls, obj: SerializedAsset) -> AssetModel:
- return cls(name=obj.name, uri=obj.uri, group=obj.group,
extra=obj.extra)
+ extra = dict(obj.extra)
+ if obj.allow_producer_teams:
+ extra["allow_producer_teams"] = obj.allow_producer_teams
+ return cls(name=obj.name, uri=obj.uri, group=obj.group, extra=extra)
def __init__(self, name: str = "", uri: str = "", **kwargs):
if not name and not uri:
@@ -376,7 +379,15 @@ class AssetModel(Base):
def to_serialized(self) -> SerializedAsset:
from airflow.serialization.definitions.assets import SerializedAsset
- return SerializedAsset(name=self.name, uri=self.uri, group=self.group,
extra=self.extra, watchers=[])
+ extra = {k: v for k, v in self.extra.items() if k !=
"allow_producer_teams"}
+ return SerializedAsset(
+ name=self.name,
+ uri=self.uri,
+ group=self.group,
+ extra=extra,
+ watchers=[],
+ allow_producer_teams=self.extra.get("allow_producer_teams", []),
+ )
def add_trigger(self, trigger: Trigger, watcher_name: str):
self.watchers.append(AssetWatcherModel(name=watcher_name,
trigger_id=trigger.id))
diff --git a/airflow-core/src/airflow/serialization/definitions/assets.py
b/airflow-core/src/airflow/serialization/definitions/assets.py
index 51fe0ed3d1d..a741fabcdec 100644
--- a/airflow-core/src/airflow/serialization/definitions/assets.py
+++ b/airflow-core/src/airflow/serialization/definitions/assets.py
@@ -120,6 +120,7 @@ class SerializedAsset(SerializedAssetBase):
group: str
extra: dict[str, Any]
watchers: MutableSequence[SerializedAssetWatcher]
+ allow_producer_teams: list[str] = attrs.field(factory=list)
def as_expression(self) -> Any:
"""
diff --git a/airflow-core/tests/unit/models/test_asset.py
b/airflow-core/tests/unit/models/test_asset.py
index f4c9e5cc866..1f4ffa579f5 100644
--- a/airflow-core/tests/unit/models/test_asset.py
+++ b/airflow-core/tests/unit/models/test_asset.py
@@ -163,3 +163,66 @@ def test_remove_reference_for_inactive_dag(
_simulate_soft_dag_deletion("test2")
remove_references_to_deleted_dags(session=session)
assert set(session.execute(select_stmt)) == set()
+
+
+class TestAssetModelAllowProducerTeams:
+ @pytest.mark.parametrize(
+ ("extra", "expected_teams"),
+ [
+ pytest.param({"allow_producer_teams": ["team_a"]}, ["team_a"],
id="with_teams"),
+ pytest.param({}, [], id="without_teams"),
+ ],
+ )
+ def test_to_serialized_allow_producer_teams(self, extra, expected_teams):
+ model = AssetModel(name="test_asset", uri="test://asset", extra=extra)
+ serialized = model.to_serialized()
+ assert serialized.allow_producer_teams == expected_teams
+ assert "allow_producer_teams" not in serialized.extra
+
+ @pytest.mark.parametrize(
+ ("teams", "expected_in_extra"),
+ [
+ pytest.param(["team_a", "team_b"], True, id="with_teams"),
+ pytest.param([], False, id="empty_teams"),
+ ],
+ )
+ def test_from_serialized_allow_producer_teams(self, teams,
expected_in_extra):
+ from airflow.serialization.definitions.assets import SerializedAsset
+
+ serialized = SerializedAsset(
+ name="test_asset",
+ uri="test://asset",
+ group="asset",
+ extra={},
+ watchers=[],
+ allow_producer_teams=teams,
+ )
+ model = AssetModel.from_serialized(serialized)
+ if expected_in_extra:
+ assert model.extra["allow_producer_teams"] == teams
+ else:
+ assert "allow_producer_teams" not in model.extra
+
+ @pytest.mark.parametrize(
+ ("teams", "extra"),
+ [
+ pytest.param(["team_analytics", "team_ml"], {"some_key":
"some_value"}, id="with_teams"),
+ pytest.param([], {}, id="empty_teams"),
+ ],
+ )
+ def test_round_trip_preserves_allow_producer_teams(self, teams, extra):
+ from airflow.serialization.definitions.assets import SerializedAsset
+
+ serialized = SerializedAsset(
+ name="test_asset",
+ uri="test://asset",
+ group="asset",
+ extra=extra,
+ watchers=[],
+ allow_producer_teams=teams,
+ )
+ model = AssetModel.from_serialized(serialized)
+ result = model.to_serialized()
+ assert result.allow_producer_teams == teams
+ assert "allow_producer_teams" not in result.extra
+ assert result.extra == extra
