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
POSTrequest will be sent to the
callback_urlprovided 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()
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")
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
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
signature = base64.b64encode(digest).decode()
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:
# .. do processing
print("Body not validated")
Last modified 3mo ago