setup webhook api route

.flake8

@@ -1,3 +1,3 @@
 [flake8]
 ignore = E501, E123
-exclude = __init__.py
+exclude = __init__.py, venv

app.py

@@ -1,4 +1,4 @@
-from controllers import ContractController, MeController, HealthController
+from controllers import ContractController, MeController, HealthController, DocusignWebhookController
 from fastapi import FastAPI, Request, Response, status
 from fastapi.responses import JSONResponse
 from time import strftime
@@ -22,13 +22,15 @@
 app.include_router(ContractController(auth).router)
 app.include_router(MeController(auth).router)
 app.include_router(HealthController(auth).router)
+app.include_router(DocusignWebhookController(auth).router)
 
 
 @app.middleware("http")
 async def after_request(request: Request, call_next: Callable[..., Awaitable[_StreamingResponse]]) -> Response:
     response: Response = await call_next(request)
     timestamp = strftime('[%Y-%b-%d %H:%M]')  # TODO this is defined in multiple spots. Make robust
     assert request.client, "Missing header data in request. No client information."
+    # TODO hide "key" query params
     logging.info('%s %s %s %s %s %s', timestamp, request.client.host, request.method, request.scope['type'], request.url, response.status_code)
     return response
 

config/env.py

@@ -20,3 +20,5 @@ def load_env(input: str) -> str:
 DOCUSIGN_IMPERSONATED_USER_ID: str = load_env("DOCUSIGN_IMPERSONATED_USER_ID")
 DOCUSIGN_PRIVATE_KEY: str = load_env("DOCUSIGN_PRIVATE_KEY")
 CONTRACT_TEMPLATE_ID: str = load_env("CONTRACT_TEMPLATE_ID")
+DOCUSIGN_QUERY_PARAM_KEY: str = load_env("WEBHOOK_QUERY_PARAM_KEY")
+DOCUSIGN_API_VERSION: str = load_env("DOCUSIGN_API_VERSION")

controllers/__init__.py

@@ -1,4 +1,5 @@
 from .contract import ContractController
 from .base_controller import BaseController
 from .health import HealthController
-from .me import MeController
+from .me import MeController
+from .docusign_webhook import DocusignWebhookController

controllers/docusign_webhook.py

@@ -0,0 +1,52 @@
+from .base_controller import BaseController
+from managers import DocusignWebhookManager
+from fastapi_cloudauth.cognito import Cognito
+from fastapi import status, Response
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel, Field
+from utilities.types.fields import DocusignWebhookEventEnum
+from datetime import datetime
+from typing import Dict, Any
+import logging
+from config.env import DOCUSIGN_API_VERSION
+
+
+class PostItem(BaseModel):
+    event: DocusignWebhookEventEnum
+    api_version: str = Field(alias="apiVersion", examples=['v2.1'])
+    url: str = Field(examples=["/restapi/v2.1/accounts/6f7fcdd0-bc7f-484b-8a15-ed3af04c16ff/envelopes/29e66716-238b-459e-a29d-5371be2bef80"])
+    retry_count: int = Field(alias="retryCount", examples=[0])
+    configuration_id: int = Field(alias="configurationId", examples=[10352224])
+    generated_date_time: datetime = Field(alias="generatedDateTime", examples=["2023-11-01T03:20:28.1366172Z"])
+    data: Dict[str, Any]
+
+
+class DocusignWebhookController(BaseController):
+
+    def __init__(self, auth: Cognito):  # type: ignore[no-any-unimported]
+        super().__init__(auth)
+        self.router.add_api_route("/docusign/webhook", self.post, methods=["POST"], response_model=None)
+
+    def post(self, key: str, item: PostItem) -> Response:
+        mgr = DocusignWebhookManager()
+        if not mgr.is_webhook_key(key):
+            return JSONResponse(
+                status_code=status.HTTP_403_FORBIDDEN,
+                content=None
+                )
+
+        if item.api_version != DOCUSIGN_API_VERSION:
+            logging.warn(f"Webhook API Version doesn't match. Current: {item.api_version}, Expected: {DOCUSIGN_API_VERSION}")
+
+        try:
+            mgr.handle_webhook_event(item.event, item.retry_count, item.generated_date_time, item.data)
+        except NotImplementedError:
+            return JSONResponse(
+                status_code=status.HTTP_501_NOT_IMPLEMENTED,
+                content=None
+                )
+
+        return JSONResponse(
+            status_code=status.HTTP_200_OK,
+            content=None
+        )

managers/__init__.py

@@ -1,2 +1,3 @@
 from .contract import ContractManager
-from .me import MeManager
+from .me import MeManager
+from .docusign_webhook import DocusignWebhookManager

managers/docusign_webhook.py

@@ -0,0 +1,142 @@
+from config.env import DOCUSIGN_QUERY_PARAM_KEY
+from utilities.types.fields import DocusignWebhookEventEnum
+from datetime import datetime
+from typing import Dict, Any
+
+
+class DocusignWebhookManager():
+
+    def is_webhook_key(self, key: str) -> bool:
+        return key == DOCUSIGN_QUERY_PARAM_KEY
+
+    def handle_webhook_event(self, event: DocusignWebhookEventEnum, retry_count: int, generated_date_time: datetime, data: Dict[str, Any]) -> None:
+        match event:
+            case DocusignWebhookEventEnum.envelope_created:
+                self._handle_envelope_created(retry_count, generated_date_time, data)
+            case DocusignWebhookEventEnum.envelope_sent:
+                self._handle_envelope_sent(retry_count, generated_date_time, data)
+            case DocusignWebhookEventEnum.envelope_delivered:
+                self._handle_envelope_delivered(retry_count, generated_date_time, data)
+            case DocusignWebhookEventEnum.envelope_completed:
+                self._handle_envelope_completed(retry_count, generated_date_time, data)
+            case DocusignWebhookEventEnum.envelope_purge:
+                self._handle_envelope_purge(retry_count, generated_date_time, data)
+            case DocusignWebhookEventEnum.envelope_resent:
+                self._handle_envelope_resent(retry_count, generated_date_time, data)
+            case DocusignWebhookEventEnum.envelope_corrected:
+                self._handle_envelope_corrected(retry_count, generated_date_time, data)
+            case DocusignWebhookEventEnum.envelope_discard:
+                self._handle_envelope_discard(retry_count, generated_date_time, data)
+            case DocusignWebhookEventEnum.envelope_voided:
+                self._handle_envelope_voided(retry_count, generated_date_time, data)
+            case DocusignWebhookEventEnum.envelope_deleted:
+                self._handle_envelope_deleted(retry_count, generated_date_time, data)
+            case DocusignWebhookEventEnum.envelope_declined:
+                self._handle_envelope_declined(retry_count, generated_date_time, data)
+            case DocusignWebhookEventEnum.recipient_sent:
+                self._handle_recipient_sent(retry_count, generated_date_time, data)
+            case DocusignWebhookEventEnum.recipient_auto_responded:
+                self._handle_recipient_auto_responded(retry_count, generated_date_time, data)
+            case DocusignWebhookEventEnum.recipient_delivered:
+                self._handle_recipient_delivered(retry_count, generated_date_time, data)
+            case DocusignWebhookEventEnum.recipient_completed:
+                self._handle_recipient_completed(retry_count, generated_date_time, data)
+            case DocusignWebhookEventEnum.recipient_declined:
+                self._handle_recipient_declined(retry_count, generated_date_time, data)
+            case DocusignWebhookEventEnum.recipient_authentication_failure:
+                self._handle_recipient_authentication_failure(retry_count, generated_date_time, data)
+            case DocusignWebhookEventEnum.recipient_resent:
+                self._handle_recipient_resent(retry_count, generated_date_time, data)
+            case DocusignWebhookEventEnum.recipient_reassign:
+                self._handle_recipient_reassign(retry_count, generated_date_time, data)
+            case DocusignWebhookEventEnum.recipient_finish_later:
+                self._handle_recipient_finish_later(retry_count, generated_date_time, data)
+            case DocusignWebhookEventEnum.recipient_delegate:
+                self._handle_recipient_delegate(retry_count, generated_date_time, data)
+            case _ as unknown:
+                raise Exception(f"Unknown docusign event '{unknown}'")
+
+    def _handle_envelope_created(self, retry_count: int, generated_date_time: datetime, data: Dict[str, Any]) -> None:
+        # TODO Here we have more info about the contract compared to our initial post route, so update contract and users DB
+        raise NotImplementedError
+
+    def _handle_envelope_sent(self, retry_count: int, generated_date_time: datetime, data: Dict[str, Any]) -> None:
+        # TODO Email is sent to signer/approver. Update contract state (and maybe users db if needed).
+        raise NotImplementedError
+
+    def _handle_envelope_delivered(self, retry_count: int, generated_date_time: datetime, data: Dict[str, Any]) -> None:
+        # An envelope has been signed from both parties but it's not done yet. We can do nothing here
+        return
+
+    def _handle_envelope_completed(self, retry_count: int, generated_date_time: datetime, data: Dict[str, Any]) -> None:
+        # TODO update contract state (and maybe users db if needed)
+        # TODO send contract PDF to google drive
+        raise NotImplementedError
+
+    def _handle_envelope_purge(self, retry_count: int, generated_date_time: datetime, data: Dict[str, Any]) -> None:
+        # TODO at this point we need to delete the contracts and any kind of it's history from our DB.
+        raise NotImplementedError
+
+    def _handle_envelope_resent(self, retry_count: int, generated_date_time: datetime, data: Dict[str, Any]) -> None:
+        # An envelope email is being sent to the signer/approver. We can do nothing here
+        return
+
+    def _handle_envelope_corrected(self, retry_count: int, generated_date_time: datetime, data: Dict[str, Any]) -> None:
+        # TODO update contract data (and maybe users db if needed)
+        raise NotImplementedError
+
+    def _handle_envelope_discard(self, retry_count: int, generated_date_time: datetime, data: Dict[str, Any]) -> None:
+        # TODO it is unknown when this event will occur
+        raise NotImplementedError
+
+    def _handle_envelope_voided(self, retry_count: int, generated_date_time: datetime, data: Dict[str, Any]) -> None:
+        # TODO update contract state (and maybe users db if needed)
+        raise NotImplementedError
+
+    def _handle_envelope_deleted(self, retry_count: int, generated_date_time: datetime, data: Dict[str, Any]) -> None:
+        # TODO update contract state (and maybe users db if needed). DO NOT DELETE IT. Purge deletes.
+        raise NotImplementedError
+
+    def _handle_envelope_declined(self, retry_count: int, generated_date_time: datetime, data: Dict[str, Any]) -> None:
+        # TODO update contract state (and maybe users db if needed)
+        raise NotImplementedError
+
+    def _handle_recipient_sent(self, retry_count: int, generated_date_time: datetime, data: Dict[str, Any]) -> None:
+        # An envelope email is being sent to the signer/approver. We can do nothing here
+        return
+
+    def _handle_recipient_auto_responded(self, retry_count: int, generated_date_time: datetime, data: Dict[str, Any]) -> None:
+        # TODO it is unknown when this event will occur
+        raise NotImplementedError
+
+    def _handle_recipient_delivered(self, retry_count: int, generated_date_time: datetime, data: Dict[str, Any]) -> None:
+        # Signer/approver has filled out all fields, but has not confirmed the contract yet. We can do nothing here
+        return
+
+    def _handle_recipient_completed(self, retry_count: int, generated_date_time: datetime, data: Dict[str, Any]) -> None:
+        # TODO Signer/approver has finished the contract. Update contract DB (and maybe users db if needed)
+        raise NotImplementedError
+
+    def _handle_recipient_declined(self, retry_count: int, generated_date_time: datetime, data: Dict[str, Any]) -> None:
+        # TODO update contract state (and maybe users db if needed)
+        raise NotImplementedError
+
+    def _handle_recipient_authentication_failure(self, retry_count: int, generated_date_time: datetime, data: Dict[str, Any]) -> None:
+        # TODO it is unknown when this event will occur
+        raise NotImplementedError
+
+    def _handle_recipient_resent(self, retry_count: int, generated_date_time: datetime, data: Dict[str, Any]) -> None:
+        # An envelope email is being sent to the signer/approver. We can do nothing here
+        return
+
+    def _handle_recipient_reassign(self, retry_count: int, generated_date_time: datetime, data: Dict[str, Any]) -> None:
+        # TODO update contract state and users db
+        raise NotImplementedError
+
+    def _handle_recipient_finish_later(self, retry_count: int, generated_date_time: datetime, data: Dict[str, Any]) -> None:
+        # TODO update contract state (and maybe users db if needed)
+        raise NotImplementedError
+
+    def _handle_recipient_delegate(self, retry_count: int, generated_date_time: datetime, data: Dict[str, Any]) -> None:
+        # TODO it is unknown when this event will occur
+        raise NotImplementedError

mypy.ini

@@ -1,6 +1,6 @@
 [mypy]
 python_version = 3.10
-exclude = lambdaFunctions/
+exclude = venv
 warn_unused_configs = True
 check_untyped_defs = True
 disallow_untyped_calls = True
@@ -32,4 +32,4 @@ ignore_missing_imports = True
 ignore_missing_imports = True
 
 [mypy-boto3]
-ignore_missing_imports = True
+ignore_missing_imports = True

utilities/types/fields.py

@@ -16,3 +16,28 @@ def phone_number(alias: str) -> int:
 class VendorTypeEnum(Enum):
     artist = 'artist'
     dealer = 'dealer'
+
+
+class DocusignWebhookEventEnum(Enum):
+    envelope_created = 'envelope-created'
+    envelope_sent = 'envelope-sent'
+    envelope_delivered = 'envelope-delivered'
+    envelope_completed = 'envelope-completed'
+    envelope_purge = 'envelope-purge'
+    envelope_resent = 'envelope-resent'
+    envelope_corrected = 'envelope-corrected'
+    envelope_discard = 'envelope-discard'
+    envelope_voided = 'envelope-voided'
+    envelope_deleted = 'envelope-deleted'
+    envelope_declined = 'envelope-declined'
+
+    recipient_sent = 'recipient-sent'
+    recipient_auto_responded = 'recipient-auto-responded'
+    recipient_delivered = 'recipient-delivered'
+    recipient_completed = 'recipient-completed'
+    recipient_declined = 'recipient-declined'
+    recipient_authentication_failure = 'recipient-authentication-failure'
+    recipient_resent = 'recipient-resent'
+    recipient_reassign = 'recipient-reassign'
+    recipient_finish_later = 'recipient-finish-later'
+    recipient_delegate = 'recipient-delegate'