add webhook to docseal

enrollments-events/app/app.py

@@ -0,0 +1,69 @@
+from http import HTTPStatus
+from typing import Any
+
+import requests
+from aws_lambda_powertools import Logger, Tracer
+from aws_lambda_powertools.event_handler.api_gateway import (
+    APIGatewayHttpResolver,
+    Response,
+)
+from aws_lambda_powertools.logging import correlation_paths
+from aws_lambda_powertools.utilities.typing import LambdaContext
+from layercake.dynamodb import DynamoDBPersistenceLayer, KeyPair
+
+from boto3clients import dynamodb_client, s3_client
+from config import BUCKET_NAME, ENROLLMENT_TABLE
+from docseal import headers
+
+logger = Logger(__name__)
+tracer = Tracer()
+app = APIGatewayHttpResolver(enable_validation=True)
+dyn = DynamoDBPersistenceLayer(ENROLLMENT_TABLE, dynamodb_client)
+
+
+@app.post('/')
+@tracer.capture_method
+def postback():
+    json_body = app.current_event.json_body
+    enrollment_id = json_body['data']['name']
+    combined_document_url = json_body['data'].get('combined_document_url')
+
+    if not combined_document_url:
+        return Response(status_code=HTTPStatus.NO_CONTENT)
+
+    try:
+        r = requests.get(combined_document_url, headers=headers)
+        r.raise_for_status()
+
+        object_key = f'certs/{enrollment_id}.pdf'
+        s3_uri = f's3://{BUCKET_NAME}/{object_key}'
+
+        s3_client.put_object(
+            Bucket=BUCKET_NAME,
+            Key=object_key,
+            Body=r.content,
+            ContentType='application/pdf',
+        )
+
+        logger.debug(f'PDF uploaded successfully to {s3_uri}')
+    except requests.exceptions.RequestException as exc:
+        logger.exception(exc)
+        raise
+
+    dyn.update_item(
+        key=KeyPair(enrollment_id, '0'),
+        cond_expr='attribute_exists(sk)',
+        update_expr='SET cert.s3_uri = :s3_uri',
+        expr_attr_values={':s3_uri': s3_uri},
+    )
+
+    return Response(status_code=HTTPStatus.NO_CONTENT)
+
+
+@logger.inject_lambda_context(correlation_id_path=correlation_paths.API_GATEWAY_HTTP)
+@tracer.capture_lambda_handler
+def lambda_handler(
+    event: dict[str, Any],
+    context: LambdaContext,
+) -> dict[str, Any]:
+    return app.resolve(event, context)