NetDiag Python Client | netdiag-client PyPI Package

Installation

pip install netdiag-client

Or using other package managers:

# Poetry
poetry add netdiag-client

# Pipenv
pipenv install netdiag-client

# uv
uv pip install netdiag-client

Requirements: Python 3.10+

Quick Start

from netdiag import NetDiagClient

# Use as context manager (recommended)
with NetDiagClient(api_key="your-api-key") as client:
    result = client.check("example.com")

    print(result.status)        # Status.HEALTHY, Status.WARNING, or Status.UNHEALTHY
    print(result.quorum)        # "3/3"
    print(result.locations)     # List of regional results

For quick health checks:

with NetDiagClient(api_key="your-api-key") as client:
    # Simple boolean check
    is_up = client.is_healthy("example.com")

    # Get status enum
    status = client.get_status("example.com")

API Reference

Constructor

NetDiagClient(
    *,
    api_key: str | None = None,
    base_url: str = "https://api.netdiag.dev",
    timeout: float = 30.0
)

Methods

check(target)

Run full diagnostics on a target.

def check(self, target: str | CheckRequest) -> CheckResponse: ...

Examples:

# Simple check with just hostname
result = client.check("example.com")

# Check with options
from netdiag import CheckRequest

result = client.check(CheckRequest(
    target="example.com",
    port=443,
    ping_count=10,
    regions="us-east,eu-west,ap-southeast"
))

check_prometheus(target)

Get diagnostics results in Prometheus metrics format.

def check_prometheus(self, target: str | CheckRequest) -> str: ...

Example:

metrics = client.check_prometheus("example.com")
# Returns Prometheus-formatted metrics string

is_healthy(target)

Quick boolean health check.

def is_healthy(self, target: str) -> bool: ...

Example:

if client.is_healthy("api.example.com"):
    print("Service is healthy!")

get_status(target)

Get the status enum for a target.

def get_status(self, target: str) -> Status: ...

Example:

from netdiag import Status

status = client.get_status("example.com")

match status:
    case Status.HEALTHY:
        print("All systems operational")
    case Status.WARNING:
        print("Degraded performance")
    case Status.UNHEALTHY:
        print("Service is down")

close()

Close the HTTP client and release resources.

def close(self) -> None: ...

Note: When using the context manager (with statement), close() is called automatically.

Request Options

When using CheckRequest:

Available Regions: us-east, us-west, eu-west, eu-central, ap-southeast, ap-northeast

Response Types

CheckResponse

@dataclass
class CheckResponse:
    run_id: str                          # Unique diagnostic run ID
    target: str                          # Target that was checked
    status: Status                       # Overall health status
    quorum: str                          # Healthy regions ratio (e.g., "3/4")
    dns_propagation_status: str          # "consistent" | "mismatched" | "unknown"
    started_at: str                      # ISO timestamp
    completed_at: str                    # ISO timestamp
    locations: list[LocationResult]      # Per-region results

LocationResult

@dataclass
class LocationResult:
    region: str
    status: Status
    ping: PingResult | None
    dns: DnsResult | None
    tls: TlsResult | None
    http: HttpResult | None

Diagnostic Results

@dataclass
class PingResult:
    status: Status
    latency_ms: float
    min_rtt_ms: float | None = None
    max_rtt_ms: float | None = None
    packet_loss_percent: float | None = None
    tcp_fallback_used: bool = False
    message: str | None = None
    error: ErrorInfo | None = None

@dataclass
class DnsResult:
    status: Status
    resolved_addresses: list[str]
    message: str | None = None
    error: ErrorInfo | None = None

@dataclass
class TlsResult:
    status: Status
    certificate_valid: bool
    days_until_expiry: int | None = None
    expires_at: str | None = None
    subject: str | None = None
    issuer: str | None = None
    protocol: str | None = None
    message: str | None = None
    error: ErrorInfo | None = None

@dataclass
class HttpResult:
    status: Status
    status_code: int | None = None
    reason_phrase: str | None = None
    response_time_ms: float | None = None
    message: str | None = None
    error: ErrorInfo | None = None

Enums

from enum import Enum

class Status(Enum):
    HEALTHY = "Healthy"
    WARNING = "Warning"
    UNHEALTHY = "Unhealthy"

class ErrorCode(Enum):
    NONE = "None"
    UNKNOWN_ERROR = "UnknownError"
    TIMEOUT = "Timeout"
    PING_FAILED = "PingFailed"
    PING_HIGH_LATENCY = "PingHighLatency"
    PING_PACKET_LOSS = "PingPacketLoss"
    DNS_FAILED = "DnsFailed"
    DNS_NXDOMAIN = "DnsNxDomain"
    TLS_FAILED = "TlsFailed"
    TLS_EXPIRED = "TlsExpired"
    TLS_EXPIRING_SOON = "TlsExpiringSoon"
    HTTP_FAILED = "HttpFailed"
    HTTP_CLIENT_ERROR = "HttpClientError"
    HTTP_SERVER_ERROR = "HttpServerError"

Error Handling

The client raises typed exceptions for different failure scenarios:

from netdiag import (
    NetDiagError,
    NetDiagApiError,
    NetDiagRateLimitError
)
import time

try:
    result = client.check("example.com")
except NetDiagRateLimitError as e:
    # Rate limited - wait and retry
    print(f"Retry after {e.retry_after_seconds} seconds")
    time.sleep(e.retry_after_seconds)
except NetDiagApiError as e:
    # API error (4xx/5xx)
    print(f"API Error: {e.status_code} - {e.body}")
except NetDiagError as e:
    # Other client errors
    print(f"Client Error: {e}")

Exception Types

Tenacity Retry Example

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type
)
from netdiag import NetDiagClient, NetDiagRateLimitError

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=60),
    retry=retry_if_exception_type(NetDiagRateLimitError)
)
def check_with_retry(client: NetDiagClient, target: str):
    return client.check(target)

with NetDiagClient(api_key="your-key") as client:
    result = check_with_retry(client, "example.com")

Advanced Usage

FastAPI Integration

from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends
from netdiag import NetDiagClient

client: NetDiagClient | None = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global client
    client = NetDiagClient(api_key="your-key")
    yield
    client.close()

app = FastAPI(lifespan=lifespan)

def get_client() -> NetDiagClient:
    assert client is not None
    return client

@app.get("/check/{target}")
def check_target(target: str, client: NetDiagClient = Depends(get_client)):
    return client.check(target)

@app.get("/metrics")
def prometheus_metrics(client: NetDiagClient = Depends(get_client)):
    return client.check_prometheus("your-service.com")

Django Integration

# services.py
from netdiag import NetDiagClient
from django.conf import settings

_client: NetDiagClient | None = None

def get_netdiag_client() -> NetDiagClient:
    global _client
    if _client is None:
        _client = NetDiagClient(api_key=settings.NETDIAG_API_KEY)
    return _client

# views.py
from django.http import JsonResponse
from .services import get_netdiag_client

def check_view(request, target):
    client = get_netdiag_client()
    result = client.check(target)
    return JsonResponse({
        "status": result.status.value,
        "quorum": result.quorum
    })

Prometheus Scraping

from flask import Flask, Response
from netdiag import NetDiagClient

app = Flask(__name__)
client = NetDiagClient(api_key="your-key")

@app.route("/metrics")
def metrics():
    prometheus_metrics = client.check_prometheus("your-service.com")
    return Response(prometheus_metrics, mimetype="text/plain")

if __name__ == "__main__":
    app.run(port=9090)

Type Hints

The package includes complete type annotations. All types are exported from the main module:

from netdiag import (
    # Client
    NetDiagClient,

    # Request/Response
    CheckRequest,
    CheckResponse,
    LocationResult,

    # Diagnostic results
    PingResult,
    DnsResult,
    TlsResult,
    HttpResult,

    # Enums
    Status,
    ErrorCode,

    # Errors
    ErrorInfo,
    NetDiagError,
    NetDiagApiError,
    NetDiagRateLimitError
)

Related