src.metadata

Metadata module.

Public exports:

  • Domain models: CredentialIssuerMetadata and related types
  • Port: MetadataRepositoryPort
  • Service: MetadataService
  • Errors: MetadataNotFoundError, MetadataSerializationError

Private (internal):

  • Adapters: postgresql_adapter, ssi_agent_metadata_adapter
 1"""Metadata module.
 2
 3Public exports:
 4- Domain models: CredentialIssuerMetadata and related types
 5- Port: MetadataRepositoryPort
 6- Service: MetadataService
 7- Errors: MetadataNotFoundError, MetadataSerializationError
 8
 9Private (internal):
10- Adapters: postgresql_adapter, ssi_agent_metadata_adapter
11"""
12
13from src.metadata.credential_issuer_metadata import CredentialIssuerMetadata
14from src.metadata.metadata_repository import (
15    MetadataNotFoundError,
16    MetadataSerializationError,
17)
18from src.metadata.metadata_repository import MetadataRepositoryPort
19from src.metadata.metadata_service import MetadataService
20
21__all__ = [
22    "CredentialIssuerMetadata",
23    "MetadataNotFoundError",
24    "MetadataSerializationError",
25    "MetadataRepositoryPort",
26    "MetadataService",
27]
@dataclass
class CredentialIssuerMetadata: 
58@dataclass
59class CredentialIssuerMetadata:
60    credential_issuer: str
61    credential_endpoint: str
62    credential_configurations_supported: dict[str, CredentialConfiguration]
63    authorization_servers: list[str] | None = None
64    nonce_endpoint: str | None = None
65    deferred_credential_endpoint: str | None = None
66    notification_endpoint: str | None = None
67    credential_response_encryption: dict[str, CredentialResponseEncryption] | None = (
68        None
69    )
70    batch_credential_issuance: dict[str, BatchCredentialIssuance] | None = None
71    display: list[Display] | None = None
CredentialIssuerMetadata( credential_issuer: str, credential_endpoint: str, credential_configurations_supported: dict[str, src.metadata.credential_issuer_metadata.CredentialConfiguration], authorization_servers: list[str] | None = None, nonce_endpoint: str | None = None, deferred_credential_endpoint: str | None = None, notification_endpoint: str | None = None, credential_response_encryption: dict[str, src.metadata.credential_issuer_metadata.CredentialResponseEncryption] | None = None, batch_credential_issuance: dict[str, src.metadata.credential_issuer_metadata.BatchCredentialIssuance] | None = None, display: list[src.metadata.credential_issuer_metadata.Display] | None = None)
credential_issuer: str
credential_endpoint: str
credential_configurations_supported: dict[str, src.metadata.credential_issuer_metadata.CredentialConfiguration]
authorization_servers: list[str] | None = None
nonce_endpoint: str | None = None
deferred_credential_endpoint: str | None = None
notification_endpoint: str | None = None
credential_response_encryption: dict[str, src.metadata.credential_issuer_metadata.CredentialResponseEncryption] | None = None
batch_credential_issuance: dict[str, src.metadata.credential_issuer_metadata.BatchCredentialIssuance] | None = None
display: list[src.metadata.credential_issuer_metadata.Display] | None = None
class MetadataNotFoundError(builtins.KeyError): 
 9class MetadataNotFoundError(KeyError):
10    """Exception raised when metadata is not found."""
11
12    pass

Exception raised when metadata is not found.

class MetadataSerializationError(builtins.ValueError): 
15class MetadataSerializationError(ValueError):
16    """Exception raised when metadata serialization/deserialization fails."""
17
18    pass

Exception raised when metadata serialization/deserialization fails.

class MetadataRepositoryPort(abc.ABC): 
21class MetadataRepositoryPort(ABC):
22    """Port: repository interface for persisting and retrieving metadata."""
23
24    @abstractmethod
25    def store(self, metadata: CredentialIssuerMetadata) -> None:
26        """Persist metadata.
27
28        Args:
29            metadata: The CredentialIssuerMetadata to store.
30        """
31        ...
32
33    @abstractmethod
34    def get(self) -> CredentialIssuerMetadata:
35        """Retrieve the latest metadata entry.
36
37        Returns:
38            The CredentialIssuerMetadata.
39
40        Raises:
41            MetadataNotFoundError: When no metadata entry exists.
42        """
43        ...

Port: repository interface for persisting and retrieving metadata.

@abstractmethod
def store( self, metadata: CredentialIssuerMetadata) -> None: 
24    @abstractmethod
25    def store(self, metadata: CredentialIssuerMetadata) -> None:
26        """Persist metadata.
27
28        Args:
29            metadata: The CredentialIssuerMetadata to store.
30        """
31        ...

Persist metadata.

Args: metadata: The CredentialIssuerMetadata to store.

@abstractmethod
def get(self) -> CredentialIssuerMetadata: 
33    @abstractmethod
34    def get(self) -> CredentialIssuerMetadata:
35        """Retrieve the latest metadata entry.
36
37        Returns:
38            The CredentialIssuerMetadata.
39
40        Raises:
41            MetadataNotFoundError: When no metadata entry exists.
42        """
43        ...

Retrieve the latest metadata entry.

Returns: The CredentialIssuerMetadata.

Raises: MetadataNotFoundError: When no metadata entry exists.

class MetadataService: 
13class MetadataService:
14    _metadata_repository: MetadataRepositoryPort
15    public_url: str
16
17    def __init__(
18        self, metadata_repository: MetadataRepositoryPort, public_url: str
19    ) -> None:
20        """Initialize the MetadataService.
21
22        Args:
23            metadata_repository: The metadata repository port.
24            public_url: The publicly accessible base URL of this issuer service.
25        """
26        self._metadata_repository = metadata_repository
27        self.public_url = public_url
28
29    def get_health(self) -> HealthStatus:
30        return HealthStatus.HEALTHY
31
32    def get_credential_issuer_metadata(self) -> CredentialIssuerMetadata:
33        """Get credential issuer metadata with URLs replaced by our own.
34
35        Returns:
36            CredentialIssuerMetadata with credential_issuer and credential_endpoint
37            replaced with our public_url.
38        """
39        return self._metadata_repository.get()
MetadataService( metadata_repository: MetadataRepositoryPort, public_url: str) 
17    def __init__(
18        self, metadata_repository: MetadataRepositoryPort, public_url: str
19    ) -> None:
20        """Initialize the MetadataService.
21
22        Args:
23            metadata_repository: The metadata repository port.
24            public_url: The publicly accessible base URL of this issuer service.
25        """
26        self._metadata_repository = metadata_repository
27        self.public_url = public_url

Initialize the MetadataService.

Args: metadata_repository: The metadata repository port. public_url: The publicly accessible base URL of this issuer service.

public_url: str
def get_health(self) -> src.metadata.metadata_service.HealthStatus: 
29    def get_health(self) -> HealthStatus:
30        return HealthStatus.HEALTHY
def get_credential_issuer_metadata(self) -> CredentialIssuerMetadata: 
32    def get_credential_issuer_metadata(self) -> CredentialIssuerMetadata:
33        """Get credential issuer metadata with URLs replaced by our own.
34
35        Returns:
36            CredentialIssuerMetadata with credential_issuer and credential_endpoint
37            replaced with our public_url.
38        """
39        return self._metadata_repository.get()

Get credential issuer metadata with URLs replaced by our own.

Returns: CredentialIssuerMetadata with credential_issuer and credential_endpoint replaced with our public_url.