fix: align API clients with the current v3 API and add CI tests

.github/workflows/test.yml

@@ -0,0 +1,25 @@
+name: Tests
+
+on:
+  push:
+    branches: [ main ]
+  pull_request:
+
+jobs:
+  unit-tests:
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: [ "3.9", "3.12" ]
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+      - name: Install package
+        run: |
+          python -m pip install --upgrade pip
+          python -m pip install .
+      - name: Run unit tests
+        run: python -m unittest discover -s tests -p "test_*_unit.py" -v

bgpkit/__init__.py

@@ -1,4 +1,5 @@
 from .bgpkit_parser import Filter, Parser, RouteElem, RouteParser
+from ._utils import BGPKITApiError
 from .bgpkit_broker import Broker, BrokerItem, CollectorItem, PeerItem
 from .bgpkit_roas import Roas, RoasItem
 from .bgpkit_ip import IpLookup, IpInfo

bgpkit/_utils.py

@@ -0,0 +1,39 @@
+"""Shared helpers for the BGPKIT API clients."""
+
+from dataclasses import fields
+from typing import Optional
+
+import requests
+
+DEFAULT_TIMEOUT = 30.0
+
+
+class BGPKITApiError(RuntimeError):
+    """Raised when an API request fails or returns an error response."""
+
+
+def from_item(cls, item: dict):
+    """Build a dataclass from an API item, ignoring unknown fields."""
+    names = {field.name for field in fields(cls)}
+    return cls(**{key: value for key, value in item.items() if key in names})
+
+
+def request_json(
+    url: str,
+    params: Optional[dict] = None,
+    verify: bool = True,
+    timeout: float = DEFAULT_TIMEOUT,
+) -> dict:
+    """GET a JSON API endpoint, raising BGPKITApiError on failures."""
+    try:
+        response = requests.get(url, params=params or {}, verify=verify, timeout=timeout)
+    except requests.RequestException as error:
+        raise BGPKITApiError(f"GET {url} failed: {error}") from error
+    if response.status_code >= 400:
+        raise BGPKITApiError(
+            f"GET {url} failed with HTTP {response.status_code}: {response.text[:200]}"
+        )
+    try:
+        return response.json()
+    except ValueError as error:
+        raise BGPKITApiError(f"GET {url} returned a non-JSON response") from error

bgpkit/bgpkit_asn.py

@@ -1,24 +1,31 @@
 from dataclasses import dataclass
 from typing import List, Optional
 
-import requests
+from ._utils import DEFAULT_TIMEOUT, from_item, request_json
 
 
 @dataclass
 class Pagination:
-    next: Optional[str] = None
-    previous: Optional[str] = None
+    page: int = 0
+    page_size: int = 0
+    total_pages: int = 0
+    total_count: int = 0
+    has_next: bool = False
+    has_prev: bool = False
 
 
 @dataclass
 class AsnInfo:
     asn: int
     name: str
     country: str = ""
+    country_name: str = ""
     org_name: str = ""
     org_id: str = ""
-    description: str = ""
-    website: str = ""
+    as2org: Optional[dict] = None
+    hegemony: Optional[dict] = None
+    peeringdb: Optional[dict] = None
+    population: Optional[dict] = None
 
 
 @dataclass
@@ -34,37 +41,76 @@ class AsnLookupResult:
 class AsnLookup:
     """BGPKIT ASN information lookup (v3/utils/asn)."""
 
-    def __init__(self, api_url: str = "https://api.bgpkit.com/v3/utils"):
+    def __init__(
+        self,
+        api_url: str = "https://api.bgpkit.com/v3/utils",
+        timeout: float = DEFAULT_TIMEOUT,
+    ):
         self.base_url = api_url.rstrip("/")
+        self.timeout = timeout
 
     def query(
         self,
         asn: str = None,
         country: str = None,
         search: str = None,
-        page: int = 1,
+        page: int = 0,
         page_size: int = 100,
     ) -> AsnLookupResult:
+        """Look up ASN information.
+
+        Args:
+            asn: AS number, or a comma-separated list of AS numbers.
+            country: Filter by ISO country code.
+            search: Free-text search over ASN names and organizations.
+            page: Page number to start from (0-indexed).
+            page_size: Results per page.
+
+        Returns:
+            AsnLookupResult with the parsed ASN entries and pagination info.
+        """
         params = {}
         if asn:
             params["asn"] = asn
         if country:
             params["country"] = country
         if search:
             params["search"] = search
-        if page:
+        if page is not None:
             params["page"] = page
         if page_size:
             params["page_size"] = min(page_size, 10000)
 
-        res = requests.get(f"{self.base_url}/asn", params=params).json()
-        data = [AsnInfo(**item) for item in res.get("data", [])]
-        pag = res.get("pagination", {})
+        response = request_json(f"{self.base_url}/asn", params=params, timeout=self.timeout)
+
+        data = []
+        for item in response.get("data", []):
+            as2org = item.get("as2org") if isinstance(item.get("as2org"), dict) else {}
+            data.append(
+                AsnInfo(
+                    asn=item.get("asn"),
+                    name=item.get("name", ""),
+                    country=item.get("country", ""),
+                    country_name=item.get("country_name", ""),
+                    org_name=as2org.get("org_name", ""),
+                    org_id=as2org.get("org_id", ""),
+                    as2org=as2org or None,
+                    hegemony=item.get("hegemony"),
+                    peeringdb=item.get("peeringdb"),
+                    population=item.get("population"),
+                )
+            )
+
+        pagination_data = response.get("pagination")
+        pagination = (
+            from_item(Pagination, pagination_data) if isinstance(pagination_data, dict) else None
+        )
+
         return AsnLookupResult(
             data=data,
-            count=res.get("count", 0),
-            page=res.get("page", page),
-            page_size=res.get("page_size", page_size),
-            updated_at=res.get("updatedAt", ""),
-            pagination=Pagination(**pag) if pag else None,
+            count=response.get("count", 0),
+            page=response.get("page", page),
+            page_size=response.get("page_size", page_size),
+            updated_at=response.get("updatedAt", ""),
+            pagination=pagination,
         )

bgpkit/bgpkit_broker.py

@@ -1,9 +1,14 @@
-from dataclasses import dataclass, field
+import json
+import os
+from dataclasses import dataclass
 from typing import List, Optional
 
-import requests as requests
 import urllib3
 
+from ._utils import DEFAULT_TIMEOUT, BGPKITApiError, from_item, request_json
+
+DEFAULT_API_URL = "https://api.bgpkit.com/v3/broker"
+
 
 def check_type(value: any, ty: type) -> bool:
     try:
@@ -22,80 +27,107 @@ class BrokerItem:
     url: str
     rough_size: int
     exact_size: int
+    delay: Optional[float] = None  # seconds since the latest catalog update; /latest items only
 
 
 @dataclass
 class PeerItem:
+    date: str
+    collector: str
     ip: str
     asn: int
-    collector: str
-    full_feed: bool = False
+    num_v4_pfxs: int
+    num_v6_pfxs: int
+    num_connected_asns: int
+
+    @property
+    def full_feed(self) -> bool:
+        """True for full-table peers (>700k IPv4 or >100k IPv6 prefixes)."""
+        return self.num_v4_pfxs > 700_000 or self.num_v6_pfxs > 100_000
 
 
 @dataclass
 class CollectorItem:
-    id: str
     name: str
     project: str
+    data_url: str
+    activated_on: str
+    deactivated_on: Optional[str]
     country: str
-    active: bool = True
-    latitude: Optional[float] = None
-    longitude: Optional[float] = None
-
 
-@dataclass
-class LatestResult:
-    """Result of the /v3/broker/latest endpoint."""
-    items: List[BrokerItem] = field(default_factory=list)
+    @property
+    def active(self) -> bool:
+        return self.deactivated_on is None
 
 
 class Broker:
     """BGPKIT Broker v3 API wrapper.
 
     Provides access to MRT data file search, peer information,
     collector metadata, and latest file discovery.
+
+    The API base URL can be overridden per instance with ``api_url`` or for the
+    whole process with the ``BGPKIT_BROKER_URL`` environment variable.
     """
 
     def __init__(
         self,
-        api_url: str = "https://api.bgpkit.com/v3/broker",
+        api_url: Optional[str] = None,
         page_size: int = 100,
         verify: bool = True,
+        timeout: float = DEFAULT_TIMEOUT,
     ):
+        if api_url is None:
+            api_url = os.environ.get("BGPKIT_BROKER_URL") or DEFAULT_API_URL
         self.base_url = api_url.rstrip("/")
         self.page_size = int(page_size)
         self.verify = verify
+        self.timeout = timeout
         if not verify:
             urllib3.disable_warnings()
 
+    def _request(self, endpoint: str, params: Optional[dict] = None) -> dict:
+        """Make a single request and return the parsed JSON body."""
+        return request_json(
+            f"{self.base_url}/{endpoint}",
+            params=params,
+            verify=self.verify,
+            timeout=self.timeout,
+        )
+
     def _paginate(self, endpoint: str, params: dict) -> dict:
-        """Fetch all pages of a paginated broker endpoint."""
-        page = 1
-        all_data = []
+        """Fetch every page of a paginated broker endpoint.
 
+        Stops when the reported total is reached, or when a page adds no new
+        items. The latter also terminates endpoints that ignore pagination
+        parameters instead of looping on the same full page forever.
+        """
+        page = 1
+        all_data: List[dict] = []
+        seen = set()
+        result = {}
         while True:
-            params["page"] = page
-            params["page_size"] = self.page_size
-            res = requests.get(
-                f"{self.base_url}/{endpoint}",
-                params=params,
-                verify=self.verify,
-            ).json()
-
-            if isinstance(res, dict):
-                data = res.get("data", [])
-                all_data.extend(data)
-                if len(data) < self.page_size:
-                    break
-            elif isinstance(res, list):
-                all_data.extend(res)
+            page_params = dict(params)
+            page_params["page"] = page
+            page_params["page_size"] = self.page_size
+
+            result = self._request(endpoint, page_params)
+            data = result.get("data", [])
+            new_items = []
+            for item in data:
+                key = json.dumps(item, sort_keys=True, default=str)
+                if key not in seen:
+                    seen.add(key)
+                    new_items.append(item)
+            all_data.extend(new_items)
+
+            total = result.get("total")
+            if not data or not new_items:
                 break
-            else:
+            if total is not None and len(all_data) >= int(total):
                 break
-
             page += 1
 
-        result = res if isinstance(res, dict) else {"data": all_data}
         result["data"] = all_data
         return result
 
@@ -123,18 +155,12 @@ def query(
             params["data_type"] = data_type
 
         result = self._paginate("search", params)
-        return [BrokerItem(**item) for item in result.get("data", [])]
+        return [from_item(BrokerItem, item) for item in result.get("data", [])]
 
     def latest(self) -> List[BrokerItem]:
-        """Get latest MRT data files across projects."""
-        res = requests.get(
-            f"{self.base_url}/latest",
-            verify=self.verify,
-        ).json()
-        if isinstance(res, list):
-            return [BrokerItem(**item) for item in res]
-        data = res.get("data", [])
-        return [BrokerItem(**item) for item in data]
+        """Get the latest MRT data file for every collector and data type."""
+        result = self._request("latest")
+        return [from_item(BrokerItem, item) for item in result.get("data", [])]
 
     def peers(
         self,
@@ -154,8 +180,8 @@ def peers(
         if collector:
             params["collector"] = collector
 
-        result = self._paginate("peers", params)
-        return [PeerItem(**item) for item in result.get("data", [])]
+        result = self._request("peers", params)
+        return [from_item(PeerItem, item) for item in result.get("data", [])]
 
     def collectors(
         self,
@@ -172,5 +198,5 @@ def collectors(
         if active is not None:
             params["active"] = str(active).lower()
 
-        result = self._paginate("collectors", params)
-        return [CollectorItem(**item) for item in result.get("data", [])]
+        result = self._request("collectors", params)
+        return [from_item(CollectorItem, item) for item in result.get("data", [])]