All checks were successful
Publish to PyPI Registry / publish (release) Successful in 1m14s
proof-measure is a separate, public, unauthenticated Dragonchain service. Adds: - UnauthenticatedClient: HMAC-free transport mirroring Client (session injection, allow_redirects=False, from_dict decoding). - ProofMeasureClient: get_security / report / health; default base URL https://proof-measure.dragonchain.com. Standalone (ProofMeasureClient()) and via DragonchainSDK.proof_measure. - Proof-measure dataclass models (decimals as strings, timestamps as int). - .gitea/workflows/publish.yml: build + twine upload to the Gitea PyPI registry on release (the SDK had no publish workflow before). Also fixes 3 pre-existing failing tests: _FakeSession.request didn't accept the allow_redirects kwarg the client now passes (added by the prior redirect change), so the suite was red at HEAD.
65 lines
2.3 KiB
Python
65 lines
2.3 KiB
Python
"""Client for the Dragonchain proof-measure service.
|
||
|
||
proof-measure is the measured-immutability / "securedBy" metric for L1–L5
|
||
verification chains. It is a separate, public, UNauthenticated service (no API
|
||
keys), so this client needs only a base URL. It exposes a network's accumulated
|
||
security as both a raw measure (cumulative hashes / stake-seconds) and a USD
|
||
valuation, plus a per-transaction "securedBy" report over interchain anchors.
|
||
"""
|
||
|
||
from typing import Optional
|
||
|
||
import requests
|
||
|
||
from .client import DEFAULT_TIMEOUT
|
||
from .models import (
|
||
CONTENT_TYPE_JSON,
|
||
HealthResponse,
|
||
ReportRequest,
|
||
SecurityResult,
|
||
TransactionReport,
|
||
)
|
||
from .unauthenticated_client import UnauthenticatedClient
|
||
|
||
#: Public production proof-measure endpoint.
|
||
DEFAULT_BASE_URL = "https://proof-measure.dragonchain.com"
|
||
|
||
|
||
class ProofMeasureClient:
|
||
"""Calls the proof-measure HTTP API.
|
||
|
||
With no arguments it targets the public production endpoint, so
|
||
``ProofMeasureClient()`` works with zero configuration.
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
base_url: str = DEFAULT_BASE_URL,
|
||
timeout: int = DEFAULT_TIMEOUT,
|
||
session: "Optional[requests.Session]" = None,
|
||
):
|
||
self._client = UnauthenticatedClient(
|
||
base_url or DEFAULT_BASE_URL, timeout=timeout, session=session
|
||
)
|
||
|
||
def get_security(self, network: str, since: Optional[int] = None) -> SecurityResult:
|
||
"""Return the security a public network (``network`` = "BTC" or "ETH")
|
||
has accumulated since ``since`` (unix seconds), as a raw measure + USD
|
||
valuation. Omit ``since`` to use the service's default window."""
|
||
path = f"/api/v1/security/{network}"
|
||
if since and since > 0:
|
||
path += f"?since={since}"
|
||
return self._client.get(path, SecurityResult)
|
||
|
||
def report(self, req: ReportRequest) -> TransactionReport:
|
||
"""Compute the per-transaction "securedBy" report for the supplied
|
||
interchain anchors: each anchor's raw + USD security since it was placed,
|
||
plus combined totals."""
|
||
return self._client.post(
|
||
"/api/v1/report", CONTENT_TYPE_JSON, req, TransactionReport
|
||
)
|
||
|
||
def health(self) -> HealthResponse:
|
||
"""Report service liveness and DB reachability."""
|
||
return self._client.get("/api/v1/health", HealthResponse)
|