potiuk commented on code in PR #58993: URL: https://github.com/apache/airflow/pull/58993#discussion_r2622464808
########## airflow-core/src/airflow/serialization/definitions/assets.py: ########## @@ -18,10 +18,292 @@ from __future__ import annotations -from airflow.sdk import AssetWatcher # TODO: Implement serialized assets. +import json +from typing import TYPE_CHECKING, Any, ClassVar, Literal +import attrs -class SerializedAssetWatcher(AssetWatcher): - """JSON serializable representation of an asset watcher.""" +from airflow.api_fastapi.execution_api.datamodels.asset import AssetProfile +from airflow.serialization.dag_dependency import DagDependency +if TYPE_CHECKING: + from collections.abc import Callable, Iterable, Iterator, MutableSequence + + from typing_extensions import Self + + from airflow.models.asset import AssetModel + + AttrsInstance = attrs.AttrsInstance +else: + AttrsInstance = object + + [email protected](frozen=True) +class SerializedAssetUniqueKey(AttrsInstance): + """ + Columns to identify an unique asset. + + :meta private: + """ + + name: str + uri: str + + @classmethod + def from_asset(cls, asset: SerializedAsset | AssetModel) -> Self: + return cls(name=asset.name, uri=asset.uri) + + @classmethod + def from_str(cls, key: str) -> Self: + return cls(**json.loads(key)) + + def to_str(self) -> str: + return json.dumps(attrs.asdict(self)) + + def asprofile(self) -> AssetProfile: + return AssetProfile(name=self.name, uri=self.uri, type="Asset") + + +class SerializedAssetBase: Review Comment: Out of curiosity why not `SerializedAssetBase(Protocol)` ? -- 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]
