|
|
@ -16,7 +16,7 @@ templates = Jinja2Templates(directory="project/users/templates")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def api_call(email: str): # pylint: disable=unused-argument
|
|
|
|
def api_call(email: str): # pylint: disable=unused-argument
|
|
|
|
# used for testing a failed api call
|
|
|
|
"""Used for testing a failed api call"""
|
|
|
|
if random.choice([0, 1]):
|
|
|
|
if random.choice([0, 1]):
|
|
|
|
raise Exception("random processing error")
|
|
|
|
raise Exception("random processing error")
|
|
|
|
|
|
|
|
|
|
|
@ -31,12 +31,21 @@ def form_example_get(request: Request):
|
|
|
|
|
|
|
|
|
|
|
|
@users_router.post("/form/")
|
|
|
|
@users_router.post("/form/")
|
|
|
|
def form_example_post(user_body: UserBody):
|
|
|
|
def form_example_post(user_body: UserBody):
|
|
|
|
|
|
|
|
"""Accepts form data, and relays request to the task queue.
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
Returns the task id, which can be used with the /task_status/
|
|
|
|
|
|
|
|
endpoint to monitor status of the submitted requests.
|
|
|
|
|
|
|
|
"""
|
|
|
|
task = sample_task.delay(user_body.email)
|
|
|
|
task = sample_task.delay(user_body.email)
|
|
|
|
return JSONResponse({"task_id": task.task_id})
|
|
|
|
return JSONResponse({"task_id": task.task_id})
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@users_router.get("/task_status/")
|
|
|
|
@users_router.get("/task_status/")
|
|
|
|
def task_status(task_id: str):
|
|
|
|
def task_status(task_id: str):
|
|
|
|
|
|
|
|
"""Check async task result.
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
Endpoint is used for XHR short polling to check the task status.
|
|
|
|
|
|
|
|
"""
|
|
|
|
task = AsyncResult(task_id)
|
|
|
|
task = AsyncResult(task_id)
|
|
|
|
state = task.state
|
|
|
|
state = task.state
|
|
|
|
|
|
|
|
|
|
|
@ -55,6 +64,7 @@ def task_status(task_id: str):
|
|
|
|
|
|
|
|
|
|
|
|
@users_router.post("/webhook_test/")
|
|
|
|
@users_router.post("/webhook_test/")
|
|
|
|
def webhook_test():
|
|
|
|
def webhook_test():
|
|
|
|
|
|
|
|
"""Endpoint simulating the synchronous handling of a webhook request to another service."""
|
|
|
|
if not random.choice([0, 1]):
|
|
|
|
if not random.choice([0, 1]):
|
|
|
|
# mimic an error
|
|
|
|
# mimic an error
|
|
|
|
raise Exception()
|
|
|
|
raise Exception()
|
|
|
@ -66,6 +76,10 @@ def webhook_test():
|
|
|
|
|
|
|
|
|
|
|
|
@users_router.post("/webhook_test_async/")
|
|
|
|
@users_router.post("/webhook_test_async/")
|
|
|
|
def webhook_test_async():
|
|
|
|
def webhook_test_async():
|
|
|
|
|
|
|
|
"""End point simulating async handling of a webhook request to another service.
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
curl -X POST http://localhost:8010/users/webhook_test_async/ -d {'data':'ping'}
|
|
|
|
|
|
|
|
"""
|
|
|
|
task = task_process_notification.delay()
|
|
|
|
task = task_process_notification.delay()
|
|
|
|
print(task.id)
|
|
|
|
print(task.id)
|
|
|
|
return "pong"
|
|
|
|
return "pong"
|
|
|
|