Skip to content

Backend Base

Abstract base class for Cardano blockchain backend implementations.

AbstractBackend

Bases: ABC

Abstract base class for Cardano blockchain backend implementations.

This class defines the interface for interacting with various Cardano blockchain data sources such as db-sync, Ogmios, or Kupo.

Source code in src/charli3_dendrite/backend/backend_base.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
class AbstractBackend(ABC):
    """Abstract base class for Cardano blockchain backend implementations.

    This class defines the interface for interacting with various Cardano blockchain
    data sources such as db-sync, Ogmios, or Kupo.
    """

    @abstractmethod
    def get_pool_utxos(  # noqa: PLR0913
        self,
        assets: list[str] | None = None,
        addresses: list[str] | None = None,
        limit: int = 1000,
        page: int = 0,
        historical: bool = True,
    ) -> PoolStateList:
        """Get UTXOs for specific assets or addresses.

        Args:
            assets (Optional[List[str]]): List of asset IDs to filter by.
            addresses (Optional[List[str]]): List of addresses to filter by.
            limit (int): Maximum number of results to return.
            page (int): Page number for pagination.
            historical (bool): Whether to include historical data.

        Returns:
            PoolStateList: List of pool state objects.
        """
        pass

    @abstractmethod
    def get_pool_in_tx(
        self,
        tx_hash: str,
        assets: list[str] | None = None,
        addresses: list[str] | None = None,
    ) -> PoolStateList:
        """Get pool state for a specific transaction.

        Args:
            tx_hash (str): The transaction hash to query.
            assets (Optional[List[str]]): List of asset IDs to filter by.
            addresses (Optional[List[str]]): List of addresses to filter by.

        Returns:
            PoolStateList: List of pool state objects for the transaction.
        """
        pass

    @abstractmethod
    def last_block(self, last_n_blocks: int = 2) -> BlockList:
        """Get information about the last n blocks.

        Args:
            last_n_blocks (int): Number of recent blocks to retrieve.

        Returns:
            BlockList: List of recent block information.
        """
        pass

    @abstractmethod
    def get_pool_utxos_in_block(self, block_no: int) -> PoolStateList:
        """Get pool UTXOs for a specific block.

        Args:
            block_no (int): The block number to query.

        Returns:
            PoolStateList: List of pool state objects for the block.
        """
        pass

    @abstractmethod
    def get_script_from_address(self, address: Address) -> ScriptReference:
        """Get script reference for a given address.

        Args:
            address (Address): The address to query.

        Returns:
            ScriptReference: Script reference for the address.
        """
        pass

    @abstractmethod
    def get_historical_order_utxos(
        self,
        stake_addresses: list[str],
        after_time: datetime | int | None = None,
        limit: int = 1000,
        page: int = 0,
    ) -> SwapTransactionList:
        """Get historical order UTXOs for given stake addresses.

        Args:
            stake_addresses (List[str]): List of stake addresses to query.
            after_time (Optional[Union[datetime, int]]): Filter results after this time.
            limit (int): Maximum number of results to return.
            page (int): Page number for pagination.

        Returns:
            SwapTransactionList: List of swap transaction objects.
        """
        pass

    @abstractmethod
    def get_order_utxos_by_block_or_tx(  # noqa: PLR0913
        self,
        stake_addresses: list[str],
        out_tx_hash: list[str] | None = None,
        in_tx_hash: list[str] | None = None,
        block_no: int | None = None,
        after_block: int | None = None,
        limit: int = 1000,
        page: int = 0,
    ) -> SwapTransactionList:
        """Get order UTXOs by block or transaction.

        Args:
            stake_addresses (List[str]): List of stake addresses to query.
            out_tx_hash (Optional[List[str]]): List of transaction hashes to filter by.
            in_tx_hash: list of input transaction hashes to filter by.
            block_no (Optional[int]): Specific block number to query.
            after_block (Optional[int]): Filter results after this block number.
            limit (int): Maximum number of results to return.
            page (int): Page number for pagination.

        Returns:
            SwapTransactionList: List of swap transaction objects.
        """
        pass

    @abstractmethod
    def get_cancel_utxos(  # noqa: PLR0913
        self,
        stake_addresses: list[str],
        block_no: int | None = None,
        after_time: datetime | int | None = None,
        limit: int = 1000,
        page: int = 0,
    ) -> SwapTransactionList:
        """Get cancelled order UTXOs.

        Args:
            stake_addresses (List[str]): List of stake addresses to query.
            block_no (Optional[int]): Specific block number to query.
            after_time (Optional[Union[datetime, int]]): Filter results after this time.
            limit (int): Maximum number of results to return.
            page (int): Page number for pagination.

        Returns:
            SwapTransactionList: List of swap transaction objects for cancelled orders.
        """
        pass

    @abstractmethod
    def get_datum_from_address(
        self,
        address: str,
        asset: str | None = None,
    ) -> str | None:
        """Get datum from a given address.

        Args:
            address: The address to query.
            asset: Assets required to be in the UTxO.

        Returns:
            The datum associated with the address, if any.
        """
        pass

get_cancel_utxos(stake_addresses: list[str], block_no: int | None = None, after_time: datetime | int | None = None, limit: int = 1000, page: int = 0) -> SwapTransactionList abstractmethod

Get cancelled order UTXOs.

Parameters:

Name Type Description Default
stake_addresses List[str]

List of stake addresses to query.

 required
block_no Optional[int]

Specific block number to query.

 None
after_time Optional[Union[datetime, int]]

Filter results after this time.

 None
limit int

Maximum number of results to return.

 1000
page int

Page number for pagination.

 0

Returns:

Name Type Description
SwapTransactionList SwapTransactionList

List of swap transaction objects for cancelled orders.
Source code in src/charli3_dendrite/backend/backend_base.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
@abstractmethod
def get_cancel_utxos(  # noqa: PLR0913
    self,
    stake_addresses: list[str],
    block_no: int | None = None,
    after_time: datetime | int | None = None,
    limit: int = 1000,
    page: int = 0,
) -> SwapTransactionList:
    """Get cancelled order UTXOs.

    Args:
        stake_addresses (List[str]): List of stake addresses to query.
        block_no (Optional[int]): Specific block number to query.
        after_time (Optional[Union[datetime, int]]): Filter results after this time.
        limit (int): Maximum number of results to return.
        page (int): Page number for pagination.

    Returns:
        SwapTransactionList: List of swap transaction objects for cancelled orders.
    """
    pass

get_datum_from_address(address: str, asset: str | None = None) -> str | None abstractmethod

Get datum from a given address.

Parameters:

Name Type Description Default
address str

The address to query.

 required
asset str | None

Assets required to be in the UTxO.

 None

Returns:

Type Description
str | None

The datum associated with the address, if any.
Source code in src/charli3_dendrite/backend/backend_base.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
@abstractmethod
def get_datum_from_address(
    self,
    address: str,
    asset: str | None = None,
) -> str | None:
    """Get datum from a given address.

    Args:
        address: The address to query.
        asset: Assets required to be in the UTxO.

    Returns:
        The datum associated with the address, if any.
    """
    pass

get_historical_order_utxos(stake_addresses: list[str], after_time: datetime | int | None = None, limit: int = 1000, page: int = 0) -> SwapTransactionList abstractmethod

Get historical order UTXOs for given stake addresses.

Parameters:

Name Type Description Default
stake_addresses List[str]

List of stake addresses to query.

 required
after_time Optional[Union[datetime, int]]

Filter results after this time.

 None
limit int

Maximum number of results to return.

 1000
page int

Page number for pagination.

 0

Returns:

Name Type Description
SwapTransactionList SwapTransactionList

List of swap transaction objects.
Source code in src/charli3_dendrite/backend/backend_base.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
@abstractmethod
def get_historical_order_utxos(
    self,
    stake_addresses: list[str],
    after_time: datetime | int | None = None,
    limit: int = 1000,
    page: int = 0,
) -> SwapTransactionList:
    """Get historical order UTXOs for given stake addresses.

    Args:
        stake_addresses (List[str]): List of stake addresses to query.
        after_time (Optional[Union[datetime, int]]): Filter results after this time.
        limit (int): Maximum number of results to return.
        page (int): Page number for pagination.

    Returns:
        SwapTransactionList: List of swap transaction objects.
    """
    pass

get_order_utxos_by_block_or_tx(stake_addresses: list[str], out_tx_hash: list[str] | None = None, in_tx_hash: list[str] | None = None, block_no: int | None = None, after_block: int | None = None, limit: int = 1000, page: int = 0) -> SwapTransactionList abstractmethod

Get order UTXOs by block or transaction.

Parameters:

Name Type Description Default
stake_addresses List[str]

List of stake addresses to query.

 required
out_tx_hash Optional[List[str]]

List of transaction hashes to filter by.

 None
in_tx_hash list[str] | None

list of input transaction hashes to filter by.

 None
block_no Optional[int]

Specific block number to query.

 None
after_block Optional[int]

Filter results after this block number.

 None
limit int

Maximum number of results to return.

 1000
page int

Page number for pagination.

 0

Returns:

Name Type Description
SwapTransactionList SwapTransactionList

List of swap transaction objects.
Source code in src/charli3_dendrite/backend/backend_base.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
@abstractmethod
def get_order_utxos_by_block_or_tx(  # noqa: PLR0913
    self,
    stake_addresses: list[str],
    out_tx_hash: list[str] | None = None,
    in_tx_hash: list[str] | None = None,
    block_no: int | None = None,
    after_block: int | None = None,
    limit: int = 1000,
    page: int = 0,
) -> SwapTransactionList:
    """Get order UTXOs by block or transaction.

    Args:
        stake_addresses (List[str]): List of stake addresses to query.
        out_tx_hash (Optional[List[str]]): List of transaction hashes to filter by.
        in_tx_hash: list of input transaction hashes to filter by.
        block_no (Optional[int]): Specific block number to query.
        after_block (Optional[int]): Filter results after this block number.
        limit (int): Maximum number of results to return.
        page (int): Page number for pagination.

    Returns:
        SwapTransactionList: List of swap transaction objects.
    """
    pass

get_pool_in_tx(tx_hash: str, assets: list[str] | None = None, addresses: list[str] | None = None) -> PoolStateList abstractmethod

Get pool state for a specific transaction.

Parameters:

Name Type Description Default
tx_hash str

The transaction hash to query.

 required
assets Optional[List[str]]

List of asset IDs to filter by.

 None
addresses Optional[List[str]]

List of addresses to filter by.

 None

Returns:

Name Type Description
PoolStateList PoolStateList

List of pool state objects for the transaction.
Source code in src/charli3_dendrite/backend/backend_base.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
@abstractmethod
def get_pool_in_tx(
    self,
    tx_hash: str,
    assets: list[str] | None = None,
    addresses: list[str] | None = None,
) -> PoolStateList:
    """Get pool state for a specific transaction.

    Args:
        tx_hash (str): The transaction hash to query.
        assets (Optional[List[str]]): List of asset IDs to filter by.
        addresses (Optional[List[str]]): List of addresses to filter by.

    Returns:
        PoolStateList: List of pool state objects for the transaction.
    """
    pass

get_pool_utxos(assets: list[str] | None = None, addresses: list[str] | None = None, limit: int = 1000, page: int = 0, historical: bool = True) -> PoolStateList abstractmethod

Get UTXOs for specific assets or addresses.

Parameters:

Name Type Description Default
assets Optional[List[str]]

List of asset IDs to filter by.

 None
addresses Optional[List[str]]

List of addresses to filter by.

 None
limit int

Maximum number of results to return.

 1000
page int

Page number for pagination.

 0
historical bool

Whether to include historical data.

 True

Returns:

Name Type Description
PoolStateList PoolStateList

List of pool state objects.
Source code in src/charli3_dendrite/backend/backend_base.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
@abstractmethod
def get_pool_utxos(  # noqa: PLR0913
    self,
    assets: list[str] | None = None,
    addresses: list[str] | None = None,
    limit: int = 1000,
    page: int = 0,
    historical: bool = True,
) -> PoolStateList:
    """Get UTXOs for specific assets or addresses.

    Args:
        assets (Optional[List[str]]): List of asset IDs to filter by.
        addresses (Optional[List[str]]): List of addresses to filter by.
        limit (int): Maximum number of results to return.
        page (int): Page number for pagination.
        historical (bool): Whether to include historical data.

    Returns:
        PoolStateList: List of pool state objects.
    """
    pass

get_pool_utxos_in_block(block_no: int) -> PoolStateList abstractmethod

Get pool UTXOs for a specific block.

Parameters:

Name Type Description Default
block_no int

The block number to query.

 required

Returns:

Name Type Description
PoolStateList PoolStateList

List of pool state objects for the block.
Source code in src/charli3_dendrite/backend/backend_base.py 
76
77
78
79
80
81
82
83
84
85
86
@abstractmethod
def get_pool_utxos_in_block(self, block_no: int) -> PoolStateList:
    """Get pool UTXOs for a specific block.

    Args:
        block_no (int): The block number to query.

    Returns:
        PoolStateList: List of pool state objects for the block.
    """
    pass

get_script_from_address(address: Address) -> ScriptReference abstractmethod

Get script reference for a given address.

Parameters:

Name Type Description Default
address Address

The address to query.

 required

Returns:

Name Type Description
ScriptReference ScriptReference

Script reference for the address.
Source code in src/charli3_dendrite/backend/backend_base.py 
88
89
90
91
92
93
94
95
96
97
98
@abstractmethod
def get_script_from_address(self, address: Address) -> ScriptReference:
    """Get script reference for a given address.

    Args:
        address (Address): The address to query.

    Returns:
        ScriptReference: Script reference for the address.
    """
    pass

last_block(last_n_blocks: int = 2) -> BlockList abstractmethod

Get information about the last n blocks.

Parameters:

Name Type Description Default
last_n_blocks int

Number of recent blocks to retrieve.

 2

Returns:

Name Type Description
BlockList BlockList

List of recent block information.
Source code in src/charli3_dendrite/backend/backend_base.py 
64
65
66
67
68
69
70
71
72
73
74
@abstractmethod
def last_block(self, last_n_blocks: int = 2) -> BlockList:
    """Get information about the last n blocks.

    Args:
        last_n_blocks (int): Number of recent blocks to retrieve.

    Returns:
        BlockList: List of recent block information.
    """
    pass