Feature: Voucher Integrations

src/aleph/sdk/client/authenticated_http.py

@@ -5,7 +5,7 @@
 import time
 from io import BytesIO
 from pathlib import Path
-from typing import Any, Dict, Mapping, NoReturn, Optional, Tuple, Union
+from typing import Any, Dict, Mapping, NoReturn, Optional, Tuple, Union, overload
 
 import aiohttp
 from aleph_message.models import (
@@ -39,6 +39,12 @@
 from .abstract import AuthenticatedAlephClient
 from .http import AlephHttpClient
 
+try:
+    from typing import override  # type: ignore
+except ImportError:
+    from typing_extensions import override  # type: ignore
+
+
 logger = logging.getLogger(__name__)
 
 try:
@@ -679,3 +685,45 @@ async def _upload_file_native(
         # nodes.
         _, status = await self._broadcast(message=message, sync=sync)
         return message, status
+
+    @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.account:
+            return self.account.get_address()
+
+        raise ValueError("No address provided and no account configured")
+
+    @override
+    async def get_vouchers(self, address: Optional[str] = None) -> list:
+        """
+        Retrieve all vouchers for the account / specific address, across EVM and Solana chains.
+        """
+        address = address or self.account.get_address()
+        return await super().get_vouchers(address=address)
+
+    @override
+    async def get_evm_vouchers(self, address: Optional[str] = None) -> list:
+        """
+        Retrieve vouchers specific to EVM chains for a specific address.
+        """
+        address = address or self.account.get_address()
+        return await super().get_evm_vouchers(address=address)
+
+    @override
+    async def get_solana_vouchers(self, address: Optional[str] = None) -> list:
+        """
+        Fetch Solana vouchers for a specific address.
+        """
+        address = address or self.account.get_address()
+        return await super().get_solana_vouchers(address=address)

src/aleph/sdk/client/http.py

@@ -42,7 +42,7 @@
 )
 from ..query.filters import MessageFilter, PostFilter
 from ..query.responses import MessagesResponse, Post, PostsResponse, PriceResponse
-from ..types import GenericMessage, StoredContent
+from ..types import GenericMessage, StoredContent, Voucher, VoucherMetadata
 from ..utils import (
     Writable,
     check_unix_socket_valid,
@@ -563,3 +563,162 @@ async def get_stored_content(
             if result
             else StoredContent(error=resp, filename=None, hash=None, url=None)
         )
+
+    def _resolve_address(self, address: str) -> str:
+        return address
+
+    async def _fetch_voucher_update(self):
+        """
+        Fetch the latest EVM voucher update (unfiltered).
+        """
+        async with AlephHttpClient(api_server=settings.API_HOST) as client:
+            post_filter = PostFilter(
+                types=["vouchers-update"], addresses=[settings.VOUCHER_SENDER]
+            )
+            vouchers_post: PostsResponse = await 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:
+                try:
+                    async with session.get(settings.VOUCHER_SOL_REGISTRY) as resp:
+                        if resp.status != 200:
+                            return {}
+
+                        try:
+                            return await resp.json()
+                        except Exception:  # Catch any exception during JSON parsing
+                            text_data = await resp.text()
+                            try:
+                                return json.loads(text_data)
+                            except json.JSONDecodeError:
+                                return {}
+                except Exception:
+                    return {}
+        except Exception:
+            return {}
+
+    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
+        evm_vouchers = await self.get_evm_vouchers(address=address)
+        vouchers.extend(evm_vouchers)
+
+        # Get Solana vouchers
+        solana_vouchers = await self.get_solana_vouchers(address=address)
+        vouchers.extend(solana_vouchers)
+
+        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 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 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:
+                try:
+                    async with session.get(url) as resp:
+                        if resp.status != 200:
+                            return None
+                        data = await resp.json()
+                        return VoucherMetadata.model_validate(data)
+                except Exception as e:
+                    logger.error(f"Error fetching metadata: {e}")
+                    return None
+        except Exception as e:
+            logger.error(f"Error creating session: {e}")
+            return None

src/aleph/sdk/conf.py

@@ -230,6 +230,12 @@ class Settings(BaseSettings):
     model_config = SettingsConfigDict(
         env_prefix="ALEPH_", case_sensitive=False, env_file=".env", extra="ignore"
     )
+    # Voucher Config
+    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"
 
 
 class MainConfiguration(BaseModel):

src/aleph/sdk/types.py

@@ -1,6 +1,7 @@
 from abc import abstractmethod
+from decimal import Decimal
 from enum import Enum
-from typing import Dict, Optional, Protocol, TypeVar
+from typing import Dict, Optional, Protocol, TypeVar, Union
 
 from pydantic import BaseModel, Field
 
@@ -100,3 +101,29 @@ class TokenType(str, Enum):
 
     GAS = "GAS"
     ALEPH = "ALEPH"
+
+
+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 = Field(..., alias="external_url")
+    image: str
+    icon: str
+    attributes: list[VoucherAttribute]
+
+
+class Voucher(BaseModel):
+    id: str
+    metadata_id: str = Field(..., alias="metadata_id")
+    name: str
+    description: str
+    external_url: str = Field(..., alias="external_url")
+    image: str
+    icon: str
+    attributes: list[VoucherAttribute]