Feature: Voucher integrations

src/aleph/sdk/client/authenticated_http.py

@@ -39,6 +39,7 @@
 from .abstract import AuthenticatedAlephClient
 from .http import AlephHttpClient
 from .services.authenticated_port_forwarder import AuthenticatedPortForwarder
+from .services.authenticated_voucher import AuthenticatedVoucher
 
 logger = logging.getLogger(__name__)
 
@@ -86,7 +87,7 @@ async def __aenter__(self):
         await super().__aenter__()
         # Override services with authenticated versions
         self.port_forwarder = AuthenticatedPortForwarder(self)
-
+        self.voucher = AuthenticatedVoucher(self)
         return self
 
     async def ipfs_push(self, content: Mapping) -> str:

src/aleph/sdk/client/http.py

@@ -61,6 +61,7 @@
     safe_getattr,
 )
 from .abstract import AlephClient
+from .services.voucher import Vouchers
 
 logger = logging.getLogger(__name__)
 
@@ -135,6 +136,7 @@ async def __aenter__(self):
         self.crn = Crn(self)
         self.scheduler = Scheduler(self)
         self.instance = Instance(self)
+        self.voucher = Vouchers(self)
 
         return self
 

src/aleph/sdk/client/services/authenticated_voucher.py

@@ -0,0 +1,62 @@
+from typing import TYPE_CHECKING, Optional, overload
+
+from typing_extensions import override
+
+from aleph.sdk.types import Voucher
+
+from .voucher import Vouchers
+
+if TYPE_CHECKING:
+    from aleph.sdk.client.abstract import AuthenticatedAlephClient
+
+
+class AuthenticatedVoucher(Vouchers):
+    """
+    This service is same logic than Vouchers but allow to don't pass address
+    to use account address
+    """
+
+    def __init__(self, client: "AuthenticatedAlephClient"):
+        super().__init__(client)
+
+    @overload
+    def _resolve_address(self, address: str) -> str: ...
+
+    @overload
+    def _resolve_address(self, address: None) -> str: ...
+
+    @override
+    def _resolve_address(self, address: Optional[str] = None) -> str:
+        """
+        Resolve the address to use. Prefer the provided address, fallback to account.
+        """
+        if address:
+            return address
+        if self._client.account:
+            return self._client.account.get_address()
+
+        raise ValueError("No address provided and no account configured")
+
+    @override
+    async def get_vouchers(self, address: Optional[str] = None) -> list[Voucher]:
+        """
+        Retrieve all vouchers for the account / specific address, across EVM and Solana chains.
+        """
+        address = address or self._client.account.get_address()
+        return await super().get_vouchers(address=address)
+
+    @override
+    async def get_evm_vouchers(self, address: Optional[str] = None) -> list[Voucher]:
+        """
+        Retrieve vouchers specific to EVM chains for a specific address.
+        """
+        address = address or self._client.account.get_address()
+        return await super().get_evm_vouchers(address=address)
+
+    @override
+    async def get_solana_vouchers(self, address: Optional[str] = None) -> list[Voucher]:
+        """
+        Fetch Solana vouchers for a specific address.
+        """
+        address = address or self._client.account.get_address()
+        return await super().get_solana_vouchers(address=address)

src/aleph/sdk/client/services/voucher.py

@@ -0,0 +1,164 @@
+from typing import Optional
+
+import aiohttp
+from aiohttp import ClientResponseError
+from aleph_message.models import Chain
+
+from aleph.sdk.conf import settings
+from aleph.sdk.query.filters import PostFilter
+from aleph.sdk.query.responses import Post, PostsResponse
+from aleph.sdk.types import Voucher, VoucherMetadata
+
+
+class Vouchers:
+    """
+    This service is made to fetch voucher (SOL / EVM)
+    """
+
+    def __init__(self, client):
+        self._client = client
+
+    # Utils
+    def _resolve_address(self, address: str) -> str:
+        return address  # Not Authenticated client so address need to be given
+
+    async def _fetch_voucher_update(self):
+        """
+        Fetch the latest EVM voucher update (unfiltered).
+        """
+
+        post_filter = PostFilter(
+            types=["vouchers-update"], addresses=[settings.VOUCHER_SENDER]
+        )
+        vouchers_post: PostsResponse = await self._client.get_posts(
+            post_filter=post_filter, page_size=1
+        )
+
+        if not vouchers_post.posts:
+            return []
+
+        message_post: Post = vouchers_post.posts[0]
+
+        nft_vouchers = message_post.content.get("nft_vouchers", {})
+        return list(nft_vouchers.items())  # [(voucher_id, voucher_data)]
+
+    async def _fetch_solana_voucher_list(self):
+        """
+        Fetch full Solana voucher registry (unfiltered).
+        """
+        try:
+            async with aiohttp.ClientSession() as session:
+                async with session.get(settings.VOUCHER_SOL_REGISTRY) as resp:
+                    resp.raise_for_status()
+                    return await resp.json()
+        except ClientResponseError:
+            return {}
+
+    async def fetch_voucher_metadata(
+        self, metadata_id: str
+    ) -> Optional[VoucherMetadata]:
+        """
+        Fetch metadata for a given voucher.
+        """
+        url = f"https://claim.twentysix.cloud/sbt/metadata/{metadata_id}.json"
+        try:
+            async with aiohttp.ClientSession() as session:
+                async with session.get(url) as resp:
+                    resp.raise_for_status()
+                    data = await resp.json()
+                    return VoucherMetadata.model_validate(data)
+        except ClientResponseError:
+            return None
+
+    async def get_solana_vouchers(self, address: str) -> list[Voucher]:
+        """
+        Fetch Solana vouchers for a specific address.
+        """
+        resolved_address = self._resolve_address(address=address)
+        vouchers: list[Voucher] = []
+
+        registry_data = await self._fetch_solana_voucher_list()
+
+        claimed_tickets = registry_data.get("claimed_tickets", {})
+        batches = registry_data.get("batches", {})
+
+        for ticket_hash, ticket_data in claimed_tickets.items():
+            claimer = ticket_data.get("claimer")
+            if claimer != resolved_address:
+                continue
+
+            batch_id = ticket_data.get("batch_id")
+            metadata_id = None
+
+            if str(batch_id) in batches:
+                metadata_id = batches[str(batch_id)].get("metadata_id")
+
+            if metadata_id:
+                metadata = await self.fetch_voucher_metadata(metadata_id)
+                if metadata:
+                    voucher = Voucher(
+                        id=ticket_hash,
+                        metadata_id=metadata_id,
+                        name=metadata.name,
+                        description=metadata.description,
+                        external_url=metadata.external_url,
+                        image=metadata.image,
+                        icon=metadata.icon,
+                        attributes=metadata.attributes,
+                    )
+                    vouchers.append(voucher)
+
+        return vouchers
+
+    async def get_evm_vouchers(self, address: str) -> list[Voucher]:
+        """
+        Retrieve vouchers specific to EVM chains for a specific address.
+        """
+        resolved_address = self._resolve_address(address=address)
+        vouchers: list[Voucher] = []
+
+        nft_vouchers = await self._fetch_voucher_update()
+        for voucher_id, voucher_data in nft_vouchers:
+            if voucher_data.get("claimer") != resolved_address:
+                continue
+
+            metadata_id = voucher_data.get("metadata_id")
+            metadata = await self.fetch_voucher_metadata(metadata_id)
+            if not metadata:
+                continue
+
+            voucher = Voucher(
+                id=voucher_id,
+                metadata_id=metadata_id,
+                name=metadata.name,
+                description=metadata.description,
+                external_url=metadata.external_url,
+                image=metadata.image,
+                icon=metadata.icon,
+                attributes=metadata.attributes,
+            )
+            vouchers.append(voucher)
+        return vouchers
+
+    async def fetch_vouchers_by_chain(self, chain: Chain, address: str):
+        if chain == Chain.SOL:
+            return await self.get_solana_vouchers(address=address)
+        else:
+            return await self.get_evm_vouchers(address=address)
+
+    async def get_vouchers(self, address: str) -> list[Voucher]:
+        """
+        Retrieve all vouchers for the account / specific adress, across EVM and Solana chains.
+        """
+        vouchers = []
+
+        # Get EVM vouchers
+        if address.startswith("0x") and len(address) == 42:
+            evm_vouchers = await self.get_evm_vouchers(address=address)
+            vouchers.extend(evm_vouchers)
+        else:
+            # Get Solana vouchers
+            solana_vouchers = await self.get_solana_vouchers(address=address)
+            vouchers.extend(solana_vouchers)
+
+        return vouchers

src/aleph/sdk/conf.py

@@ -92,6 +92,12 @@ class Settings(BaseSettings):
     )
     SCHEDULER_URL: ClassVar[str] = "https://scheduler.api.aleph.cloud/"
 
+    VOUCHER_METDATA_TEMPLATE_URL: str = (
+        "https://claim.twentysix.cloud/sbt/metadata/{}.json"
+    )
+    VOUCHER_SOL_REGISTRY: str = "https://api.claim.twentysix.cloud/v1/registry/sol"
+    VOUCHER_SENDER: str = "0xB34f25f2c935bCA437C061547eA12851d719dEFb"
+
     # Web3Provider settings
     TOKEN_DECIMALS: ClassVar[int] = 18
     TX_TIMEOUT: ClassVar[int] = 60 * 3

src/aleph/sdk/types.py

@@ -1,5 +1,6 @@
 from abc import abstractmethod
 from datetime import datetime
+from decimal import Decimal
 from enum import Enum
 from typing import Any, Dict, List, Literal, Optional, Protocol, TypeVar, Union
 
@@ -289,3 +290,29 @@ class Ports(BaseModel):
 
 
 AllForwarders = RootModel[Dict[ItemHash, Ports]]
+
+
+class VoucherAttribute(BaseModel):
+    value: Union[str, Decimal]
+    trait_type: str = Field(..., alias="trait_type")
+    display_type: Optional[str] = Field(None, alias="display_type")
+
+
+class VoucherMetadata(BaseModel):
+    name: str
+    description: str
+    external_url: str
+    image: str
+    icon: str
+    attributes: list[VoucherAttribute]
+
+
+class Voucher(BaseModel):
+    id: str
+    metadata_id: str
+    name: str
+    description: str
+    external_url: str
+    image: str
+    icon: str
+    attributes: list[VoucherAttribute]

tests/unit/conftest.py

@@ -306,3 +306,65 @@ def post(self, *_args, **_kwargs):
     client._http_session = http_session
 
     return client
+
+
+@pytest.fixture
+def make_mock_aiohttp_session():
+    def _make(mocked_json_response):
+        mock_response = AsyncMock()
+        mock_response.json.return_value = mocked_json_response
+        mock_response.raise_for_status.return_value = None
+
+        session = MagicMock()
+
+        get_cm = AsyncMock()
+        get_cm.__aenter__.return_value = mock_response
+        session.get.return_value = get_cm
+
+        session_cm = AsyncMock()
+        session_cm.__aenter__.return_value = session
+        return session_cm
+
+    return _make
+
+
+# Constants needed for voucher tests
+MOCK_ADDRESS = "0x1234567890123456789012345678901234567890"
+MOCK_SOLANA_ADDRESS = "abcdefghijklmnopqrstuvwxyz123456789"
+MOCK_METADATA_ID = "metadata123"
+MOCK_VOUCHER_ID = "voucher123"
+MOCK_METADATA = {
+    "name": "Test Voucher",
+    "description": "A test voucher",
+    "external_url": "https://example.com",
+    "image": "https://example.com/image.png",
+    "icon": "https://example.com/icon.png",
+    "attributes": [
+        {"trait_type": "Test Trait", "value": "Test Value"},
+        {"trait_type": "Numeric Trait", "value": "123", "display_type": "number"},
+    ],
+}
+
+MOCK_EVM_VOUCHER_DATA = [
+    (MOCK_VOUCHER_ID, {"claimer": MOCK_ADDRESS, "metadata_id": MOCK_METADATA_ID})
+]
+
+MOCK_SOLANA_REGISTRY = {
+    "claimed_tickets": {
+        "solticket123": {"claimer": MOCK_SOLANA_ADDRESS, "batch_id": "batch123"}
+    },
+    "batches": {"batch123": {"metadata_id": MOCK_METADATA_ID}},
+}
+
+
+@pytest.fixture
+def mock_post_response():
+    mock_post = MagicMock()
+    mock_post.content = {
+        "nft_vouchers": {
+            MOCK_VOUCHER_ID: {"claimer": MOCK_ADDRESS, "metadata_id": MOCK_METADATA_ID}
+        }
+    }
+    posts_response = MagicMock()
+    posts_response.posts = [mock_post]
+    return posts_response