src.credential_configurations.ssi_agent_credential_configurations_client_adapter

  1"""SSI-Agent Adapter for credential configurations operations."""
  2
  3from dataclasses import asdict, dataclass
  4from typing import override
  5
  6import msgspec
  7
  8from src.lib.http_client import HttpClient, RequestsHttpClient
  9
 10from .credential_configurations_client_port import (
 11    CredentialConfigurationNotFound,
 12    CredentialConfigurationsClientError,
 13    CredentialConfigurationsClientPort,
 14)
 15from .models import (
 16    CredentialConfiguration,
 17    Display,
 18)
 19
 20
 21# Response model for issuer metadata
 22@dataclass
 23class _SsiAgentIssuerMetadata:
 24    """SSI agent issuer metadata response."""
 25
 26    credential_issuer: str
 27    credential_endpoint: str
 28    credential_configurations_supported: dict[str, CredentialConfiguration]
 29
 30
 31@dataclass
 32class _SsiAgentDisplay:
 33    """Display information for a credential configuration."""
 34
 35    name: str | None = None
 36    locale: str | None = None
 37    logo: dict[str, str | None] | None = None
 38
 39    @staticmethod
 40    def from_configuration(configuration: Display) -> "_SsiAgentDisplay":
 41        return _SsiAgentDisplay(
 42            name=configuration.name,
 43            logo=configuration.logo,
 44            locale=configuration.locale,
 45        )
 46
 47
 48@dataclass
 49class _SsiAgentAddPayload:
 50    """Payload for Create (and therefore update) credential configuration on service"""
 51
 52    credential_configuration_id: str | None = None
 53    format: str | None = None
 54    type: list[str] | None = None
 55    display: list[_SsiAgentDisplay] | None = None
 56
 57    @staticmethod
 58    def from_credential_configuration(
 59        configuration: CredentialConfiguration,
 60    ) -> "_SsiAgentAddPayload":
 61        assert configuration.credential_definition is not None, (
 62            "`.credential_definition.type` is required, and must be a list of strings"
 63        )
 64        type_list = configuration.credential_definition.type
 65
 66        if configuration.credential_metadata.display is None:
 67            displays = []
 68        else:
 69            displays = [
 70                _SsiAgentDisplay.from_configuration(display)
 71                for display in configuration.credential_metadata.display
 72            ]
 73
 74        return _SsiAgentAddPayload(
 75            credential_configuration_id=configuration.credential_configuration_id,
 76            format=configuration.format,
 77            type=type_list,
 78            display=displays,
 79        )
 80
 81
 82class SsiAgentCredentialConfigurationsClientAdapter(CredentialConfigurationsClientPort):
 83    """Adapter for SSI Agent credential configurations API."""
 84
 85    _ssi_agent_admin_base_url: str
 86    _ssi_agent_issuer_base_url: str
 87    _http_client: HttpClient
 88
 89    def __init__(
 90        self,
 91        ssi_agent_url: str,
 92        http_client: HttpClient | None = None,
 93    ) -> None:
 94        """Initialize the adapter.
 95
 96        Args:
 97            ssi_agent_url: The admin base URL of the SSI agent.
 98            http_client: The HTTP client to use for requests.
 99                Defaults to requests module.
100        """
101        self._ssi_agent_admin_base_url = ssi_agent_url.rstrip("/")
102        self._ssi_agent_issuer_base_url = ssi_agent_url.rstrip("/")
103        if http_client is not None:
104            self._http_client = http_client
105        else:
106            self._http_client = RequestsHttpClient()
107
108    @override
109    def create(self, configuration: CredentialConfiguration) -> CredentialConfiguration:
110        """Create a new credential configuration.
111
112        Args:
113            configuration: The credential configuration to create.
114
115        Returns:
116            The created credential configuration.
117
118        Raises:
119            CredentialConfigurationsClientError: When upstream service returns an error.
120            ValueError: When the credential configuration ID is not set.
121        """
122        payload = _SsiAgentAddPayload.from_credential_configuration(configuration)
123
124        # For SSI Agent, POST to /v0/credential-configurations with existing
125        # ID updates it
126        response = self._http_client.post(
127            f"{self._ssi_agent_admin_base_url}/v0/credential-configurations",
128            json=asdict(payload),
129        )
130
131        if response.status_code == 404:
132            raise CredentialConfigurationNotFound(
133                f"Configuration {configuration.credential_configuration_id} not found"
134            )
135
136        if 400 <= response.status_code < 600:
137            raise CredentialConfigurationsClientError(
138                f"Upstream error: {response.status_code} - {response.text}"
139            )
140
141        # Rather than returning the input, we re-fetch it, so we get the result as it is
142        # merged and enriched with the issuer config.
143        configuration = self.get(configuration.credential_configuration_id)
144        return configuration
145
146    @override
147    def get(self, configuration_id: str) -> CredentialConfiguration:
148        """Retrieve a credential configuration by ID.
149
150        Args:
151            configuration_id: The unique credential configuration identifier.
152
153        Returns:
154            The matching CredentialConfiguration.
155
156        Raises:
157            CredentialConfigurationNotFound: When not found.
158            CredentialConfigurationsClientError: When upstream service returns an error.
159        """
160        # Get issuer metadata which contains all credential configurations
161        all = self.list()
162
163        # Find by credential_configuration_id
164        for config in all:
165            if config.credential_configuration_id == configuration_id:
166                return config
167
168        raise CredentialConfigurationNotFound(configuration_id)
169
170    @override
171    def list(self) -> list[CredentialConfiguration]:
172        """List all credential configurations.
173
174        Returns:
175            A list of all credential configurations.
176
177        Raises:
178            CredentialConfigurationsClientError: When upstream service returns an error.
179        """
180        response = self._http_client.get(
181            f"{self._ssi_agent_admin_base_url}/.well-known/openid-credential-issuer",
182        )
183
184        if 400 <= response.status_code < 600:
185            raise CredentialConfigurationsClientError(
186                f"Upstream error: {response.status_code} - {response.text}"
187            )
188
189        try:
190            metadata: _SsiAgentIssuerMetadata = msgspec.json.decode(
191                response.content, type=_SsiAgentIssuerMetadata
192            )
193        except msgspec.DecodeError as e:
194            raise CredentialConfigurationsClientError(
195                f"Invalid response from upstream: {e}"
196            ) from e
197
198        configurations: list[CredentialConfiguration] = []
199        for (
200            id,
201            configuration,
202        ) in metadata.credential_configurations_supported.items():
203            configuration.credential_configuration_id = id
204            configurations.append(configuration)
205
206        return configurations
207
208    @override
209    def update(self, configuration: CredentialConfiguration) -> CredentialConfiguration:
210        """Update an existing credential configuration.
211
212        Implemented by "create" with an existing id.
213        See self.create()
214        """
215        return self.create(configuration)
216
217    @override
218    def delete(self, configuration_id: str) -> None:
219        """Delete a credential configuration.
220
221        Args:
222            configuration_id: The unique credential configuration identifier.
223
224        Raises:
225            CredentialConfigurationNotFound: When not found.
226            CredentialConfigurationsClientError: When upstream service returns an error.
227        """
228        raise NotImplementedError(
229            "Delete is not implemented for credential configurations"
230        )