Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions src/luziadev/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -40,6 +45,11 @@
"RetryOptions",
"RetryContext",
"Token",
"TokenListResponse",
"FiatCurrency",
"FiatCurrencyListResponse",
"EnabledFilter",
"Pagination",
"ErrorCode",
"is_luzia_error",
"is_retryable_error",
Expand Down
4 changes: 4 additions & 0 deletions src/luziadev/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand Down Expand Up @@ -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]:
Expand Down
95 changes: 90 additions & 5 deletions src/luziadev/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -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", {})),
)


Expand Down
54 changes: 54 additions & 0 deletions src/luziadev/resources/fiat_currencies.py
Original file line number Diff line number Diff line change
@@ -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", {}))
57 changes: 57 additions & 0 deletions src/luziadev/resources/tokens.py
Original file line number Diff line number Diff line change
@@ -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", {}))