duniterpy.api package

Subpackages

Submodules

duniterpy.api.client module

class duniterpy.api.client.API(connection_handler: ConnectionHandler, headers: dict | None = None)

Bases: object

API is a class used as an abstraction layer over the http/websocket libraries.

connect_ws(path: str) → WSConnection

Connect to a websocket

Parameters:

path – the url path

Returns:

request_url(path: str, method: str = 'GET', rtype: str = 'json', schema: dict | None = None, json_data: dict | None = None, bma_errors: bool = False, **kwargs: Any) → Any

Requests wrapper in order to use API parameters.

Parameters:

  • path – the request path

  • method – Http method ‘GET’ or ‘POST’ (optional, default=’GET’)

  • rtype – Response type (optional, default RESPONSE_JSON, can be RESPONSE_TEXT, RESPONSE_HTTP)

  • schema – Json Schema to validate response (optional, default None)

  • json_data – Json data as dict (optional, default None)

  • bma_errors – Set it to True to handle Duniter Error Response (optional, default False)

Returns:

reverse_url(scheme: str, path: str) → str

Reverses the url using scheme and path given in parameter.

Parameters:

  • scheme – Scheme of the url

  • path – Path of the url

Returns:

class duniterpy.api.client.Client(_endpoint: str | Endpoint, proxy: str | None = None)

Bases: object

Main class to create an API client

connect_ws(path: str = '') → WSConnection

Connect to a websocket in order to use API parameters

Parameters:

path – the url path

Returns:

get(url_path: str, params: dict | None = None, rtype: str = 'json', schema: dict | None = None) → Any

GET request on endpoint host + url_path

Parameters:

  • url_path – Url encoded path following the endpoint

  • params – Url query string parameters dictionary (optional, default None)

  • rtype – Response type (optional, default RESPONSE_JSON, can be RESPONSE_TEXT, RESPONSE_HTTP)

  • schema – Json Schema to validate response (optional, default None)

Returns:

post(url_path: str, params: dict | None = None, rtype: str = 'json', schema: dict | None = None) → Any

POST request on endpoint host + url_path

Parameters:

  • url_path – Url encoded path following the endpoint

  • params – Url query string parameters dictionary (optional, default None)

  • rtype – Response type (optional, default RESPONSE_JSON, can be RESPONSE_TEXT, RESPONSE_HTTP)

  • schema – Json Schema to validate response (optional, default None)

Returns:

query(query: str, variables: dict | None = None, schema: dict | None = None) → Any

GraphQL query or mutation request on endpoint

Parameters:

  • query – GraphQL query string

  • variables – Variables for the query (optional, default None)

  • schema – Json Schema to validate response (optional, default None)

Returns:

class duniterpy.api.client.WSConnection(connection: WebSocket)

Bases: object

Abstraction layer on websocket library

close() → None

Close the web socket connection

Returns:

receive_json(timeout: float | None = None) → Any

Wait for json data from the web socket connection

Parameters:

timeout – Timeout in seconds

Returns:

receive_str(timeout: float | None = None) → str

Wait for a data string from the web socket connection

Parameters:

timeout – Timeout in seconds

Returns:

send_str(data: str)

Send a data string to the web socket connection

Parameters:

data – Data string

Returns:

exception duniterpy.api.client.WSConnectionException

Bases: Exception

duniterpy.api.client.parse_error(text: str) → dict

Validate and parse the BMA answer from websocket

Parameters:

text – the bma error

Returns:

the json data

duniterpy.api.client.parse_response(response: str, schema: dict) → Any

Validate and parse the BMA answer

Parameters:

  • response – Response content

  • schema – The expected response structure

Returns:

the json data

duniterpy.api.client.parse_text(text: str, schema: dict) → Any

Validate and parse the BMA answer from websocket

Parameters:

  • text – the bma answer

  • schema – dict for jsonschema

Returns:

the json data

duniterpy.api.endpoint module

class duniterpy.api.endpoint.BMAEndpoint(host: str, ipv4: str, ipv6: str, port: int)

Bases: Endpoint

API = 'BASIC_MERKLED_API'
conn_handler(proxy: str | None = None) → ConnectionHandler

Return connection handler instance for the endpoint

Parameters:

proxy – Proxy url

Returns:

classmethod from_inline(inline: str) → BMAEndpointType

Return BMAEndpoint instance from endpoint string

Parameters:

inline – Endpoint string

Returns:

inline() → str

Return endpoint string

Returns:

re_inline = re.compile('^BASIC_MERKLED_API(?: (?P<host>(((?!-))(xn--|_)?[a-z0-9-]{0,61}[a-z0-9]\\.)*(xn--)?([a-z0-9][a-z0-9\\-]{0,60}|[a-z0-9-]{1,30}\\.[a-z]{2,})))?(?: (?P<ipv4>(?:(?:25[0-5]|(?:2[0-4]|1?\\d)?\\d)\\.){3}(?:)
class duniterpy.api.endpoint.ConnectionHandler(http_scheme: str, ws_scheme: str, address: str, port: int, path: str, proxy: str | None = None)

Bases: object

Helper class used by other API classes to ease passing address connection information.

class duniterpy.api.endpoint.ESCoreEndpoint(host: str, port: int)

Bases: Endpoint

API = 'ES_CORE_API'
conn_handler(proxy: str | None = None) → ConnectionHandler

Return connection handler instance for the endpoint

Parameters:

proxy – Proxy url

Returns:

classmethod from_inline(inline: str) → ESCoreEndpointType

Return ESCoreEndpoint instance from endpoint string

Parameters:

inline – Endpoint string

Returns:

inline() → str

Return endpoint string

Returns:

re_inline = re.compile('^ES_CORE_API (?P<host>(?:(((?!-))(xn--|_)?[a-z0-9-]{0,61}[a-z0-9]\\.)*(xn--)?([a-z0-9][a-z0-9\\-]{0,60}|[a-z0-9-]{1,30}\\.[a-z]{2,}))|(?:(?:(?:25[0-5]|(?:2[0-4]|1?\\d)?\\d)\\.){3}(?:25[0-5]|(?:2[0-4])
class duniterpy.api.endpoint.ESSubscribtionEndpoint(host: str, port: int)

Bases: Endpoint

API = 'ES_SUBSCRIPTION_API'
conn_handler(proxy: str | None = None) → ConnectionHandler

Return connection handler instance for the endpoint

Parameters:

proxy – Proxy url

Returns:

classmethod from_inline(inline: str) → ESSubscribtionEndpointType

Return ESSubscribtionEndpoint instance from endpoint string

Parameters:

inline – Endpoint string

Returns:

inline() → str

Return endpoint string

Returns:

re_inline = re.compile('^ES_SUBSCRIPTION_API (?P<host>(?:(((?!-))(xn--|_)?[a-z0-9-]{0,61}[a-z0-9]\\.)*(xn--)?([a-z0-9][a-z0-9\\-]{0,60}|[a-z0-9-]{1,30}\\.[a-z]{2,}))|(?:(?:(?:25[0-5]|(?:2[0-4]|1?\\d)?\\d)\\.){3}(?:25[0-5]|()
class duniterpy.api.endpoint.ESUserEndpoint(host: str, port: int)

Bases: Endpoint

API = 'ES_USER_API'
conn_handler(proxy: str | None = None) → ConnectionHandler

Return connection handler instance for the endpoint

Parameters:

proxy – Proxy url

Returns:

classmethod from_inline(inline: str) → ESUserEndpointType

Return ESUserEndpoint instance from endpoint string

Parameters:

inline – Endpoint string

Returns:

inline() → str

Return endpoint string

Returns:

re_inline = re.compile('^ES_USER_API (?P<host>(?:(((?!-))(xn--|_)?[a-z0-9-]{0,61}[a-z0-9]\\.)*(xn--)?([a-z0-9][a-z0-9\\-]{0,60}|[a-z0-9-]{1,30}\\.[a-z]{2,}))|(?:(?:(?:25[0-5]|(?:2[0-4]|1?\\d)?\\d)\\.){3}(?:25[0-5]|(?:2[0-4])
class duniterpy.api.endpoint.Endpoint

Bases: object

conn_handler(proxy: str | None = None) → ConnectionHandler
classmethod from_inline(inline: str) → EndpointType
inline() → str
class duniterpy.api.endpoint.GVAEndpoint(flags: str, host: str, ipv4: str, ipv6: str, port: int, path: str)

Bases: Endpoint

API = 'GVA'
conn_handler(proxy: str | None = None) → ConnectionHandler

Return connection handler instance for the endpoint

Parameters:

proxy – Proxy url

Returns:

endpoint_format = '^GVA(?: (?P<flags>[S]))?(?: (?P<host>(((?!-))(xn--|_)?[a-z0-9-]{0,61}[a-z0-9]\\.)*(xn--)?([a-z0-9][a-z0-9\\-]{0,60}|[a-z0-9-]{1,30}\\.[a-z]{2,})))?(?: (?P<ipv4>(?:(?:25[0-5]|(?:2[0-4]|1?\\d)?\\d)\\.){3}(?:25[0-5]|(?:2[0-4]|1?\\d)?\\d)))?(?: (?P<ipv6>(?:[0-9a-fA-F]{1,4}:){7}[0-9a-fA-F]{1,4}|(?:[0-9a-fA-F]{1,4}:){1,7}:|(?:[0-9a-fA-F]{1,4}:){1,6}:[0-9a-fA-F]{1,4}|(?:[0-9a-fA-F]{1,4}:){1,5}(?::[0-9a-fA-F]{1,4}){1,2}|(?:[0-9a-fA-F]{1,4}:){1,4}(?::[0-9a-fA-F]{1,4}){1,3}|(?:[0-9a-fA-F]{1,4}:){1,3}(?::[0-9a-fA-F]{1,4}){1,4}|(?:[0-9a-fA-F]{1,4}:){1,2}(?::[0-9a-fA-F]{1,4}){1,5}|[0-9a-fA-F]{1,4}:(?:(?::[0-9a-fA-F]{1,4}){1,6})|:(?:(?::[0-9a-fA-F]{1,4}){1,7}|:)|fe80:(?::[0-9a-fA-F]{1,4}){0,4}%[0-9a-zA-Z]+|::(?:ffff(?::0{1,4})?:)?(?:(?:25[0-5]|(?:2[0-4]|1?\\d)?\\d)\\.){3}(?:25[0-5]|(?:2[0-4]|1?\\d)?\\d)|(?:[0-9a-fA-F]{1,4}:){1,4}:(?:(?:25[0-5]|(?:2[0-4]|1?\\d)?\\d)\\.){3}(?:25[0-5]|(?:2[0-4]|1?\\d)?\\d)))? (?P<port>[1-9]\\d{0,3}|0|[1-5]\\d{4}|6[0-4]\\d{3}|65[0-4]\\d{2}|655[0-2]\\d|6553[0-5])(?: (?P<path>[/\\w \\.-]*/?))?$'
classmethod from_inline(inline: str) → GVAEndpointType

Return GVAEndpoint instance from endpoint string

Parameters:

inline – Endpoint string

Returns:

inline() → str

Return endpoint string

Returns:

re_inline = re.compile('^GVA(?: (?P<flags>[S]))?(?: (?P<host>(((?!-))(xn--|_)?[a-z0-9-]{0,61}[a-z0-9]\\.)*(xn--)?([a-z0-9][a-z0-9\\-]{0,60}|[a-z0-9-]{1,30}\\.[a-z]{2,})))?(?: (?P<ipv4>(?:(?:25[0-5]|(?:2[0-4]|1?\\d)?\\d)\\.))
class duniterpy.api.endpoint.SecuredBMAEndpoint(host: str, ipv4: str, ipv6: str, port: int, path: str)

Bases: BMAEndpoint

API = 'BMAS'
conn_handler(proxy: str | None = None) → ConnectionHandler

Return connection handler instance for the endpoint

Parameters:

proxy – Proxy url

Returns:

classmethod from_inline(inline: str) → SecuredBMAEndpointType

Return SecuredBMAEndpoint instance from endpoint string

Parameters:

inline – Endpoint string

Returns:

inline() → str

Return endpoint string

Returns:

re_inline = re.compile('^BMAS(?: (?P<host>(((?!-))(xn--|_)?[a-z0-9-]{0,61}[a-z0-9]\\.)*(xn--)?([a-z0-9][a-z0-9\\-]{0,60}|[a-z0-9-]{1,30}\\.[a-z]{2,})))?(?: (?P<ipv4>(?:(?:25[0-5]|(?:2[0-4]|1?\\d)?\\d)\\.){3}(?:25[0-5]|(?:2[)
class duniterpy.api.endpoint.UnknownEndpoint(api: str, properties: list)

Bases: Endpoint

API = None
conn_handler(proxy: str | None = None) → ConnectionHandler

Return connection handler

Parameters:

proxy – Proxy address

Returns:

classmethod from_inline(inline: str) → UnknownEndpointType

Return UnknownEndpoint instance from endpoint string

Parameters:

inline – Endpoint string

Returns:

inline() → str

Return endpoint string

Returns:

class duniterpy.api.endpoint.WS2PEndpoint(ws2pid: str, host: str, port: int, path: str)

Bases: Endpoint

API = 'WS2P'
conn_handler(proxy: str | None = None) → ConnectionHandler

Return connection handler instance for the endpoint

Parameters:

proxy – Proxy url

Returns:

classmethod from_inline(inline: str) → WS2PEndpointType

Return WS2PEndpoint instance from endpoint string

Parameters:

inline – Endpoint string

Returns:

inline() → str

Return endpoint string

Returns:

re_inline = re.compile('^WS2P (?P<ws2pid>[0-9a-f]{8}) (?P<host>(?:(((?!-))(xn--|_)?[a-z0-9-]{0,61}[a-z0-9]\\.)*(xn--)?([a-z0-9][a-z0-9\\-]{0,60}|[a-z0-9-]{1,30}\\.[a-z]{2,}))|(?:(?:(?:25[0-5]|(?:2[0-4]|1?\\d)?\\d)\\.){3}(?:)
duniterpy.api.endpoint.endpoint(value: Any) → Any

Convert an endpoint string to the corresponding Endpoint instance type

Parameters:

value – Endpoint string or subclass

Returns:

duniterpy.api.endpoint.fix_host_ipv4_mix_up(host: str, ipv4: str) → Tuple[str, str]

duniterpy.api.errors module

exception duniterpy.api.errors.DuniterError(data: dict)

Bases: Exception

Handle duniter error

Module contents