bugraoz93 commented on code in PR #50987:
URL: https://github.com/apache/airflow/pull/50987#discussion_r2107791040
##########
airflow-ctl/src/airflowctl/api/operations.py:
##########
@@ -198,6 +202,76 @@ def list_by_alias(self) -> AssetAliasCollectionResponse |
ServerResponseError:
except ServerResponseError as e:
raise e
+ def create_event(
+ self, asset_event_body: CreateAssetEventsBody
+ ) -> AssetEventResponse | ServerResponseError:
+ """Create an asset event."""
+ try:
+ self.response = self.client.post("assets/events",
json=asset_event_body.model_dump())
+ return
AssetEventResponse.model_validate_json(self.response.content)
+ except ServerResponseError as e:
+ raise e
+
+ def materialize(self, asset_id: str) -> DAGRunResponse |
ServerResponseError:
+ """Materialize an asset."""
+ try:
+ self.response = self.client.post(f"assets/{asset_id}/materialize")
+ return DAGRunResponse.model_validate_json(self.response.content)
+ except ServerResponseError as e:
+ raise e
+
+ def get_queued_events(self, asset_id: str) ->
QueuedEventCollectionResponse | ServerResponseError:
+ """Get queued events for an asset."""
+ try:
+ self.response = self.client.get(f"assets/{asset_id}/queuedEvents")
+ return
QueuedEventCollectionResponse.model_validate_json(self.response.content)
+ except ServerResponseError as e:
+ raise e
+
+ def get_dag_queued_events(
+ self, dag_id: str, before: str
+ ) -> QueuedEventCollectionResponse | ServerResponseError:
+ """Get queued events for a dag."""
+ try:
+ self.response =
self.client.get(f"dags/{dag_id}/assets/queuedEvents", params={"before": before})
+ return
QueuedEventCollectionResponse.model_validate_json(self.response.content)
+ except ServerResponseError as e:
+ raise e
+
+ def get_dag_queued_event(self, dag_id: str, asset_id: str) ->
QueuedEventResponse | ServerResponseError:
+ """Get a queued event for a dag."""
+ try:
+ self.response =
self.client.get(f"dags/{dag_id}/assets/{asset_id}/queuedEvents")
+ return
QueuedEventResponse.model_validate_json(self.response.content)
+ except ServerResponseError as e:
+ raise e
+
+ def delete_queued_events(self, asset_id: str) -> str | ServerResponseError:
+ """Delete a queued event for an asset."""
+ try:
+ self.response =
self.client.delete(f"assets/{asset_id}/queuedEvents/")
Review Comment:
Sure. We can delete it from all of the delete operations.
I am away from pc for a couple of hours, and I will push the changes soon.
Thanks for the review!
--
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]
