Home

ping()

GET: /ping

Endpoint to check if the server is running.

Returns:

Name	Type	Description
Response		Response with status 200 if the server is running.

Source code in app.py

@app.get("/ping")
def ping():
    """
    # GET: /ping

    Endpoint to check if the server is running.

    Returns:
        Response: Response with status 200 if the server is running.
    """
    try:
        client = grpcclient.InferenceServerClient(
            url=config.grcp_model_server_address, verbose=False
        )
        return Response(status_code=200)
    except Exception:
        return Response(status_code=400)

predict_bucket(input_location=Header(None), inference_parameters=Header(None), webhook_url=Header(None), write_to_gcs=Header(False), input_bucket_name=Header(None), output_bucket_name=Header(None), examination_id=Header(None))

POST: /bucket_invocations

Endpoint to process an image and send it to the inference server.

Headers

Input-Location: Location of the image in the GCS bucket. Webhook-Url: URL to send the results of the inference. Write-To-GCS: Bool flag to write the results to a GCS bucket. False by default. Input-Bucket-Name: Name of the input bucket. Output-Bucket-Name: Name of the output bucket. Examination-ID: ID of the examination, used to track the request results. Inference-Parameters: Parameters to send to the inference server. JSON string with the following keys

- nerve_zone_landmarks: optional, landmarks of the nerve zone returned by retinal_app

- nerve_zone_slice_indices: optional, slice indices of the nerve zone returned by retinal_app

- mm_crop_zone: how much to crop from the center of the image.

- mm_crop_zone_nerve: how much to crop from the center of the image for nerve zone.

- exam_center_coordinate: center coordinates of the image (obtained from fovea center model).

- slice_thickness: slice thickness parameter of the exam.

- pixel_spacing_column: pixel spacing column parameter of the exam.

- type_of_scan: type of scan, should be macula, widescan, optic_disk

- zone_of_interest: zone of interest to process. Could be "fovea", "nerve"

- scan_protocol: scan protocol, should be "VERTICAL_3D", "HORIZONTAL_3D", "UNKNOWN"

- num_slices: number of slices in the exam

Returns:

Name	Type	Description
		JSON with the results of the inference:
filename		Name of the file that was processed. >1 if multiple files.
status		Status of the request. Can be "sent" or "error".
result_path		Path to the result in the GCS bucket. >1 if multiple files.
request_uuid		UUID of the request, generated by the server. Used to track the request results. >1 if multiple files, in correspondence with the filename.

Raises:

Type	Description
Response	Error response if the content type is not supported.

Source code in app.py

@app.post("/bucket_invocations")
def predict_bucket(
    input_location: str = Header(None),
    inference_parameters: str = Header(None),
    webhook_url: str = Header(None),
    write_to_gcs: bool = Header(False),
    input_bucket_name: str = Header(None),
    output_bucket_name: str = Header(None),
    examination_id: str = Header(None),
):
    """
    # POST: /bucket_invocations

    Endpoint to process an image and send it to the inference server.

    Headers:
        *Input-Location*: Location of the image in the GCS bucket.
        *Webhook-Url*: URL to send the results of the inference.
        *Write-To-GCS*: Bool flag to write the results to a GCS bucket. False by default.
        *Input-Bucket-Name*: Name of the input bucket.
        *Output-Bucket-Name*: Name of the output bucket.
        *Examination-ID*: ID of the examination, used to track the request results.
        *Inference-Parameters*: Parameters to send to the inference server. JSON string with the following keys

            - nerve_zone_landmarks: optional, landmarks of the nerve zone returned by retinal_app

            - nerve_zone_slice_indices: optional, slice indices of the nerve zone returned by retinal_app

            - mm_crop_zone: how much to crop from the center of the image.

            - mm_crop_zone_nerve: how much to crop from the center of the image for nerve zone.

            - exam_center_coordinate: center coordinates of the image (obtained from fovea center model).

            - slice_thickness: slice thickness parameter of the exam.

            - pixel_spacing_column: pixel spacing column parameter of the exam.

            - type_of_scan: type of scan, should be macula, widescan, optic_disk

            - zone_of_interest: zone of interest to process. Could be "fovea", "nerve"

            - scan_protocol: scan protocol, should be "VERTICAL_3D", "HORIZONTAL_3D", "UNKNOWN"

            - num_slices: number of slices in the exam


    Returns:
        JSON with the results of the inference:
        filename: Name of the file that was processed. >1 if multiple files.
        status: Status of the request. Can be "sent" or "error".
        result_path: Path to the result in the GCS bucket. >1 if multiple files.
        request_uuid: UUID of the request, generated by the server. Used to track the request results. >1 if multiple files, in correspondence with the filename.

    Raises:
        Response: Error response if the content type is not supported.
    """

    webhook_response = _check_webhook(webhook_url, examination_id)
    if webhook_response.status_code != 200:
        return webhook_response

    # get input and output locations
    input_bucket_to_use = (
        input_bucket_name if input_bucket_name is not None else config.input_bucket_name
    )
    output_bucket_to_use = (
        output_bucket_name if output_bucket_name is not None else config.output_bucket_name
    )

    start = time.time()
    images = asyncio.run(
        _read_from_gcp_bucket_async(input_bucket_to_use, input_location, examination_id, logger)
    )
    elapsed = time.time() - start

    logger.info(json.dumps({
        "status": "INFO",
        "message": f"Read {len(images)} images from GCP bucket {input_bucket_to_use}/{input_location}. Took {elapsed} seconds.",
        "examination_id": examination_id
    }))

    try:
        request_uuids = []
        result_paths = []
        filenames = []
        statuses = []

        for filename, image in images:
            client = grpcclient.InferenceServerClient(
                url=config.grcp_model_server_address,
                verbose=False,
                channel_args=(("grpc.lb_policy_name", "round_robin"),),
            )
            model_config = client.get_model_config(
                model_name=model_name, model_version=model_version, as_json=True
            )["config"]
            img = image
            img = img[np.newaxis, ..., np.newaxis].astype(np.float32)

            inputs = [
                grpcclient.InferInput("IMAGE", img.shape, np_to_triton_dtype(img.dtype)),
                grpcclient.InferInput("INPUT_JSON", (1, 1), "BYTES"),
            ]

            inputs[0].set_data_from_numpy(img)

            slice_idx = int(filename.split(".")[1])
            inference_params = inference_parameters.replace("'", '"')
            dict_inference_parameters = json.loads(inference_params)
            dict_inference_parameters["slice_idx"] = slice_idx

            zone_to_run = define_zone_to_run(dict_inference_parameters)
            request_uuid = str(uuid.uuid4())
            if zone_to_run is not None:
                dict_inference_parameters["zone_of_interest"] = zone_to_run
                inference_params = json.dumps(dict_inference_parameters)
                inputs[1].set_data_from_numpy(np.array([[inference_params]] * 1, dtype=np.object_))

                outputs = [
                    grpcclient.InferRequestedOutput(model_config["output"][i]["name"])
                    for i in range(len(model_config["output"]))
                ]

                request_uuids.append(request_uuid)
                result_paths.append(
                    f"{output_bucket_to_use}/{config.output_folder_name}/{request_uuid}.json"
                )
                filenames.append(filename)

                statuses.append("sent")

                response = client.async_infer(
                    model_name=model_name,
                    model_version=model_version,
                    inputs=inputs,
                    outputs=outputs,
                    callback=partial(
                        result_image_bucket_callback,
                        model_config=model_config,
                        filename=filename,
                        request_uuid=request_uuid,
                        client=client,
                        webhook_url=webhook_url,
                        write_to_gcs=write_to_gcs,
                        output_bucket_name=output_bucket_to_use,
                        examination_id=examination_id,
                    ),
                )
            else:
                filenames.append(filename)
                request_uuids.append(request_uuid)
                result_paths.append(None)
                statuses.append("no_zone_to_run")

        return JSONResponse(
            content={
                "filename": filenames if len(filenames) > 1 else filenames[0],
                "request_uuid": request_uuids if len(request_uuids) > 1 else request_uuids[0],
                "result_path": result_paths if len(result_paths) > 1 else result_paths[0],
                "status": statuses if len(statuses) > 1 else statuses[0],
                "examination_id": examination_id,
            },
            status_code=200,
        )

    except Exception as e:
        return JSONResponse(
            content={"error": str(e), "examination_id": examination_id}, status_code=400
        )

predict_bucket_azure_uae(input_location=Header(None), inference_parameters=Header(None), webhook_url=Header(None), write_to_gcs=Header(False), input_bucket_name=Header(None), output_bucket_name=Header(None), examination_id=Header(None))

POST: /bucket_invocations

Endpoint to process an image and send it to the inference server.

Headers

Input-Location: Location of the image in the Azure Blob bucket. Webhook-Url: URL to send the results of the inference. Write-To-GCS: Bool flag to write the results to a GCS bucket. False by default. Input-Bucket-Name: Name of the input bucket. Output-Bucket-Name: Name of the output bucket. Examination-ID: ID of the examination, used to track the request results. Inference-Parameters: Parameters to send to the inference server. JSON string with the following keys

- scan_width: width of the scan window.

- mm_crop_zone: how much to crop from the center of the image.

- exam_center_coordinate: center coordinates of the image (obtained from fovea center model).

- pixel_spacing_column: pixel spacing column parameter of the exam.

Returns:

Name	Type	Description
		JSON with the results of the inference:
filename		Name of the file that was processed. >1 if multiple files.
status		Status of the request. Can be "sent" or "error".
result_path		Path to the result in the GCS bucket. >1 if multiple files.
request_uuid		UUID of the request, generated by the server. Used to track the request results. >1 if multiple files, in correspondence with the filename.

Raises:

Type	Description
Response	Error response if the content type is not supported.

Source code in app.py

@app.post("/bucket_invocations_azure_uae")
def predict_bucket_azure_uae(
    input_location: str = Header(None),
    inference_parameters: str = Header(None),
    webhook_url: str = Header(None),
    write_to_gcs: bool = Header(False),
    input_bucket_name: str = Header(None),
    output_bucket_name: str = Header(None),
    examination_id: str = Header(None),
):
    """
    # POST: /bucket_invocations

    Endpoint to process an image and send it to the inference server.

    Headers:
        *Input-Location*: Location of the image in the Azure Blob bucket.
        *Webhook-Url*: URL to send the results of the inference.
        *Write-To-GCS*: Bool flag to write the results to a GCS bucket. False by default.
        *Input-Bucket-Name*: Name of the input bucket.
        *Output-Bucket-Name*: Name of the output bucket.
        *Examination-ID*: ID of the examination, used to track the request results.
        *Inference-Parameters*: Parameters to send to the inference server. JSON string with the following keys

            - scan_width: width of the scan window.

            - mm_crop_zone: how much to crop from the center of the image.

            - exam_center_coordinate: center coordinates of the image (obtained from fovea center model).

            - pixel_spacing_column: pixel spacing column parameter of the exam.

    Returns:
        JSON with the results of the inference:
        filename: Name of the file that was processed. >1 if multiple files.
        status: Status of the request. Can be "sent" or "error".
        result_path: Path to the result in the GCS bucket. >1 if multiple files.
        request_uuid: UUID of the request, generated by the server. Used to track the request results. >1 if multiple files, in correspondence with the filename.

    Raises:
        Response: Error response if the content type is not supported.
    """

    webhook_response = _check_webhook(webhook_url, examination_id)
    if webhook_response.status_code != 200:
        return webhook_response

    # get input and output locations
    input_bucket_to_use = (
        input_bucket_name if input_bucket_name is not None else config.input_bucket_name
    )
    output_bucket_to_use = (
        output_bucket_name if output_bucket_name is not None else config.output_bucket_name
    )

    images = _read_from_azure_uae_bucket(input_bucket_to_use, input_location)

    try:
        # Process the image contents here (e.g., save it, analyze it, etc.)
        response = _process_images(
            images,
            inference_parameters,
            output_bucket_to_use,
            webhook_url,
            write_to_gcs,
            examination_id,
        )
        return JSONResponse(
            content={
                "filename": response["filenames"],
                "request_uuid": response["request_uuids"],
                "result_path": response["result_paths"],
                "status": "sent",
                "examination_id": examination_id,
            },
            status_code=200,
        )
    except Exception as e:
        return JSONResponse(
            content={"error": str(e), "examination_id": examination_id}, status_code=400
        )

predict_image(image=File(...), inference_parameters=Header(None), webhook_url=Header(None), examination_id=Header(None))

POST: /invocations

Endpoint to process an image and send it to the inference server.

Parameters:

Name	Type	Description	Default
image	UploadFile	Image file to process (in the request body).	File(...)

Headers

Inference-Parameters: Parameters to send to the inference server. JSON string with the following keys:

- slice_idx: index of the slice to process.

- nerve_zone_landmarks: optional, landmarks of the nerve zone returned by retinal_app

- nerve_zone_slice_indices: optional, slice indices of the nerve zone returned by retinal_app

- mm_crop_zone: how much to crop from the center of the image.

- mm_crop_zone_nerve: how much to crop from the center of the image for nerve zone.

- exam_center_coordinate: center coordinates of the image (obtained from fovea center model).

- slice_thickness: slice thickness parameter of the exam.

- pixel_spacing_column: pixel spacing column parameter of the exam.

- zone_of_interest: zone of interest to process. Could be "fovea", "nerve"

- num_slices: number of slices in the exam

- type_of_scan: type of scan, should be macula, widescan, optic_disk

- scan_protocol: scan protocol, should be "VERTICAL_3D", "HORIZONTAL_3D", "UNKNOWN"

Content-Type: Type of the image. Can be "image/jpeg", "image/png", "image/tiff", "image/bmp", "image/jpg". Webhook-URL: URL to send the results of the inference. Examination-ID: ID of the examination, used to track the request results.

Returns:

Type	Description
	JSON with the results of the inference:
	filename: Name of the file that was processed.
	status: Status of the request. Can be "sent" or "error".
	request_uuid: UUID of the request, generated by the server. Used to track the request results.

Raises:

Type	Description
Response	Error response if the content type is not supported.

Source code in app.py

@app.post("/invocations")
def predict_image(
    image: UploadFile = File(...),
    inference_parameters: str = Header(None),
    webhook_url: str = Header(None),
    examination_id: str = Header(None),
):
    """
    # POST: /invocations

    Endpoint to process an image and send it to the inference server.

    Args:
        image (UploadFile): Image file to process (in the request body).

    Headers:
        *Inference-Parameters*: Parameters to send to the inference server. JSON string with the following keys:

            - slice_idx: index of the slice to process.

            - nerve_zone_landmarks: optional, landmarks of the nerve zone returned by retinal_app

            - nerve_zone_slice_indices: optional, slice indices of the nerve zone returned by retinal_app

            - mm_crop_zone: how much to crop from the center of the image.

            - mm_crop_zone_nerve: how much to crop from the center of the image for nerve zone.

            - exam_center_coordinate: center coordinates of the image (obtained from fovea center model).

            - slice_thickness: slice thickness parameter of the exam.

            - pixel_spacing_column: pixel spacing column parameter of the exam.

            - zone_of_interest: zone of interest to process. Could be "fovea", "nerve"

            - num_slices: number of slices in the exam

            - type_of_scan: type of scan, should be macula, widescan, optic_disk

            - scan_protocol: scan protocol, should be "VERTICAL_3D", "HORIZONTAL_3D", "UNKNOWN"

        *Content-Type*: Type of the image. Can be "image/jpeg", "image/png", "image/tiff", "image/bmp", "image/jpg".
        *Webhook-URL*: URL to send the results of the inference.
        *Examination-ID*: ID of the examination, used to track the request results.

    Returns:
        JSON with the results of the inference:
        - filename: Name of the file that was processed.
        - status: Status of the request. Can be "sent" or "error".
        - request_uuid: UUID of the request, generated by the server. Used to track the request results.

    Raises:
        Response: Error response if the content type is not supported.
    """

    client = grpcclient.InferenceServerClient(
        url=config.grcp_model_server_address,
        verbose=False,
        channel_args=(("grpc.lb_policy_name", "round_robin"),),
    )  # , concurrency=1, connection_timeout=10)
    model_config = client.get_model_config(
        model_name=config.model_name, model_version=config.model_version, as_json=True
    )["config"]

    content_type = image.content_type

    webhook_response = _check_webhook(webhook_url, examination_id)
    if webhook_response.status_code != 200:
        return webhook_response

    if content_type not in config.available_content_types:
        return Response(
            status=415,
            content="Cannot decode image data. Is content_type correct?",
            media_type="text/plain",
        )

    try:
        inference_params = inference_parameters.replace("'", '"')

        inference_params_dict = json.loads(inference_params)
        zone_to_run = define_zone_to_run(inference_params_dict)
        request_uuid = str(uuid.uuid4())
        if zone_to_run is not None:
            contents = image.file.read()

            image_bytes = np.frombuffer(contents, dtype=np.uint8)

            img = cv2.imdecode(image_bytes, cv2.IMREAD_GRAYSCALE)
            img = img[np.newaxis, ..., np.newaxis].astype(np.float32)

            inputs = [
                grpcclient.InferInput("IMAGE", img.shape, np_to_triton_dtype(img.dtype)),
                grpcclient.InferInput("INPUT_JSON", (1, 1), "BYTES"),
            ]
            inputs[0].set_data_from_numpy(img)

            inference_params_dict["zone_of_interest"] = zone_to_run
            inference_params = json.dumps(inference_params_dict)

            inputs[1].set_data_from_numpy(np.array([[inference_params]] * 1, dtype=np.object_))

            outputs = [
                grpcclient.InferRequestedOutput(model_config["output"][i]["name"])
                for i in range(len(model_config["output"]))
            ]

            response = client.async_infer(
                model_name=config.model_name,
                model_version=config.model_version,
                inputs=inputs,
                outputs=outputs,
                callback=partial(
                    result_callback,
                    model_config=model_config,
                    filename=image.filename,
                    request_uuid=request_uuid,
                    client=client,
                    webhook_url=webhook_url,
                    examination_id=examination_id,
                ),
            )

            return JSONResponse(
                content={
                    "filename": image.filename,
                    "status": "sent",
                    "request_uuid": request_uuid,
                },
                status_code=200,
            )
        else:
            return JSONResponse(
                content={
                    "filename": image.filename,
                    "message": "No zone to run",
                    "status": "no_zone_to_run",
                    "request_uuid": request_uuid,
                },
                status_code=200,
            )
    except Exception as e:
        return JSONResponse(
            content={
                "filename": image.filename,
                "message": str(e),
                "status": "error",
                "request_uuid": request_uuid,
            },
            status_code=400,
        )

read_from_gcp_bucket(input_bucket, prefix)

Function to read images from a GCP bucket.

Parameters:

Name	Type	Description	Default
prefix	str	Prefix to search for images in the bucket.	required

Returns:

Name	Type	Description
List	Tuple[str, ndarray]	List of images read from the bucket.

Source code in app.py

def read_from_gcp_bucket(input_bucket: str, prefix: str) -> Tuple[str, np.ndarray]:
    """
    Function to read images from a GCP bucket.

    Args:
        prefix (str): Prefix to search for images in the bucket.

    Returns:
        List: List of images read from the bucket.
    """

    # make client
    client = storage.Client()

    # get bucket
    bucket = client.bucket(input_bucket)

    # get blobs
    blobs = bucket.list_blobs(prefix=prefix)

    images = []
    for blob in blobs:
        # read image
        image = cv2.imdecode(
            np.frombuffer(blob.download_as_string(), dtype=np.uint8), cv2.IMREAD_GRAYSCALE
        )
        images.append((blob.name, image))

    return images

result_callback(model_config, filename, request_uuid, result, error, client, webhook_url, examination_id)

Callback function to process the result of the inference request. Sends a webhook action to the callback service.

Parameters:

Name	Type	Description	Default
model_config	dict	Model configuration dictionary.	required
filename	str	Name of the file that was processed.	required
initial_resolution	tuple	Initial resolution of the image.	required
result	list	List of output tensors.	required
error	Exception	Error that occurred during the request.	required
request_uuid	str	UUID of the request.	required
client	object	Triton client object.	required
webhook_url	str	URL of the webhook service.	required
examination_id	str	ID of the examination.	required

Source code in app.py

def result_callback(
    model_config: dict,
    filename: str,
    request_uuid: str,
    result: Optional[list],
    error: Optional[Exception],
    client: object,
    webhook_url: str,
    examination_id: str,
) -> None:
    """
    Callback function to process the result of the inference request.
    Sends a webhook action to the callback service.

    Args:
        model_config (dict): Model configuration dictionary.
        filename (str): Name of the file that was processed.
        initial_resolution (tuple): Initial resolution of the image.
        result (list): List of output tensors.
        error (Exception): Error that occurred during the request.
        request_uuid (str): UUID of the request.
        client (object): Triton client object.
        webhook_url (str): URL of the webhook service.
        examination_id (str): ID of the examination.
    """

    if error is None:
        output_data = result.as_numpy(model_config["output"][0]["name"])[0]
        status_message = {
            "id": request_uuid,
            "status": "COMPLETED",
            "output": json.loads(output_data),
            "filename": filename,
        }
        response = requests.post(webhook_url, json=status_message)

        if response.status_code == 200:
            log_message = json.dumps(
                {
                    "model": model_config["name"],
                    "examination_id": examination_id,
                    "status": "COMPLETED",
                    "filename": filename,
                    "request_uuid": request_uuid,
                }
            )

            logger.info(log_message)
        else:
            log_message = json.dumps(
                {
                    "model": model_config["name"],
                    "examination_id": examination_id,
                    "status": "ERROR",
                    "filename": filename,
                    "request_uuid": request_uuid,
                    "error": f"Error sending webhook action for {filename, request_uuid}",
                }
            )
            logger.error(log_message)
    else:
        status_message = {
            "id": request_uuid,
            "status": "FAILED",
            "error": str(error),
            "filename": filename,
        }
        response = requests.post(webhook_url, json=status_message)

        if response.status_code == 200:
            log_message = json.dumps(
                {
                    "model": model_config["name"],
                    "examination_id": examination_id,
                    "status": "FAILED",
                    "filename": filename,
                    "request_uuid": request_uuid,
                    "error": str(error),
                }
            )
            logger.error(log_message)
        else:
            log_message = json.dumps(
                {
                    "model": model_config["name"],
                    "examination_id": examination_id,
                    "status": "ERROR",
                    "filename": filename,
                    "request_uuid": request_uuid,
                    "error": f"Error sending webhook action for {filename, request_uuid}",
                }
            )
            logger.error(log_message)

    client.close()

result_image_bucket_callback(model_config, filename, result, error, client, request_uuid, webhook_url, output_bucket_name, examination_id, write_to_gcs=False)

Callback function to process the result of the inference request. Writes the result to a GCS bucket.

Parameters:

Name	Type	Description	Default
model_config	dict	Model configuration dictionary.	required
filename	str	Name of the file that was processed.	required
initial_resolution	tuple	Initial resolution of the image.	required
result	list	List of output tensors.	required
error	Exception	Error that occurred during the request.	required
request_uuid	str	UUID of the request.	required
client	object	Triton client object.	required
webhook_url	str	URL of the webhook service.	required
write_to_gcs	bool	Flag to write the result to a GCS bucket.	False
output_bucket_name	str	Name of the output bucket.	required
examination_id	str	ID of the examination.	required

Source code in app.py

def result_image_bucket_callback(
    model_config: dict,
    filename: str,
    result: Optional[list],
    error: Optional[Exception],
    client: object,
    request_uuid: str,
    webhook_url: str,
    output_bucket_name: str,
    examination_id: str,
    write_to_gcs: bool = False,
) -> None:
    """
    Callback function to process the result of the inference request.
    Writes the result to a GCS bucket.

    Args:
        model_config (dict): Model configuration dictionary.
        filename (str): Name of the file that was processed.
        initial_resolution (tuple): Initial resolution of the image.
        result (list): List of output tensors.
        error (Exception): Error that occurred during the request.
        request_uuid (str): UUID of the request.
        client (object): Triton client object.
        webhook_url (str): URL of the webhook service.
        write_to_gcs (bool): Flag to write the result to a GCS bucket.
        output_bucket_name (str): Name of the output bucket.
        examination_id (str): ID of the examination.
    """
    if error is None:
        output_data = result.as_numpy(model_config["output"][0]["name"])[0].decode("utf-8")
        status_message = {
            "id": request_uuid,
            "status": "COMPLETED",
            "output": json.loads(output_data),
            "filename": filename,
        }
    else:
        output_data = {"error": str(error)}
        status_message = {
            "id": request_uuid,
            "status": "FAILED",
            "error": str(error),
            "filename": filename,
        }

    if write_to_gcs:
        write_json_to_gcs(
            output_bucket_name, output_data, f"{config.output_folder_name}/{request_uuid}.json"
        )

    client.close()

    response = requests.post(webhook_url, json=status_message)

    if response.status_code == 200:
        log_message = json.dumps(
            {
                "model": model_config["name"],
                "examination_id": examination_id,
                "status": "COMPLETED" if error is None else "FAILED",
                "filename": filename,
                "request_uuid": request_uuid,
            }
        )
        logger.info(log_message)
    else:
        log_message = json.dumps(
            {
                "model": model_config["name"],
                "examination_id": examination_id,
                "status": "ERROR",
                "filename": filename,
                "request_uuid": request_uuid,
                "error": f"Error sending webhook action for {filename, request_uuid}",
            }
        )
        logger.error(log_message)

write_json_to_gcs(bucket_name, json_data, output_path)

Write JSON data to a specific folder in a GCS bucket.

Parameters:

Name	Type	Description	Default
bucket_name	str	The name of the GCS bucket.	required
json_data	dict	The JSON data to be written.	required
output_path	str	The GCS path where the JSON file will be stored (e.g., 'folder/output_file.json').	required

Source code in app.py

def write_json_to_gcs(bucket_name: str, json_data: dict, output_path: str):
    """
    Write JSON data to a specific folder in a GCS bucket.

    Args:
        bucket_name (str): The name of the GCS bucket.
        json_data (dict): The JSON data to be written.
        output_path (str): The GCS path where the JSON file will be stored (e.g., 'folder/output_file.json').
    """
    # Convert the JSON data to a string
    json_str = json.dumps(json_data, indent=4)

    # Run the blocking GCS code in a thread pool
    _upload_to_gcs(json_str, bucket_name, output_path)
    log_message = json.dumps(
        {
            "model": config.model_name,
            "status": "COMPLETED",
            "filename": output_path,
        }
    )
    logger.info(log_message)