uktrade-hmrc-licensing-management-client

API client for HMRC internal licensing management API

Send licence to HMRC example

import requests.exceptions
from django.core.cache import cache

from hmrc_licensing_management import utils
from hmrc_licensing_management.api import Client, Licence, APIResponse, Result


class CachedClient(Client):
    """Example client used that stores the access token for as long as it's valid."""

    session_key = "CDS_CLIENT_GET_ACCESS_TOKEN_KEY"

    def get_access_token(self, *, scope: str | None = None) -> utils.AccessToken:
        # Look for access token in cache
        token = cache.get(self.session_key, None)

        if not token:
            # Fetch a new token
            token = super().get_access_token(scope=scope)

            # Store token in cache for as long as it's valid
            cache.add(self.session_key, token, timeout=token["expires_in"])

        return token


def example_implementation():
    # 1. Create client (use example class above to cache access token)
    client = CachedClient(
        # Production or sandbox hmrc API url
        hmrc_api_base_url="https://test-api.service.hmrc.gov.uk",
        # Client id and secret retrieved from HMRC API application
        hmrc_api_client_id="test-client-id",
        hmrc_api_client_secret="test-client-secret",
        # TOTP secret required for a HMRC "Privileged application"
        hmrc_api_totp_secret="test-totp-secret",
    )

    # 2. Create a licence payload
    # Licence serializer details omitted as it will be specific to your licence type
    licence_reference = "GBSIL123456"
    licence_data = Licence(...)

    # 3. Send a licence to HMRC
    try:
        api_response: APIResponse = client.send_licence_details(
            licence_reference, licence_data
        )

        if api_response.result == Result.accepted:
            print("Handle accepted licence")
        else:
            print("handle rejected licence")

    # The client can raise HTTP errors with attached notes
    except requests.exceptions.HTTPError as e:
        # Handle any http errors
        ...

    # Handle any unknown errors not explicitly raised by the API client
    except Exception as e:
        ...

All available serializers to create a licence payload can be found here.

NOTE: Do not use serializers found in hmrc_licensing_management/api/_serializers.py.

They have been autogenerated from the OpenAPI spec and have been updated with extra validation.

The serializers have been autogenerated from the licensing management Open API specification.

Several useful constants and utility functions can be found here and here:

Usage data callback example

This library provides a class-based view to subclass for processing usage data.

This is the HMRC API it uses: https://developer.service.hmrc.gov.uk/api-documentation/docs/api/service/push-pull-notifications-api/1.0

The licence management api usage data payload is found here:

Shown below is an example implementation with the following:

• A model to store the incoming data
• A view that inherits from HMRCPushPullCallbackView and stores the data in HMRCPushPullNotification

import logging

from django.core.serializers.json import DjangoJSONEncoder
from django.db import models, transaction
from django.utils.decorators import method_decorator
from django.views.decorators.csrf import csrf_exempt

from hmrc_licensing_management.usage import (
    HMRCPushPullCallbackView,
    HMRCPushPullResponseItem,
    NotifyUsage,
)

logger = logging.getLogger(__name__)


#
# Model used to store the usage data (taken from ECIL)
#
class HMRCPushPullNotification(models.Model):
    """Stores data received from HMRC's push pull notification API.

    Example payload to store in model:
        {
        "notificationId": "1ed5f407-8096-40d1-87ef-9a2a103eeb85",  # /PS-IGNORE
        "boxId": "50dca3fc-c37c-4f03-b719-63571333624c",
        "messageContentType": "application/json",
        "message": "[PAYLOAD]",
        "status": "PENDING",
        "createdDateTime": "2020-06-01T10:20:23.160+0000"
        }
    """

    class MessageContentType(models.TextChoices):
        application_json = "application/json"
        application_xml = "application/xml"

    class Status(models.TextChoices):
        pending = "PENDING"
        failed = "FAILED"
        acknowledged = "ACKNOWLEDGED"

    #
    # Fields containing data from HMRC
    #
    notification_id = models.TextField(
        help_text="Unique identifier for a notification."
    )
    box_id = models.TextField(
        help_text="Unique identifier for a box the notification was sent to."
    )
    message_content_type = models.CharField(
        max_length=20,
        choices=MessageContentType.choices,
        help_text="Content type of the message.",
    )
    message = models.JSONField(
        help_text=(
            "The notification message defined by messageContentType (JSON or XML). "
            "If this is JSON then it will have been escaped. "
            "Details on the structure of this data can be found in the documentation for the HMRC "
            "API that created the notification."
        ),
        encoder=DjangoJSONEncoder,
    )
    status = models.CharField(
        max_length=20, choices=Status.choices, help_text="Status of the notification."
    )
    created_datetime = models.DateTimeField(
        help_text="ISO-8601 UTC date and time the notification was created."
    )

    #
    # Fields added for ECIL
    #
    received_at = models.DateTimeField(
        auto_now_add=True,
        help_text="Date and time the notification was received by ECIL.",
    )
    processed = models.BooleanField(
        default=False, help_text="Indicates if the notification was processed."
    )
    processed_at = models.DateTimeField(
        null=True,
        default=None,
        help_text="Date and time the notification was processed.",
    )


#
# View to store the incoming usage data.
#
@method_decorator(csrf_exempt, name="dispatch")
@method_decorator(transaction.atomic, name="post")
class LicenceDetailsUsageCallbackView(HMRCPushPullCallbackView):
    def process_payload(self, payload: HMRCPushPullResponseItem) -> None:
        """Process the incoming usage data payload from HMRC

        Notes from HMRC:
        Design your application to process duplicate push notifications as a single notification (idempotency)

        The push notification system is designed to send ‘At least once’ to guarantee delivery.
        In most cases, this means notifications will be sent once and successfully received.
        In rare cases of network disruption, messages may be sent more than once.
        You should design your application to process duplicate notifications as a single notification.
        This will prevent errors and provide a consistent outcome for your application and users.
        """

        message = NotifyUsage.model_validate_json(payload.message)

        record, created = (
            HMRCPushPullNotification.objects.select_for_update().get_or_create(
                defaults={
                    "box_id": payload.boxId,
                    "message_content_type": payload.messageContentType.value,
                    "message": message.model_dump(
                        exclude_none=True, exclude_unset=True
                    ),
                    "status": payload.status.value,
                    "created_datetime": payload.createdDateTime,
                },
                notification_id=payload.notificationId,
            )
        )

        if created:
            logger.info(
                "HMRCPushPullNotification record created. notification_id: %s",
                payload.notificationId,
            )
        else:
            logger.info(
                "HMRCPushPullNotification record ignored. notification_id: %s",
                payload.notificationId,
            )

        # Do something with the data
        # e.g. trigger a task to do something with the HMRCPushPullNotification record.
        process_hmrc_push_pull_notifications.delay()