📗
Integrating Webhooks
Process Unitary's API responses in a scalable way
Webhooks are currently the best and fastest way to access results and act on them as they come.
Whenever a job finishes processing, a
POST
request will be sent to the callback_url
provided in the body of the job request. (please check the API Integration reference for more details on how to send a job request).The following Python (FastAPI) server will listen for results and log the jobs that are having at least one GARM category detected.
from fastapi import Request, FastAPI
app = FastAPI()
@app.post("/results")
async def get_body(request: Request):
result = await request.json()
if result["is_error"] is False:
garm_categories = result["result"].get("garm_categories", [])
if len(garm_categories) > 0:
print("GARM category detected")
return "OK"
There are currently two ways to set up the secret key:
- 1.Request for one to be generated and shared with you.
- 2.Set up a custom one and share it with us.
Please send the chosen option to your dedicated Account Manager.
In the following example, the payload is validated according to the secret key that has previously been set up.
from fastapi import Header, Request, FastAPI
import base64
import hashlib
import hmac
SECRET_KEY = "test_key"
app = FastAPI()
def compute_signature(payload: str, secret_key: str) -> str:
digest = hmac.new(
secret_key.encode(), msg=payload.encode("utf-8"), digestmod=hashlib.sha256
).digest()
signature = base64.b64encode(digest).decode()
return signature
@app.post("/results")
async def get_body(request: Request, x_hub_signature_256: str | None = Header(default=None)):
result = await request.body()
if compute_signature(result.decode(), SECRET_KEY) == x_hub_signature_256:
print("Body validated")
# .. do processing
else:
print("Body not validated")
return "OK"
Last modified 1mo ago