Webhook tasks using FlyteAgents #3058
Conversation
Signed-off-by: Ketan Umare <[email protected]>
Signed-off-by: Ketan Umare <[email protected]>
Code Review Agent Run Status
|
Signed-off-by: Ketan Umare <[email protected]>
Signed-off-by: Ketan Umare <[email protected]>
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #3058 +/- ##
==========================================
+ Coverage 72.87% 76.45% +3.58%
==========================================
Files 205 210 +5
Lines 21553 21673 +120
Branches 2746 2764 +18
==========================================
+ Hits 15707 16571 +864
+ Misses 5062 4340 -722
+ Partials 784 762 -22 ☔ View full report in Codecov by Sentry. |
Code Review Agent Run Status
|
Signed-off-by: Ketan Umare <[email protected]>
kumare3
changed the title
kumare3
requested review from
wild-endeavor,
eapolinario,
pingsutw,
cosmicBboy,
samhita-alla,
thomasjpfan and
Future-Outlier
as code owners
Code Review Agent Run #8c3d4cActionable Suggestions - 6
Additional Suggestions - 1
Review Details
AI Code Review powered by |
Changelist by BitoThis pull request implements the following key changes.
|
flytekit/extras/webhook/constants.py
Outdated
| TASK_TYPE = "webhook" | ||
|
|
||
| URL_KEY = "url" | ||
| METHOD_KEY = "method" | ||
| HEADERS_KEY = "headers" | ||
| BODY_KEY = "body" | ||
| SHOW_BODY_KEY = "show_body" | ||
| SHOW_URL_KEY = "show_url" |
There was a problem hiding this comment.
Consider adding type hints to the constant declarations to improve code maintainability and IDE support. For example: TASK_TYPE: str = "webhook"
Code suggestion
Check the AI-generated fix before applying
| TASK_TYPE = "webhook" | |
| URL_KEY = "url" | |
| METHOD_KEY = "method" | |
| HEADERS_KEY = "headers" | |
| BODY_KEY = "body" | |
| SHOW_BODY_KEY = "show_body" | |
| SHOW_URL_KEY = "show_url" | |
| TASK_TYPE: str = "webhook" | |
| URL_KEY: str = "url" | |
| METHOD_KEY: str = "method" | |
| HEADERS_KEY: str = "headers" | |
| BODY_KEY: str = "body" | |
| SHOW_BODY_KEY: str = "show_body" | |
| SHOW_URL_KEY: str = "show_url" |
Code Review Run #8c3d4c
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
flytekit/extras/webhook/agent.py
Outdated
| url = final_dict.get(URL_KEY) | ||
| body = final_dict.get(BODY_KEY) | ||
| headers = final_dict.get(HEADERS_KEY) | ||
| method = final_dict.get(METHOD_KEY) | ||
| method = http.HTTPMethod(method) | ||
| show_body = final_dict.get(SHOW_BODY_KEY, False) | ||
| show_url = final_dict.get(SHOW_URL_KEY, False) | ||
|
|
||
| session = await self._get_session() | ||
|
|
||
| text = None | ||
| if method == http.HTTPMethod.GET: | ||
| response = await session.get(url, headers=headers) | ||
| text = await response.text() | ||
| else: | ||
| response = await session.post(url, json=body, headers=headers) | ||
| text = await response.text() | ||
| if response.status != 200: | ||
| return Resource( | ||
| phase=TaskExecution.FAILED, | ||
| message=f"Webhook failed with status code {response.status}, response: {text}", | ||
| ) | ||
| final_response = { | ||
| "status_code": response.status, | ||
| "body": text, | ||
| } | ||
| if show_body: | ||
| final_response["input_body"] = body | ||
| if show_url: | ||
| final_response["url"] = url | ||
|
|
||
| return Resource( | ||
| phase=TaskExecution.SUCCEEDED, | ||
| outputs={"info": final_response}, | ||
| message="Webhook was successfully invoked!", | ||
| ) | ||
| except Exception as e: | ||
| return Resource(phase=TaskExecution.FAILED, message=str(e)) |
There was a problem hiding this comment.
The do method is quite long and handles multiple responsibilities including HTTP request handling, response processing, and error handling. Consider breaking it down into smaller, focused methods for better maintainability.
Code suggestion
Check the AI-generated fix before applying
| url = final_dict.get(URL_KEY) | |
| body = final_dict.get(BODY_KEY) | |
| headers = final_dict.get(HEADERS_KEY) | |
| method = final_dict.get(METHOD_KEY) | |
| method = http.HTTPMethod(method) | |
| show_body = final_dict.get(SHOW_BODY_KEY, False) | |
| show_url = final_dict.get(SHOW_URL_KEY, False) | |
| session = await self._get_session() | |
| text = None | |
| if method == http.HTTPMethod.GET: | |
| response = await session.get(url, headers=headers) | |
| text = await response.text() | |
| else: | |
| response = await session.post(url, json=body, headers=headers) | |
| text = await response.text() | |
| if response.status != 200: | |
| return Resource( | |
| phase=TaskExecution.FAILED, | |
| message=f"Webhook failed with status code {response.status}, response: {text}", | |
| ) | |
| final_response = { | |
| "status_code": response.status, | |
| "body": text, | |
| } | |
| if show_body: | |
| final_response["input_body"] = body | |
| if show_url: | |
| final_response["url"] = url | |
| return Resource( | |
| phase=TaskExecution.SUCCEEDED, | |
| outputs={"info": final_response}, | |
| message="Webhook was successfully invoked!", | |
| ) | |
| except Exception as e: | |
| return Resource(phase=TaskExecution.FAILED, message=str(e)) | |
| return await self._process_webhook(final_dict) | |
| except Exception as e: | |
| return Resource(phase=TaskExecution.FAILED, message=str(e)) | |
| async def _make_http_request(self, method: http.HTTPMethod, url: str, headers: dict, body: dict = None) -> tuple: | |
| session = await self._get_session() | |
| if method == http.HTTPMethod.GET: | |
| response = await session.get(url, headers=headers) | |
| else: | |
| response = await session.post(url, json=body, headers=headers) | |
| text = await response.text() | |
| return response, text | |
| def _build_response(self, response: aiohttp.ClientResponse, text: str, body: dict = None, url: str = None, | |
| show_body: bool = False, show_url: bool = False) -> dict: | |
| final_response = { | |
| "status_code": response.status, | |
| "body": text, | |
| } | |
| if show_body: | |
| final_response["input_body"] = body | |
| if show_url: | |
| final_response["url"] = url | |
| return final_response | |
| async def _process_webhook(self, final_dict: dict) -> Resource: | |
| url = final_dict.get(URL_KEY) | |
| body = final_dict.get(BODY_KEY) | |
| headers = final_dict.get(HEADERS_KEY) | |
| method = http.HTTPMethod(final_dict.get(METHOD_KEY)) | |
| show_body = final_dict.get(SHOW_BODY_KEY, False) | |
| show_url = final_dict.get(SHOW_URL_KEY, False) | |
| response, text = await self._make_http_request(method, url, headers, body) | |
| if response.status != 200: | |
| return Resource( | |
| phase=TaskExecution.FAILED, | |
| message=f"Webhook failed with status code {response.status}, response: {text}", | |
| ) | |
| final_response = self._build_response(response, text, body, url, show_body, show_url) | |
| return Resource( | |
| phase=TaskExecution.SUCCEEDED, | |
| outputs={"info": final_response}, | |
| message="Webhook was successfully invoked!", | |
| ) |
Code Review Run #8c3d4c
Consider adding a timeout type check to ensure timeout is a positive integer before making the HTTP request. The current implementation assumes timeout is always valid.
Code suggestion
Check the AI-generated fix before applying
@@ -53,8 +53,11 @@ class WebhookAgent(SyncAgentBase):
async def _make_http_request(
self, method: http.HTTPMethod, url: str, headers: dict, data: dict, timeout: int
) -> tuple:
+ if not isinstance(timeout, int) or timeout <= 0:
+ raise ValueError(f'timeout must be a positive integer, got {timeout}')
+
if method == http.HTTPMethod.GET:
response = await self._client.get(url, headers=headers, params=data, timeout=timeout)
Code Review Run #49f39f
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
flytekit/extras/webhook/agent.py
Outdated
| outputs={"info": final_response}, | ||
| message="Webhook was successfully invoked!", | ||
| ) | ||
| except Exception as e: |
There was a problem hiding this comment.
Catching generic 'Exception' may hide bugs. Consider catching specific exceptions like 'aiohttp.ClientError'.
Code suggestion
Check the AI-generated fix before applying
| except Exception as e: | |
| except (aiohttp.ClientError, ValueError) as e: |
Code Review Run #8c3d4c
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
| async def test_do_post_success(mock_task_template, mock_aiohttp_session): | ||
| mock_response = AsyncMock() | ||
| mock_response.status = 200 | ||
| mock_response.text = "Success" |
There was a problem hiding this comment.
Consider using AsyncMock(return_value='Success') directly for text instead of assigning it as a property. This would better match the async nature of the response.
Code suggestion
Check the AI-generated fix before applying
| mock_response.text = "Success" | |
| mock_response.text = AsyncMock(return_value="Success") |
Code Review Run #8c3d4c
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
| show_url=True | ||
| ) | ||
|
|
||
| settings = SerializationSettings(image_config=ImageConfig.auto_default_image()) |
There was a problem hiding this comment.
Consider adding test cases to verify error handling when invalid inputs are provided to WebhookTask. For example, testing with invalid URLs, unsupported HTTP methods, or malformed headers/body.
Code suggestion
Check the AI-generated fix before applying
@@ -48,0 +49,25 @@
+ def test_webhook_task_invalid_inputs():
+ # Test invalid URL
+ with pytest.raises(ValueError):
+ WebhookTask(
+ name="test_task",
+ url="invalid-url",
+ method=http.HTTPMethod.POST
+ )
+
+ # Test invalid method
+ with pytest.raises(ValueError):
+ WebhookTask(
+ name="test_task",
+ url="http://example.com",
+ method="INVALID"
+ )
+
+ # Test invalid headers
+ with pytest.raises(ValueError):
+ WebhookTask(
+ name="test_task",
+ url="http://example.com",
+ method=http.HTTPMethod.POST,
+ headers="invalid-headers"
+ )
Code Review Run #8c3d4c
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
Signed-off-by: Ketan Umare <[email protected]>
Signed-off-by: Ketan Umare <[email protected]>
Code Review Agent Run #cec794Actionable Suggestions - 5
Additional Suggestions - 1
Review Details
AI Code Review powered by |
| status, text = await self._make_http_request(method, url, headers, body) | ||
| if status != 200: | ||
| return Resource( | ||
| phase=TaskExecution.FAILED, |
There was a problem hiding this comment.
Consider adding error handling for the _process_webhook call. The method could raise other exceptions besides aiohttp.ClientError that should be caught.
Code suggestion
Check the AI-generated fix before applying
| phase=TaskExecution.FAILED, | |
| return Resource(phase=TaskExecution.FAILED, message=f"HTTP client error: {str(e)}") | |
| except Exception as e: | |
| return Resource(phase=TaskExecution.FAILED, message=f"Webhook processing error: {str(e)}") |
Code Review Run #cec794
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
flytekit/extras/webhook/agent.py
Outdated
| async def _make_http_request(self, method: http.HTTPMethod, url: str, headers: dict, data: dict = None) -> tuple: | ||
| # TODO This is a potential performance bottleneck. Consider using a connection pool. To do this, we need to | ||
| # create a session object and reuse it for multiple requests. This will reduce the overhead of creating a new | ||
| # connection for each request. The problem for not doing so is local execution, does not have a common event | ||
| # loop and agent executor creates a new event loop for each request (in the mixin). | ||
| async with aiohttp.ClientSession() as session: | ||
| if method == http.HTTPMethod.GET: | ||
| response = await session.get(url, headers=headers, params=data) | ||
| else: | ||
| response = await session.post(url, json=data, headers=headers) | ||
| text = await response.text() | ||
| return response.status, text |
There was a problem hiding this comment.
Consider adding error handling for HTTP request failures in _make_http_request. The method should handle connection errors, timeouts and other potential aiohttp exceptions to provide better error reporting.
Code suggestion
Check the AI-generated fix before applying
- if method == http.HTTPMethod.GET:
- response = await session.get(url, headers=headers, params=data)
- else:
- response = await session.post(url, json=data, headers=headers)
- text = await response.text()
- return response.status, text
+ try:
+ if method == http.HTTPMethod.GET:
+ response = await session.get(url, headers=headers, params=data)
+ else:
+ response = await session.post(url, json=data, headers=headers)
+ text = await response.text()
+ return response.status, text
+ except aiohttp.ClientError as e:
+ raise RuntimeError(f"HTTP request failed: {str(e)}")
+ except asyncio.TimeoutError:
+ raise RuntimeError("HTTP request timed out")
+ except Exception as e:
+ raise RuntimeError(f"Unexpected error during HTTP request: {str(e)}")
Code Review Run #cec794
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
| self._url = url | ||
| self._method = method | ||
| self._headers = headers | ||
| self._data = data |
There was a problem hiding this comment.
Consider adding validation for the data parameter to ensure it is JSON serializable before sending the request. This could help catch serialization issues early.
Code suggestion
Check the AI-generated fix before applying
@@ -85,2 +85,7 @@
self._headers = headers
+ if data is not None:
+ import json
+ try:
+ json.dumps(data)
+ except (TypeError, ValueError) as e:
+ raise ValueError(f"The data parameter must be JSON serializable. Error: {str(e)}")
self._data = data
Code Review Run #cec794
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
| DATA_KEY: self._data or {}, | ||
| SHOW_DATA_KEY: self._show_data, |
There was a problem hiding this comment.
Consider if renaming from body to data maintains backward compatibility. This change could potentially break existing code that relies on the body parameter.
Code suggestion
Check the AI-generated fix before applying
| DATA_KEY: self._data or {}, | |
| SHOW_DATA_KEY: self._show_data, | |
| DATA_KEY: self._data or self._body or {}, | |
| SHOW_DATA_KEY: self._show_data if self._show_data is not None else self._show_body, |
Code Review Run #cec794
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
| async def test_do_post_success(mock_task_template, mock_aiohttp_session): | ||
| mock_response = AsyncMock() | ||
| mock_response.status = 200 | ||
| mock_response.text = AsyncMock(return_value="Success") |
There was a problem hiding this comment.
Consider adding error handling for the text property mock. The test assumes the text will always be successfully retrieved, but in real scenarios this could fail.
Code suggestion
Check the AI-generated fix before applying
| mock_response.text = AsyncMock(return_value="Success") | |
| # Test success case | |
| mock_response.text = AsyncMock(return_value="Success") | |
| # Test error case | |
| mock_response_error = AsyncMock() | |
| mock_response_error.status = 200 | |
| mock_response_error.text = AsyncMock(side_effect=Exception("Failed to get response text")) |
Code Review Run #cec794
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
Signed-off-by: Ketan Umare <[email protected]>
Signed-off-by: Ketan Umare <[email protected]>
Code Review Agent Run #882444Actionable Suggestions - 5
Additional Suggestions - 1
Review Details
AI Code Review powered by |
|
|
||
| async def _make_http_request( | ||
| self, method: http.HTTPMethod, url: str, headers: dict, data: dict, timeout: int | ||
| ) -> tuple: |
There was a problem hiding this comment.
The _make_http_request method's return type annotation tuple is too generic. Consider specifying the exact types being returned (int for status and str for text).
Code suggestion
Check the AI-generated fix before applying
| ) -> tuple: | |
| ) -> tuple[int, str]: |
Code Review Run #882444
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
| self, task_template: TaskTemplate, output_prefix: str, inputs: Optional[LiteralMap] = None, **kwargs | ||
| ) -> Resource: | ||
| try: | ||
| final_dict = self._get_final_dict(task_template, inputs) |
There was a problem hiding this comment.
Consider adding error handling for self._get_final_dict() call to handle potential exceptions from format_dict() or literal_map_string_repr()
Code suggestion
Check the AI-generated fix before applying
| final_dict = self._get_final_dict(task_template, inputs) | |
| try: | |
| final_dict = self._get_final_dict(task_template, inputs) | |
| except ValueError as e: | |
| return Resource(phase=TaskExecution.FAILED, message=f"Failed to format webhook data: {str(e)}") |
Code Review Run #882444
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
| timeout_sec = final_dict.get(TIMEOUT_SEC, 10) | ||
|
|
||
| status, text = await self._make_http_request(method, url, headers, body, timeout_sec) |
There was a problem hiding this comment.
Consider validating the timeout_sec value to ensure it's a positive integer. A negative or zero timeout could cause unexpected behavior.
Code suggestion
Check the AI-generated fix before applying
| timeout_sec = final_dict.get(TIMEOUT_SEC, 10) | |
| status, text = await self._make_http_request(method, url, headers, body, timeout_sec) | |
| timeout_sec = final_dict.get(TIMEOUT_SEC, 10) | |
| if not isinstance(timeout_sec, int) or timeout_sec <= 0: | |
| timeout_sec = 10 | |
| status, text = await self._make_http_request(method, url, headers, body, timeout_sec) |
Code Review Run #882444
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
| show_data=True, | ||
| show_url=True, | ||
| description="Test Webhook Task", | ||
| timeout=60, |
There was a problem hiding this comment.
Consider using timedelta for timeout parameter instead of raw integer value for better type safety and clarity. The WebhookTask already handles converting timedelta to seconds internally.
Code suggestion
Check the AI-generated fix before applying
| timeout=60, | |
| timeout=timedelta(seconds=60), |
Code Review Run #882444
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
|
What's the purpose and the use case for it? If it's to be used like a standard If the intent was to somehow address the flyteorg/flyte#2317
then I don't believe it will be useful directly... |
Code Review Agent Run #a6a0b4Actionable Suggestions - 0Additional Suggestions - 10
Review Details
AI Code Review powered by |
Signed-off-by: Ketan Umare <[email protected]>
kumare3
force-pushed
the
rest-api-agent
branch
from
8c78c2d to
293a704
Compare
Great question, the reason to have it is to provide a simple canonical way to fire endpoints synchronously, without having to create new containers. This makes it possible to have simple notifications directly in your workflow wherever even in |
Code Review Agent Run #49f39fActionable Suggestions - 7
Review Details
AI Code Review powered by |
| if method == http.HTTPMethod.GET: | ||
| response = await self._client.get(url, headers=headers, params=data, timeout=timeout) | ||
| else: | ||
| response = await self._client.post(url, json=data, headers=headers, timeout=timeout) | ||
| return response.status_code, response.text |
There was a problem hiding this comment.
Consider adding error handling for potential httpx request failures. The current implementation may not properly handle network timeouts or connection errors.
Code suggestion
Check the AI-generated fix before applying
| if method == http.HTTPMethod.GET: | |
| response = await self._client.get(url, headers=headers, params=data, timeout=timeout) | |
| else: | |
| response = await self._client.post(url, json=data, headers=headers, timeout=timeout) | |
| return response.status_code, response.text | |
| try: | |
| if method == http.HTTPMethod.GET: | |
| response = await self._client.get(url, headers=headers, params=data, timeout=timeout) | |
| else: | |
| response = await self._client.post(url, json=data, headers=headers, timeout=timeout) | |
| return response.status_code, response.text | |
| except (httpx.RequestError, httpx.TimeoutException) as e: | |
| return 500, str(e) |
Code Review Run #49f39f
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
|
|
||
| def __init__(self, client: Optional[httpx.AsyncClient] = None): | ||
| super().__init__(task_type_name=TASK_TYPE) | ||
| self._client = client or httpx.AsyncClient() |
There was a problem hiding this comment.
Consider managing the lifecycle of the httpx.AsyncClient() by implementing a cleanup method to properly close the client when it's no longer needed. The client should be closed to free up system resources.
Code suggestion
Check the AI-generated fix before applying
- def __init__(self, client: Optional[httpx.AsyncClient] = None):
- super().__init__(task_type_name=TASK_TYPE)
- self._client = client or httpx.AsyncClient()
+ def __init__(self, client: Optional[httpx.AsyncClient] = None):
+ super().__init__(task_type_name=TASK_TYPE)
+ self._client = client or httpx.AsyncClient()
+
+ async def cleanup(self):
+ if self._client:
+ await self._client.aclose()
Code Review Run #49f39f
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
| mock_httpx_client.post.return_value = mock_response | ||
|
|
||
| agent = WebhookAgent(client=mock_httpx_client) | ||
| result = await agent.do(mock_task_template, output_prefix="", inputs=LiteralMap({})) |
There was a problem hiding this comment.
Consider adding assertions to verify that mock_httpx_client.post was called with the expected arguments from mock_task_template.
Code suggestion
Check the AI-generated fix before applying
| result = await agent.do(mock_task_template, output_prefix="", inputs=LiteralMap({})) | |
| result = await agent.do(mock_task_template, output_prefix="", inputs=LiteralMap({})) | |
| mock_httpx_client.post.assert_called_once_with( | |
| mock_task_template.custom[URL_KEY], | |
| headers=mock_task_template.custom.get(HEADERS_KEY, {}), | |
| json=mock_task_template.custom.get(DATA_KEY, {})) |
Code Review Run #49f39f
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
| async def test_do_failure(mock_task_template): | ||
| mock_response = AsyncMock(name="httpx.Response") | ||
| mock_response.status_code = 500 | ||
| mock_response.text = "Internal Server Error" | ||
| mock_httpx_client = AsyncMock(name="httpx.AsyncClient") | ||
| mock_httpx_client.post.return_value = mock_response |
There was a problem hiding this comment.
Consider adding test cases for different HTTP status codes (4xx, 5xx) to ensure comprehensive error handling coverage in the webhook agent.
Code suggestion
Check the AI-generated fix before applying
| async def test_do_failure(mock_task_template): | |
| mock_response = AsyncMock(name="httpx.Response") | |
| mock_response.status_code = 500 | |
| mock_response.text = "Internal Server Error" | |
| mock_httpx_client = AsyncMock(name="httpx.AsyncClient") | |
| mock_httpx_client.post.return_value = mock_response | |
| @pytest.mark.parametrize("status_code,error_message", [ | |
| (400, "Bad Request"), | |
| (401, "Unauthorized"), | |
| (403, "Forbidden"), | |
| (404, "Not Found"), | |
| (500, "Internal Server Error"), | |
| (503, "Service Unavailable") | |
| ]) | |
| async def test_do_failure(mock_task_template, status_code, error_message): | |
| mock_response = AsyncMock(name="httpx.Response") | |
| mock_response.status_code = status_code | |
| mock_response.text = error_message | |
| mock_httpx_client = AsyncMock(name="httpx.AsyncClient") | |
| mock_httpx_client.post.return_value = mock_response |
Code Review Run #49f39f
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
| async def test_do_get_failure(mock_task_template): | ||
| mock_task_template.custom[METHOD_KEY] = "GET" | ||
| mock_task_template.custom.pop(DATA_KEY) | ||
| mock_task_template.custom[SHOW_DATA_KEY] = False | ||
|
|
||
| mock_response = AsyncMock(name="httpx.Response") | ||
| mock_response.status_code = 500 | ||
| mock_response.text = "Internal Server Error" | ||
| mock_httpx_client = AsyncMock(name="httpx.AsyncClient") | ||
| mock_httpx_client.get.return_value = mock_response | ||
|
|
||
| agent = WebhookAgent(client=mock_httpx_client) |
There was a problem hiding this comment.
The test case for GET failure duplicates much of the setup code from the POST failure test. Consider using a fixture or helper method to reduce duplication.
Code suggestion
Check the AI-generated fix before applying
@pytest.fixture
+def mock_error_response():
+ mock_response = AsyncMock(name="httpx.Response")
+ mock_response.status_code = 500
+ mock_response.text = "Internal Server Error"
+ return mock_response
+
[email protected]
+def mock_http_client(mock_error_response):
+ mock_client = AsyncMock(name="httpx.AsyncClient")
+ mock_client.post.return_value = mock_error_response
+ mock_client.get.return_value = mock_error_response
+ return mock_client
+
@@ -69,31 +80,19 @@
-async def test_do_failure(mock_task_template):
- mock_response = AsyncMock(name="httpx.Response")
- mock_response.status_code = 500
- mock_response.text = "Internal Server Error"
- mock_httpx_client = AsyncMock(name="httpx.AsyncClient")
- mock_httpx_client.post.return_value = mock_response
+async def test_do_failure(mock_task_template, mock_http_client):
+ agent = WebhookAgent(client=mock_http_client)
- agent = WebhookAgent(client=mock_httpx_client)
@@ -84,16 +95,8 @@
-async def test_do_get_failure(mock_task_template):
+async def test_do_get_failure(mock_task_template, mock_http_client):
mock_task_template.custom[METHOD_KEY] = "GET"
mock_task_template.custom.pop(DATA_KEY)
mock_task_template.custom[SHOW_DATA_KEY] = False
-
- mock_response = AsyncMock(name="httpx.Response")
- mock_response.status_code = 500
- mock_response.text = "Internal Server Error"
- mock_httpx_client = AsyncMock(name="httpx.AsyncClient")
- mock_httpx_client.get.return_value = mock_response
-
- agent = WebhookAgent(client=mock_httpx_client)
+ agent = WebhookAgent(client=mock_http_client)
Code Review Run #49f39f
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
There have been numerous requests from folks to support invoking arbitrary apis or using webhooks to notify.
This agent supports calling webhooks like slack, github etc directly from a flyte workflow
Possible to write workflows like,
Summary by Bito
This PR implements WebhookTask and WebhookAgent components in Flytekit, transitioning from aiohttp to httpx for HTTP operations. The implementation supports GET/POST methods with configurable timeouts and dynamic input parameters. The changes include comprehensive documentation and enhanced validation for pip arguments. The system is made more robust through proper error handling, validation, and graceful failure recovery mechanisms.Unit tests added: True
Estimated effort to review (1-5, lower is better): 5