src.offers.ssi_agent_offers_client_adapter

  1"""SSI-Agent Adapter for offer operations."""
  2
  3from dataclasses import asdict
  4from typing import override
  5from attr import dataclass
  6import msgspec
  7import requests
  8
  9from src.awards.awards_service_port import Award
 10from src.offers.models import Offer
 11from src.offers.offers_client_port import OfferNotFound
 12from src.offers.offers_client_port import OffersClientError
 13from src.offers.offers_client_port import OffersClientPort
 14
 15@dataclass
 16class _CredentialOffer:
 17    credential_issuer: str
 18    credential_configuration_ids: list[str]
 19    grants: dict[str, dict[str, str]]
 20
 21@dataclass
 22class _SsiAgentOfferResponse:
 23   """SSI agent response on fetching a single offer from the admin api"""
 24   id: str
 25   grant_types: list[str]
 26   credential_offer_uri: dict[str, str]
 27   credential_offer: dict[str, _CredentialOffer]
 28   subject_id: str | None
 29   credential_ids: list[str]
 30   form_url_encoded_credential_offer: str
 31   pre_authorized_code: str
 32   credential_response: str | None
 33   status: str
 34   tx_code: str | None
 35   delivery_options: str | None
 36
 37class SsiAgentOffersClientAdapter(OffersClientPort):
 38    """Adapter for SSI Agent offers API."""
 39
 40    _ssi_agent_admin_base_url: str
 41    _timeout: int
 42
 43    def __init__(
 44        self,
 45        ssi_agent_url: str,
 46    ) -> None:
 47        """Initialize the adapter.
 48
 49        Args:
 50            ssi_agent_url: The admin base URL of the SSI agent.
 51            requests_client: The HTTP client to use for requests.
 52        """
 53        self._ssi_agent_admin_base_url = ssi_agent_url.rstrip("/")
 54        self._timeout = 10
 55
 56    @override
 57    def create(self, offer_id: str) -> str:
 58        """Create an offer in the SSI agent.
 59
 60        Args:
 61            offer_id: The offer identifier to create.
 62
 63        Returns:
 64            The credential offer URI.
 65        """
 66
 67        # TODO: subject must be the Award, that must be passed in
 68        award = Award()
 69        self._create_credential_for_subject(offer_id, award)
 70        offer_uri = self._create_offer(offer_id)
 71        return offer_uri
 72
 73    @override
 74    def get(self, offer_id: str) -> Offer:
 75        """Retrieve an offer from the SSI agent.
 76
 77        Args:
 78            offer_id: The offer identifier to retrieve.
 79
 80        Returns:
 81            The Offer object with the URI.
 82
 83        Raises:
 84            OfferNotFound: When the offer is not found in the upstream service.
 85        """
 86        response = requests.get(
 87            f"{self._ssi_agent_admin_base_url}/v0/offers/{offer_id}",
 88            timeout=self._timeout,
 89        )
 90
 91        if response.status_code == 404:
 92            raise OfferNotFound(f"Offer {offer_id} not found")
 93
 94        if 400 <= response.status_code < 600:
 95            raise OffersClientError(
 96                f"Upstream error: {response.status_code} - {response.content.decode()}"
 97            )
 98
 99        response_data: _SsiAgentOfferResponse = msgspec.json.decode(
100            response.content, type=_SsiAgentOfferResponse
101        )
102        uri: str = response_data.form_url_encoded_credential_offer
103
104        return Offer(
105            offer_id=offer_id,
106            award_id="",
107            uri=uri,
108        )
109
110    def _create_credential_for_subject(self, offer_id: str, award: Award) -> None:
111        response = requests.post(
112            f"{self._ssi_agent_admin_base_url}/v0/offers",
113            json= {
114                "offerId": offer_id,
115                "credential": asdict(award),
116                "credentialConfigurationId": "OpenBadgeCredential",
117                "expiresAt": "3025-10-24 11:34:00Z"
118            },
119            timeout=self._timeout,
120        )
121
122        if 400 <= response.status_code < 600:
123            raise OffersClientError(
124                f"Upstream error: {response.status_code} - {response.content.decode()}"
125            )
126
127
128
129    def _create_offer(self, offer_id: str) -> str:
130        response = requests.post(
131            f"{self._ssi_agent_admin_base_url}/v0/offers",
132            json= {
133                "offerId": offer_id,
134                "credentialConfigurationIds": ["OpenBadgeCredential"],
135            },
136            timeout=self._timeout,
137        )
138
139        if 400 <= response.status_code < 600:
140            raise OffersClientError(
141                f"Upstream error: {response.status_code} - {response.content.decode()}"
142            )
143
144        return response.text