from typing import Callable
from celestia.types.das import SamplingStats
from celestia.node_api.rpc.abc import Wrapper
[docs]
class DasClient(Wrapper):
""" Client for interacting with Celestia's Das API."""
[docs]
async def sampling_stats(self, *, deserializer: Callable | None = None) -> SamplingStats:
""" Returns the current statistics over the DA sampling process.
Args:
deserializer (Callable | None): Custom deserializer. Defaults to :meth:`~celestia.types.das.SamplingStats.deserializer`.
Returns:
SamplingStats: The current sampling statistics.
"""
deserializer = deserializer if deserializer is not None else SamplingStats.deserializer
return await self._rpc.call("das.SamplingStats", (), deserializer)
[docs]
async def wait_catch_up(self) -> None:
""" Blocks until DASer finishes catching up to the network head.
Returns:
None
"""
return await self._rpc.call("das.WaitCatchUp")