glclient.lsps 

  1from dataclasses import dataclass, is_dataclass, asdict, field
  2
  3import typing as t
  4import json
  5import time
  6import binascii
  7
  8import glclient.glclient as native
  9
 10import logging
 11
 12logger = logging.getLogger(__name__)
 13
 14class EnhancedJSONEncoder(json.JSONEncoder):
 15    def default(self, o):
 16        if is_dataclass(o):
 17            return asdict(o)
 18        elif isinstance(o, NoParams):
 19            return dict()
 20        elif isinstance(o, type) and o.__name__ == "NoParams":
 21            return dict()
 22        return super().default(o)
 23
 24
 25class AsDataClassDescriptor:
 26    """Descriptor that allows to initialize a nested dataclass from a nested directory"""
 27
 28    def __init__(self, *, cls):
 29        self._cls = cls
 30
 31    def __set_name__(self, owner, name):
 32        self._name = f"_{name}"
 33
 34    def __get__(self, obj, type):
 35        return getattr(obj, self._name, None)
 36
 37    def __set__(self, obj, value):
 38        if isinstance(value, self._cls):
 39            setattr(obj, self._name, value)
 40        else:
 41            setattr(obj, self._name, self._cls(**value))
 42
 43
 44def _dump_json_bytes(object: t.Any) -> bytes:
 45    json_str: str = json.dumps(object, cls=EnhancedJSONEncoder)
 46    json_bytes: bytes = json_str.encode("utf-8")
 47    return json_bytes
 48
 49
 50@dataclass
 51class ProtocolList:
 52    protocols: t.List[int]
 53
 54
 55@dataclass
 56class Lsps1Options:
 57    minimum_channel_confirmations: t.Optional[int]
 58    minimum_onchain_payment_confirmations: t.Optional[int]
 59    supports_zero_channel_reserve: t.Optional[bool]
 60    min_onchain_payment_size_sat: t.Optional[int]
 61    max_channel_expiry_blocks: t.Optional[int]
 62    min_initial_client_balance_sat: t.Optional[int]
 63    min_initial_lsp_balance_sat: t.Optional[int]
 64    max_initial_client_balance_sat: t.Optional[int]
 65    min_channel_balance_sat: t.Optional[int]
 66    max_channel_balance_sat: t.Optional[int]
 67
 68
 69class NoParams:
 70    pass
 71
 72
 73class LspClient:
 74    def __init__(self, native: native.LspClient):
 75        self._native = native
 76
 77    def _rpc_call(
 78        self,
 79        peer_id: str,
 80        method_name: str,
 81        param_json: bytes,
 82        json_rpc_id: t.Optional[str] = None,
 83    ) -> bytes:
 84        logger.debug("Request lsp to peer %s and method %s", peer_id, method_name)
 85        peer_id_bytes = bytes.fromhex(peer_id)
 86        if json_rpc_id is None:
 87            return self._native.rpc_call(peer_id_bytes, method_name, param_json)
 88        else:
 89            return self._native.rpc_call_with_json_rpc_id(
 90                peer_id_bytes, method_name, param_json, json_rpc_id=json_rpc_id
 91            )
 92
 93    def list_lsp_servers(self) -> t.List[str]:
 94        return self._native.list_lsp_servers()
 95
 96    def list_protocols(self, peer_id, json_rpc_id: t.Optional[str] = None) -> ProtocolList:
 97        json_bytes = _dump_json_bytes(NoParams)
 98        result = self._rpc_call(
 99            peer_id, "lsps0.list_protocols", json_bytes, json_rpc_id=json_rpc_id
100        )
101        response_dict = json.loads(result)
102        return ProtocolList(**response_dict)
class EnhancedJSONEncoder(json.encoder.JSONEncoder): 
15class EnhancedJSONEncoder(json.JSONEncoder):
16    def default(self, o):
17        if is_dataclass(o):
18            return asdict(o)
19        elif isinstance(o, NoParams):
20            return dict()
21        elif isinstance(o, type) and o.__name__ == "NoParams":
22            return dict()
23        return super().default(o)

Extensible JSON http://json.org encoder for Python data structures.

Supports the following objects and types by default:

+-------------------+---------------+ | Python | JSON | +===================+===============+ | dict | object | +-------------------+---------------+ | list, tuple | array | +-------------------+---------------+ | str | string | +-------------------+---------------+ | int, float | number | +-------------------+---------------+ | True | true | +-------------------+---------------+ | False | false | +-------------------+---------------+ | None | null | +-------------------+---------------+

To extend this to recognize other objects, subclass and implement a .default() method with another method that returns a serializable object for o if possible, otherwise it should call the superclass implementation (to raise TypeError).

def default(self, o): 
16    def default(self, o):
17        if is_dataclass(o):
18            return asdict(o)
19        elif isinstance(o, NoParams):
20            return dict()
21        elif isinstance(o, type) and o.__name__ == "NoParams":
22            return dict()
23        return super().default(o)

Implement this method in a subclass such that it returns a serializable object for o, or calls the base implementation (to raise a TypeError).

For example, to support arbitrary iterators, you could implement default like this::

def default(self, o):
    try:
        iterable = iter(o)
    except TypeError:
        pass
    else:
        return list(iterable)
    # Let the base class default method raise the TypeError
    return JSONEncoder.default(self, o)
Inherited Members
json.encoder.JSONEncoder
JSONEncoder
encode
iterencode
class AsDataClassDescriptor: 
26class AsDataClassDescriptor:
27    """Descriptor that allows to initialize a nested dataclass from a nested directory"""
28
29    def __init__(self, *, cls):
30        self._cls = cls
31
32    def __set_name__(self, owner, name):
33        self._name = f"_{name}"
34
35    def __get__(self, obj, type):
36        return getattr(obj, self._name, None)
37
38    def __set__(self, obj, value):
39        if isinstance(value, self._cls):
40            setattr(obj, self._name, value)
41        else:
42            setattr(obj, self._name, self._cls(**value))

Descriptor that allows to initialize a nested dataclass from a nested directory

AsDataClassDescriptor(*, cls) 
29    def __init__(self, *, cls):
30        self._cls = cls
@dataclass
class ProtocolList: 
51@dataclass
52class ProtocolList:
53    protocols: t.List[int]
ProtocolList(protocols: List[int])
@dataclass
class Lsps1Options: 
56@dataclass
57class Lsps1Options:
58    minimum_channel_confirmations: t.Optional[int]
59    minimum_onchain_payment_confirmations: t.Optional[int]
60    supports_zero_channel_reserve: t.Optional[bool]
61    min_onchain_payment_size_sat: t.Optional[int]
62    max_channel_expiry_blocks: t.Optional[int]
63    min_initial_client_balance_sat: t.Optional[int]
64    min_initial_lsp_balance_sat: t.Optional[int]
65    max_initial_client_balance_sat: t.Optional[int]
66    min_channel_balance_sat: t.Optional[int]
67    max_channel_balance_sat: t.Optional[int]
Lsps1Options( minimum_channel_confirmations: Optional[int], minimum_onchain_payment_confirmations: Optional[int], supports_zero_channel_reserve: Optional[bool], min_onchain_payment_size_sat: Optional[int], max_channel_expiry_blocks: Optional[int], min_initial_client_balance_sat: Optional[int], min_initial_lsp_balance_sat: Optional[int], max_initial_client_balance_sat: Optional[int], min_channel_balance_sat: Optional[int], max_channel_balance_sat: Optional[int])
class NoParams: 
70class NoParams:
71    pass
class LspClient: 
 74class LspClient:
 75    def __init__(self, native: native.LspClient):
 76        self._native = native
 77
 78    def _rpc_call(
 79        self,
 80        peer_id: str,
 81        method_name: str,
 82        param_json: bytes,
 83        json_rpc_id: t.Optional[str] = None,
 84    ) -> bytes:
 85        logger.debug("Request lsp to peer %s and method %s", peer_id, method_name)
 86        peer_id_bytes = bytes.fromhex(peer_id)
 87        if json_rpc_id is None:
 88            return self._native.rpc_call(peer_id_bytes, method_name, param_json)
 89        else:
 90            return self._native.rpc_call_with_json_rpc_id(
 91                peer_id_bytes, method_name, param_json, json_rpc_id=json_rpc_id
 92            )
 93
 94    def list_lsp_servers(self) -> t.List[str]:
 95        return self._native.list_lsp_servers()
 96
 97    def list_protocols(self, peer_id, json_rpc_id: t.Optional[str] = None) -> ProtocolList:
 98        json_bytes = _dump_json_bytes(NoParams)
 99        result = self._rpc_call(
100            peer_id, "lsps0.list_protocols", json_bytes, json_rpc_id=json_rpc_id
101        )
102        response_dict = json.loads(result)
103        return ProtocolList(**response_dict)
LspClient(native: LspClient) 
75    def __init__(self, native: native.LspClient):
76        self._native = native
def list_lsp_servers(self) -> List[str]: 
94    def list_lsp_servers(self) -> t.List[str]:
95        return self._native.list_lsp_servers()
def list_protocols( self, peer_id, json_rpc_id: Optional[str] = None) -> glclient.lsps.ProtocolList: 
 97    def list_protocols(self, peer_id, json_rpc_id: t.Optional[str] = None) -> ProtocolList:
 98        json_bytes = _dump_json_bytes(NoParams)
 99        result = self._rpc_call(
100            peer_id, "lsps0.list_protocols", json_bytes, json_rpc_id=json_rpc_id
101        )
102        response_dict = json.loads(result)
103        return ProtocolList(**response_dict)