src.offers.ssi_agent_offers_client_adapter

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