Peeringdb

PeeringDB API client for fetching prefix information.

PeeringDBPrefixInfo

Bases: BaseModel

Prefix information from PeeringDB for a given ASN.

Attributes:

Name	Type	Description
asn	int	The Autonomous System Number
prefixes4	int | None	Number of IPv4 prefixes announced, or None if not available
prefixes6	int | None	Number of IPv6 prefixes announced, or None if not available
name	str | None	Network name from PeeringDB, or None if not available

Source code in gso/services/peeringdb.py

class PeeringDBPrefixInfo(BaseModel):
    """Prefix information from PeeringDB for a given ASN.

    Attributes:
        asn: The Autonomous System Number
        prefixes4: Number of IPv4 prefixes announced, or None if not available
        prefixes6: Number of IPv6 prefixes announced, or None if not available
        name: Network name from PeeringDB, or None if not available
    """

    asn: int
    prefixes4: int | None
    prefixes6: int | None
    name: str | None

_parse_429_wait_time(response, fallback_wait)

Parse wait time from a 429 rate limit response.

Parameters:

Name	Type	Description	Default
response	Response	The 429 response from PeeringDB	required
fallback_wait	int	Fallback wait time in seconds if not parseable	required

Returns:

Type	Description
int	Wait time in seconds.

Source code in gso/services/peeringdb.py

def _parse_429_wait_time(response: requests.Response, fallback_wait: int) -> int:
    """Parse wait time from a 429 rate limit response.

    Args:
        response: The 429 response from PeeringDB
        fallback_wait: Fallback wait time in seconds if not parseable

    Returns:
        Wait time in seconds.
    """
    # Try Retry-After header first
    retry_after = response.headers.get("Retry-After")
    if retry_after:
        try:
            return int(retry_after)
        except ValueError:
            pass

    # Try to parse from JSON message: "Expected available in X seconds"
    try:
        data = response.json()
        if isinstance(data, dict) and "message" in data:
            message = data["message"]
            if isinstance(message, str):
                match = re.search(r"Expected available in (\d+) seconds", message)
                if match:
                    return int(match.group(1))
    except Exception:  # noqa: S110, BLE001
        pass

    # Fallback to configured wait time
    return fallback_wait

create_session()

Create a requests session with retry logic and timeout.

Returns:

Type	Description
Session	A configured requests.Session with retry and timeout defaults.

Source code in gso/services/peeringdb.py

def create_session() -> requests.Session:
    """Create a requests session with retry logic and timeout.

    Returns:
        A configured requests.Session with retry and timeout defaults.
    """
    session = requests.Session()

    # Get settings
    params = load_oss_params()
    peeringdb_config = params.PEERINGDB

    # Set API key if configured
    if peeringdb_config.api_key:
        session.headers["Authorization"] = f"Api-Key {peeringdb_config.api_key}"

    # Configure retry for server errors (not 429, we handle that separately)
    retry_strategy = Retry(
        total=peeringdb_config.default_retries,
        backoff_factor=0.5,
        status_forcelist=[500, 502, 503, 504],
        allowed_methods=["GET"],
    )
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("https://", adapter)
    session.mount("http://", adapter)

    # Track last request time for simple pacing
    session._last_request_time = 0.0  # type: ignore[attr-defined]  # noqa: SLF001
    session._peeringdb_config = peeringdb_config  # type: ignore[attr-defined]  # noqa: SLF001

    return session

fetch_prefix_info(asn, session=None)

Fetch prefix information for a given ASN from PeeringDB.

Handles HTTP 429 rate limiting with automatic retries.

Parameters:

Name	Type	Description	Default
asn	int	The Autonomous System Number to query	required
session	Session | None	Optional requests session. If None, a new session will be created.	None

Returns:

Type	Description
PeeringDBPrefixInfo | None	PeeringDBPrefixInfo if the ASN is found, None otherwise.

Raises:

Type	Description
RequestException	If the API request fails after retries.

Source code in gso/services/peeringdb.py

def fetch_prefix_info(asn: int, session: requests.Session | None = None) -> PeeringDBPrefixInfo | None:
    """Fetch prefix information for a given ASN from PeeringDB.

    Handles HTTP 429 rate limiting with automatic retries.

    Args:
        asn: The Autonomous System Number to query
        session: Optional requests session. If None, a new session will be created.

    Returns:
        PeeringDBPrefixInfo if the ASN is found, None otherwise.

    Raises:
        requests.RequestException: If the API request fails after retries.
    """
    if session is None:
        session = create_session()

    # Get PeeringDB config from session
    peeringdb_config = getattr(session, "_peeringdb_config", None)
    if peeringdb_config is None:
        # Fallback if session was created without config
        params_obj = load_oss_params()
        peeringdb_config = params_obj.PEERINGDB

    url = f"{peeringdb_config.api_base}/net"
    params: dict[str, str | int] = {
        "asn": asn,
        "fields": "asn,info_prefixes4,info_prefixes6,name",
    }

    # Check if API key is configured for logging
    api_key_configured = "Authorization" in session.headers

    # Simple retry loop for 429 errors
    for attempt in range(1, peeringdb_config.max_429_retries + 1):
        try:
            # Simple pacing: wait if needed before making request
            last_request_time = getattr(session, "_last_request_time", 0.0)
            elapsed = time.time() - last_request_time
            if elapsed < peeringdb_config.request_delay:
                time.sleep(peeringdb_config.request_delay - elapsed)

            # Make request
            session._last_request_time = time.time()  # type: ignore[attr-defined]  # noqa: SLF001
            response = session.get(url, params=params, timeout=peeringdb_config.default_timeout)

            # Handle 429 specifically
            if response.status_code == HTTPStatus.TOO_MANY_REQUESTS:
                wait_seconds = _parse_429_wait_time(response, peeringdb_config.fallback_429_wait)
                logger.warning(
                    "PeeringDB rate limit hit, will retry",
                    asn=asn,
                    attempt=attempt,
                    max_attempts=peeringdb_config.max_429_retries,
                    wait_seconds=wait_seconds,
                    api_key_configured=api_key_configured,
                )

                # If we've exhausted retries, raise
                if attempt >= peeringdb_config.max_429_retries:
                    logger.error(
                        "PeeringDB rate limit retries exhausted",
                        asn=asn,
                        attempts=attempt,
                        api_key_configured=api_key_configured,
                    )
                    msg = f"PeeringDB rate limit exceeded after {attempt} attempts for ASN {asn}"
                    raise ApiException(HTTPStatus(response.status_code), msg, response)

                # Wait and retry
                time.sleep(wait_seconds)
                continue

            # For non-429 responses, proceed normally
            response.raise_for_status()

            try:
                data = response.json()
            except ValueError as e:
                logger.exception("Invalid JSON response from PeeringDB", asn=asn, error=str(e))
                msg = f"Invalid JSON response: {e}"
                raise requests.RequestException(msg) from e

            # Validate response structure
            if not isinstance(data, dict):
                logger.error("Unexpected response format from PeeringDB", asn=asn, response_type=type(data).__name__)
                msg = "Unexpected response format"
                raise requests.RequestException(msg)  # noqa: TRY301

            if not data.get("data"):
                logger.info("No PeeringDB data found for ASN", asn=asn)
                return None

            # Take the first result (should be the only one for a unique ASN)
            net_list = data["data"]
            if not isinstance(net_list, list) or len(net_list) == 0:
                logger.info("No PeeringDB network data for ASN", asn=asn)
                return None

            net_info = net_list[0]

            return PeeringDBPrefixInfo(
                asn=net_info.get("asn"),
                prefixes4=net_info.get("info_prefixes4"),
                prefixes6=net_info.get("info_prefixes6"),
                name=net_info.get("name"),
            )

        except requests.RequestException as e:
            # Don't retry non-429 errors
            logger.exception("Failed to fetch PeeringDB data", asn=asn, error=str(e))
            raise

    # Should not reach here, but for safety
    msg = f"Unexpected exit from retry loop for ASN {asn}"
    raise requests.RequestException(msg)