From b5459f9190a0c2eefc5e62ef3bc03d5972ab2017 Mon Sep 17 00:00:00 2001 From: Helder Vasconcelos Date: Sun, 10 May 2026 17:30:45 +0100 Subject: [PATCH] Add tokens and fiat-currencies resources - Extend Token model: chain_id and address are now Optional (chainless canonical tokens have both None); new fields id, name, total_supply, logo_url, tags, canonical_id. - New FiatCurrency, Pagination, TokenListResponse and FiatCurrencyListResponse models. - New TokensResource (list / get) and FiatCurrenciesResource (list / get) exposed on the Luzia client as `luzia.tokens` and `luzia.fiat_currencies`. - Wires the new public /v1/tokens and /v1/fiat-currencies API endpoints. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/luziadev/__init__.py | 10 +++ src/luziadev/client.py | 4 + src/luziadev/models.py | 95 +++++++++++++++++++++-- src/luziadev/resources/fiat_currencies.py | 54 +++++++++++++ src/luziadev/resources/tokens.py | 57 ++++++++++++++ 5 files changed, 215 insertions(+), 5 deletions(-) create mode 100644 src/luziadev/resources/fiat_currencies.py create mode 100644 src/luziadev/resources/tokens.py diff --git a/src/luziadev/__init__.py b/src/luziadev/__init__.py index ac81769..7393218 100644 --- a/src/luziadev/__init__.py +++ b/src/luziadev/__init__.py @@ -7,16 +7,21 @@ ) from luziadev.models import ( Exchange, + FiatCurrency, + FiatCurrencyListResponse, Market, MarketListResponse, OHLCVCandle, OHLCVResponse, + Pagination, RateLimitInfo, Ticker, TickerListResponse, Token, + TokenListResponse, ) from luziadev.resources.exchanges import ExchangeType +from luziadev.resources.fiat_currencies import EnabledFilter from luziadev.resources.markets import MarketType from luziadev.retry import RetryContext, RetryOptions from luziadev.websocket import LuziaWebSocket @@ -40,6 +45,11 @@ "RetryOptions", "RetryContext", "Token", + "TokenListResponse", + "FiatCurrency", + "FiatCurrencyListResponse", + "EnabledFilter", + "Pagination", "ErrorCode", "is_luzia_error", "is_retryable_error", diff --git a/src/luziadev/client.py b/src/luziadev/client.py index 52c006c..eff9cdf 100644 --- a/src/luziadev/client.py +++ b/src/luziadev/client.py @@ -12,9 +12,11 @@ ) from luziadev.models import RateLimitInfo from luziadev.resources.exchanges import ExchangesResource +from luziadev.resources.fiat_currencies import FiatCurrenciesResource from luziadev.resources.history import HistoryResource from luziadev.resources.markets import MarketsResource from luziadev.resources.tickers import TickersResource +from luziadev.resources.tokens import TokensResource from luziadev.retry import RetryOptions, with_retry @@ -54,6 +56,8 @@ def __init__( self.tickers = TickersResource(self) self.markets = MarketsResource(self) self.history = HistoryResource(self) + self.tokens = TokensResource(self) + self.fiat_currencies = FiatCurrenciesResource(self) @property def rate_limit_info(self) -> Optional[RateLimitInfo]: diff --git a/src/luziadev/models.py b/src/luziadev/models.py index db27dfe..d62d11e 100644 --- a/src/luziadev/models.py +++ b/src/luziadev/models.py @@ -72,20 +72,105 @@ def from_dict(cls, data: dict) -> Ticker: @dataclass(frozen=True) class Token: - """On-chain token referenced by a DEX market.""" + """Canonical asset or on-chain token. - address: str = "" + Chain-bound rows have ``chain_id`` and ``address`` set; chainless canonical + rows (id ``crypto:SYMBOL``) have both ``None``. + """ + + # Composite id ("crypto:USDC", "ethereum:WETH"). Always present from the + # /v1/tokens endpoint; may be empty when constructed from legacy nested + # baseToken/quoteToken payloads on DEX markets. + id: str = "" symbol: str = "" + name: str = "" decimals: int = 0 - chain_id: str = "" + address: Optional[str] = None + chain_id: Optional[str] = None + total_supply: Optional[str] = None + logo_url: Optional[str] = None + tags: tuple[str, ...] = () + # For on-chain rows, the chainless ``crypto:SYMBOL`` row this token mirrors. + canonical_id: Optional[str] = None @classmethod def from_dict(cls, data: dict) -> Token: + tags_raw = data.get("tags") or [] return cls( - address=data.get("address", ""), + id=data.get("id", ""), symbol=data.get("symbol", ""), + name=data.get("name", ""), decimals=int(data.get("decimals", 0)), - chain_id=data.get("chainId", ""), + address=data.get("address"), + chain_id=data.get("chainId"), + total_supply=data.get("totalSupply"), + logo_url=data.get("logoUrl"), + tags=tuple(tags_raw), + canonical_id=data.get("canonicalId"), + ) + + +@dataclass(frozen=True) +class FiatCurrency: + """ISO 4217 fiat currency referenced by markets.""" + + code: str = "" + name: str = "" + symbol: Optional[str] = None + enabled: bool = True + + @classmethod + def from_dict(cls, data: dict) -> FiatCurrency: + return cls( + code=data.get("code", ""), + name=data.get("name", ""), + symbol=data.get("symbol"), + enabled=bool(data.get("enabled", True)), + ) + + +@dataclass(frozen=True) +class Pagination: + """Pagination metadata returned alongside paginated list responses.""" + + total: int = 0 + page: int = 1 + limit: int = 20 + pages: int = 0 + + @classmethod + def from_dict(cls, data: dict) -> Pagination: + return cls( + total=int(data.get("total", 0)), + page=int(data.get("page", 1)), + limit=int(data.get("limit", 20)), + pages=int(data.get("pages", 0)), + ) + + +@dataclass(frozen=True) +class TokenListResponse: + data: tuple[Token, ...] = () + pagination: Pagination = Pagination() + + @classmethod + def from_dict(cls, data: dict) -> TokenListResponse: + return cls( + data=tuple(Token.from_dict(t) for t in data.get("data", [])), + pagination=Pagination.from_dict(data.get("pagination", {})), + ) + + +@dataclass(frozen=True) +class FiatCurrencyListResponse: + data: tuple[FiatCurrency, ...] = () + pagination: Pagination = Pagination() + + @classmethod + def from_dict(cls, data: dict) -> FiatCurrencyListResponse: + return cls( + data=tuple(FiatCurrency.from_dict(f) for f in data.get("data", [])), + pagination=Pagination.from_dict(data.get("pagination", {})), ) diff --git a/src/luziadev/resources/fiat_currencies.py b/src/luziadev/resources/fiat_currencies.py new file mode 100644 index 0000000..ecd7cbb --- /dev/null +++ b/src/luziadev/resources/fiat_currencies.py @@ -0,0 +1,54 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING, Literal, Optional, Union + +from luziadev.models import FiatCurrency, FiatCurrencyListResponse + +if TYPE_CHECKING: + from luziadev.client import Luzia + + +# Filter for the ``enabled`` query parameter. +EnabledFilter = Union[bool, Literal["all"]] + + +class FiatCurrenciesResource: + """List and look up ISO 4217 fiat currencies referenced by markets.""" + + def __init__(self, client: Luzia) -> None: + self._client = client + + async def list( + self, + *, + search: Optional[str] = None, + enabled: EnabledFilter = True, + page: int = 1, + limit: int = 50, + ) -> FiatCurrencyListResponse: + """List fiat currencies. + + Args: + search: Case-insensitive search across code and name. + enabled: ``True`` returns enabled only (default), ``False`` disabled + only, ``"all"`` for both. + page: Page number, 1-based. + limit: Items per page, max 200. + + Example: + >>> page = await luzia.fiat_currencies.list(search="EUR") + """ + query: dict = {"page": page, "limit": limit} + if search is not None: + query["search"] = search + if enabled == "all": + query["enabled"] = "all" + else: + query["enabled"] = "true" if enabled else "false" + data = await self._client.request("/v1/fiat-currencies", query=query) + return FiatCurrencyListResponse.from_dict(data) + + async def get(self, code: str) -> FiatCurrency: + """Look up a single fiat currency by ISO 4217 code (e.g. ``"USD"``).""" + data = await self._client.request(f"/v1/fiat-currencies/{code.upper()}") + return FiatCurrency.from_dict(data.get("data", {})) diff --git a/src/luziadev/resources/tokens.py b/src/luziadev/resources/tokens.py new file mode 100644 index 0000000..87df4ac --- /dev/null +++ b/src/luziadev/resources/tokens.py @@ -0,0 +1,57 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING, Optional + +from luziadev.models import Token, TokenListResponse + +if TYPE_CHECKING: + from luziadev.client import Luzia + + +class TokensResource: + """List and look up canonical assets and on-chain tokens.""" + + def __init__(self, client: Luzia) -> None: + self._client = client + + async def list( + self, + *, + search: Optional[str] = None, + chain_id: Optional[str] = None, + has_chain: Optional[bool] = None, + page: int = 1, + limit: int = 20, + ) -> TokenListResponse: + """List tokens with optional filtering and pagination. + + Args: + search: Case-insensitive search across symbol, name, and id. + chain_id: Filter by chain id (e.g. ``"ethereum"``). + has_chain: ``True`` returns on-chain tokens only, ``False`` returns + chainless canonical tokens only. Omit for both. + page: Page number, 1-based. + limit: Items per page, max 100. + + Example: + >>> page = await luzia.tokens.list(search="USDC") + >>> for t in page.data: + ... print(t.id, t.chain_id) + """ + query: dict = {"page": page, "limit": limit} + if search is not None: + query["search"] = search + if chain_id is not None: + query["chainId"] = chain_id + if has_chain is not None: + query["hasChain"] = "true" if has_chain else "false" + data = await self._client.request("/v1/tokens", query=query) + return TokenListResponse.from_dict(data) + + async def get(self, token_id: str) -> Token: + """Look up a single token by composite id (e.g. ``"crypto:USDC"``).""" + # The colon needs URL-encoding when embedded in a path segment + from urllib.parse import quote + + data = await self._client.request(f"/v1/tokens/{quote(token_id, safe='')}") + return Token.from_dict(data.get("data", {}))