dominikhei commented on code in PR #53323:
URL: https://github.com/apache/airflow/pull/53323#discussion_r2203557096
##########
airflow-ctl/src/airflowctl/api/operations.py:
##########
@@ -388,150 +361,104 @@ def bulk(self, connections: BulkBodyConnectionBody) ->
BulkResponse | ServerResp
return BulkResponse.model_validate_json(self.response.content)
except ServerResponseError as e:
raise e
+ # return super().execute_operation(method="patch",url =
"connections",data_model = BulkResponse,json=connections.model_dump())
def create_defaults(self) -> None | ServerResponseError:
"""Create default connections."""
- try:
- self.response = self.client.post("connections/defaults")
- return None
- except ServerResponseError as e:
- raise e
+ return super().execute_operation(method="post",
url="connections/defaults")
def delete(self, conn_id: str) -> str | ServerResponseError:
"""Delete a connection."""
- try:
- self.client.delete(f"connections/{conn_id}")
- return conn_id
- except ServerResponseError as e:
- raise e
+ return super().execute_operation(method="delete",
url=f"connections/{conn_id}")
def update(
self,
connection: ConnectionBody,
) -> ConnectionResponse | ServerResponseError:
"""Update a connection."""
- try:
- self.response = self.client.patch(
- f"connections/{connection.connection_id}",
json=connection.model_dump()
- )
- return
ConnectionResponse.model_validate_json(self.response.content)
- except ServerResponseError as e:
- raise e
+ return super().execute_operation(
+ method="patch",
+ url=f"connections/{connection.connection_id}",
+ data_model=ConnectionResponse,
+ json=connection.model_dump(),
+ )
def test(
self,
connection: ConnectionBody,
) -> ConnectionTestResponse | ServerResponseError:
"""Test a connection."""
- try:
- self.response = self.client.post("connections/test",
json=connection.model_dump())
- return
ConnectionTestResponse.model_validate_json(self.response.content)
- except ServerResponseError as e:
- raise e
+ return super().execute_operation(
+ method="post", url="connections/test",
data_model=ConnectionTestResponse
+ )
class DagOperations(BaseOperations):
"""Dag operations."""
def get(self, dag_id: str) -> DAGResponse | ServerResponseError:
"""Get a DAG."""
- try:
- self.response = self.client.get(f"dags/{dag_id}")
- return DAGResponse.model_validate_json(self.response.content)
- except ServerResponseError as e:
- raise e
+ return super().execute_operation(method="get", url="dags/{dag_id}",
data_model=DAGResponse)
Review Comment:
you forgot to make it into an f-string here :)
--
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]
