📝 docs(async-api.mdx): update endpoint path for checking task status to improve consistency and clarity
🐛 fix(endpoints.py): update endpoint path for checking task status to match the updated path in the documentation 🐛 fix(schemas.py): add TaskResponse schema to properly handle task response data 🐛 fix(locustfile.py): update endpoint path for polling task status to match the updated path in the endpoints 🐛 fix(test_endpoints.py): update helper function and test cases to use the new task response structure and endpoint path for polling task status
This commit is contained in:
parent
a6ba2a24d8
commit
af35ae315e
5 changed files with 34 additions and 12 deletions
|
|
@ -29,11 +29,11 @@ curl -X POST \
|
||||||
|
|
||||||
## Checking Task Status
|
## Checking Task Status
|
||||||
|
|
||||||
You can check the status of an asynchronous task by making a GET request to the `/task/{task_id}/status` endpoint.
|
You can check the status of an asynchronous task by making a GET request to the `/task/{task_id}` endpoint.
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
curl -X GET \
|
curl -X GET \
|
||||||
http://localhost:3000/api/v1/task/<task_id>/status \
|
http://localhost:3000/api/v1/task/<task_id> \
|
||||||
-H 'x-api-key: <your_api_key>'
|
-H 'x-api-key: <your_api_key>'
|
||||||
```
|
```
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -20,6 +20,7 @@ from langflow.interface.custom.custom_component import CustomComponent
|
||||||
|
|
||||||
from langflow.api.v1.schemas import (
|
from langflow.api.v1.schemas import (
|
||||||
ProcessResponse,
|
ProcessResponse,
|
||||||
|
TaskResponse,
|
||||||
TaskStatusResponse,
|
TaskStatusResponse,
|
||||||
UploadFileResponse,
|
UploadFileResponse,
|
||||||
CustomComponentCode,
|
CustomComponentCode,
|
||||||
|
|
@ -145,9 +146,15 @@ async def process_flow(
|
||||||
session_id,
|
session_id,
|
||||||
)
|
)
|
||||||
task_result = task.status
|
task_result = task.status
|
||||||
|
|
||||||
|
if task_id:
|
||||||
|
task_response = TaskResponse(id=task_id, href=f"api/v1/task/{task_id}")
|
||||||
|
else:
|
||||||
|
task_response = None
|
||||||
|
|
||||||
return ProcessResponse(
|
return ProcessResponse(
|
||||||
result=task_result,
|
result=task_result,
|
||||||
id=task_id,
|
task=task_response,
|
||||||
session_id=session_id,
|
session_id=session_id,
|
||||||
backend=str(type(task_service.backend)),
|
backend=str(type(task_service.backend)),
|
||||||
)
|
)
|
||||||
|
|
@ -173,7 +180,7 @@ async def process_flow(
|
||||||
raise HTTPException(status_code=500, detail=str(e)) from e
|
raise HTTPException(status_code=500, detail=str(e)) from e
|
||||||
|
|
||||||
|
|
||||||
@router.get("/task/{task_id}/status", response_model=TaskStatusResponse)
|
@router.get("/task/{task_id}", response_model=TaskStatusResponse)
|
||||||
async def get_task_status(task_id: str):
|
async def get_task_status(task_id: str):
|
||||||
task_service = get_task_service()
|
task_service = get_task_service()
|
||||||
task = task_service.get_task(task_id)
|
task = task_service.get_task(task_id)
|
||||||
|
|
|
||||||
|
|
@ -47,11 +47,18 @@ class UpdateTemplateRequest(BaseModel):
|
||||||
template: dict
|
template: dict
|
||||||
|
|
||||||
|
|
||||||
|
class TaskResponse(BaseModel):
|
||||||
|
"""Task response schema."""
|
||||||
|
|
||||||
|
id: Optional[str] = Field(None)
|
||||||
|
href: Optional[str] = Field(None)
|
||||||
|
|
||||||
|
|
||||||
class ProcessResponse(BaseModel):
|
class ProcessResponse(BaseModel):
|
||||||
"""Process response schema."""
|
"""Process response schema."""
|
||||||
|
|
||||||
result: Any
|
result: Any
|
||||||
id: Optional[str] = None
|
task: Optional[TaskResponse] = None
|
||||||
session_id: Optional[str] = None
|
session_id: Optional[str] = None
|
||||||
backend: Optional[str] = None
|
backend: Optional[str] = None
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -19,7 +19,7 @@ class NameTest(FastHttpUser):
|
||||||
while True:
|
while True:
|
||||||
with self.rest(
|
with self.rest(
|
||||||
"GET",
|
"GET",
|
||||||
f"/task/{task_id}/status",
|
f"/task/{task_id}",
|
||||||
name="task_status",
|
name="task_status",
|
||||||
headers=self.headers,
|
headers=self.headers,
|
||||||
) as response:
|
) as response:
|
||||||
|
|
|
||||||
|
|
@ -25,10 +25,10 @@ def run_post(client, flow_id, headers, post_data):
|
||||||
|
|
||||||
|
|
||||||
# Helper function to poll task status
|
# Helper function to poll task status
|
||||||
def poll_task_status(client, headers, task_id, max_attempts=20, sleep_time=1):
|
def poll_task_status(client, headers, href, max_attempts=20, sleep_time=1):
|
||||||
for _ in range(max_attempts):
|
for _ in range(max_attempts):
|
||||||
task_status_response = client.get(
|
task_status_response = client.get(
|
||||||
f"api/v1/task/{task_id}/status",
|
href,
|
||||||
headers=headers,
|
headers=headers,
|
||||||
)
|
)
|
||||||
if (
|
if (
|
||||||
|
|
@ -544,11 +544,15 @@ def test_async_task_processing(distributed_client, added_flow, created_api_key):
|
||||||
assert response.status_code == 200, response.json()
|
assert response.status_code == 200, response.json()
|
||||||
|
|
||||||
# Extract the task ID from the response
|
# Extract the task ID from the response
|
||||||
task_id = response.json().get("id")
|
task = response.json().get("task")
|
||||||
|
task_id = task.get("id")
|
||||||
|
task_href = task.get("href")
|
||||||
assert task_id is not None
|
assert task_id is not None
|
||||||
|
assert task_href is not None
|
||||||
|
assert task_href == f"api/v1/task/{task_id}"
|
||||||
|
|
||||||
# Polling the task status using the helper function
|
# Polling the task status using the helper function
|
||||||
task_status_json = poll_task_status(distributed_client, headers, task_id)
|
task_status_json = poll_task_status(distributed_client, headers, task_href)
|
||||||
assert task_status_json is not None, "Task did not complete in time"
|
assert task_status_json is not None, "Task did not complete in time"
|
||||||
|
|
||||||
# Validate that the task completed successfully and the result is as expected
|
# Validate that the task completed successfully and the result is as expected
|
||||||
|
|
@ -576,11 +580,15 @@ def test_async_task_processing_vector_store(
|
||||||
assert "FAILURE" not in response.json()["result"]
|
assert "FAILURE" not in response.json()["result"]
|
||||||
|
|
||||||
# Extract the task ID from the response
|
# Extract the task ID from the response
|
||||||
task_id = response.json().get("id")
|
task = response.json().get("task")
|
||||||
|
task_id = task.get("id")
|
||||||
|
task_href = task.get("href")
|
||||||
assert task_id is not None
|
assert task_id is not None
|
||||||
|
assert task_href is not None
|
||||||
|
assert task_href == f"api/v1/task/{task_id}"
|
||||||
|
|
||||||
# Polling the task status using the helper function
|
# Polling the task status using the helper function
|
||||||
task_status_json = poll_task_status(client, headers, task_id, max_attempts=40)
|
task_status_json = poll_task_status(client, headers, task_href)
|
||||||
assert task_status_json is not None, "Task did not complete in time"
|
assert task_status_json is not None, "Task did not complete in time"
|
||||||
|
|
||||||
# Validate that the task completed successfully and the result is as expected
|
# Validate that the task completed successfully and the result is as expected
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue