1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
|
import json
from typing import Annotated, Any
import requests
from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi.responses import HTMLResponse
from generalresearch.redis_helper import RedisConfig
from starlette.responses import RedirectResponse
from jb.api.auth import get_authenticated_user
from jb.config import JB_EVENTS_STREAM, settings
from jb.decorators import get_redis_config
from jb.flow.monitoring import emit_mturk_notification_event
from jb.models.auth import User
from jb.models.event import MTurkEvent
from jb.settings import render_base_html
common_router = APIRouter(prefix="", tags=["API"], include_in_schema=True)
@common_router.get(path="/work/", response_class=HTMLResponse)
async def work(request: Request):
"""
HTML Page that the worker lands on in an iFrame.
They can either be previewing the HIT, or have already accepted it.
"""
amt_assignment_id = request.query_params.get("assignmentId", None)
worker_id = request.query_params.get("workerId", None)
amt_hit_id = request.query_params.get("hitId", None)
print(f"work: {amt_assignment_id=} {worker_id=} {amt_hit_id=}")
if (
not worker_id
or not amt_assignment_id
or amt_assignment_id == "ASSIGNMENT_ID_NOT_AVAILABLE"
):
return RedirectResponse(
url=f"/preview/?{request.url.query}" if request.url.query else "/preview/",
status_code=302,
)
return HTMLResponse(render_base_html())
RedisConfigDep = Annotated[RedisConfig, Depends(get_redis_config)]
@common_router.post(path=f"/{settings.sns_path}/", include_in_schema=False)
async def mturk_notifications(request: Request, redis_config: RedisConfigDep):
"""
Our SNS topic will POST to this endpoint whenever we get a new message
"""
try:
message = await request.json()
except Exception:
raise HTTPException(status_code=500, detail="No POST data")
match message.get("Type"):
case "SubscriptionConfirmation":
subscribe_url = message["SubscribeURL"]
print("Confirming SNS subscription...")
requests.get(subscribe_url)
case "Notification":
msg = json.loads(message["Message"])
print("Received MTurk event:", msg)
enqueue_mturk_notifications(msg=msg, redis_config=redis_config)
case _:
raise HTTPException(status_code=500, detail="Invalid JSON")
return {"status": "ok"}
def enqueue_mturk_notifications(msg: dict[str, Any], redis_config: RedisConfig) -> None:
redis_client = redis_config.create_redis_client()
for evt in msg["Events"]:
event = MTurkEvent.from_sns(evt)
emit_mturk_notification_event(
event_type=event.event_type, amt_hit_type_id=event.amt_hit_type_id
)
print("enqueue_mturk_notifications", event.model_dump_json())
redis_client.xadd(JB_EVENTS_STREAM, {"data": event.model_dump_json()})
@common_router.get(path="/work/direct/", response_class=HTMLResponse)
async def work_direct(
request: Request,
user: Annotated[User, Depends(get_authenticated_user)],
):
"""
View for direct work (not on AMT).
Makes sure user is authenticated.
"""
# todo: emit event
return HTMLResponse(render_base_html())
|
