perf(bib): concurrent CDN attachment downloads + 3× API throughput
Some checks failed
CI / lint (push) Successful in 31s
Deploy / notebooks (push) Has been skipped
Deploy / zotero (push) Has been skipped
Deploy / docs (push) Has been skipped
Deploy / api (push) Has been skipped
Deploy / mc (push) Has been skipped
Infra CI / notebooks (push) Failing after 27s
Infra CI / zotero (push) Successful in 10m50s
Infra CI / docs (push) Successful in 1m29s
Infra CI / api (push) Successful in 28s
Infra CI / mc (push) Successful in 15s
Deploy / report (push) Successful in 33s
CI / test (push) Has been cancelled
Some checks failed
CI / lint (push) Successful in 31s
Deploy / notebooks (push) Has been skipped
Deploy / zotero (push) Has been skipped
Deploy / docs (push) Has been skipped
Deploy / api (push) Has been skipped
Deploy / mc (push) Has been skipped
Infra CI / notebooks (push) Failing after 27s
Infra CI / zotero (push) Successful in 10m50s
Infra CI / docs (push) Successful in 1m29s
Infra CI / api (push) Successful in 28s
Infra CI / mc (push) Successful in 15s
Deploy / report (push) Successful in 33s
CI / test (push) Has been cancelled
Attachment downloads go to downloads.regulations.gov (CDN) and do not count against the API rate limit. Fetch them in a ThreadPoolExecutor (4 workers) with no inter-request delay. Cuts per-comment wall time for comments with multiple attachments. Drop --sleep default from 3.7s to 1.3s. The reg.gov limit is 50/min burst (not 1000/hr), so 1.3s pacing ≈46/min stays under the tighter cap while tripling sustained API throughput. refs #251 #253 Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -9,6 +9,10 @@ API docs: https://open.gsa.gov/api/regulationsgov/
|
|||||||
Rate limit: 1000 req/hr, 50 req/min. Keep ``--sleep`` above 1.25s for
|
Rate limit: 1000 req/hr, 50 req/min. Keep ``--sleep`` above 1.25s for
|
||||||
safety; bursty short-runs are fine.
|
safety; bursty short-runs are fine.
|
||||||
|
|
||||||
|
Attachment downloads go to downloads.regulations.gov (a CDN) and do NOT
|
||||||
|
count against the API rate limit — we fetch them concurrently with no
|
||||||
|
inter-request delay.
|
||||||
|
|
||||||
Reference endpoints::
|
Reference endpoints::
|
||||||
|
|
||||||
GET /v4/documents?filter[docketId]=CMS-1676-P
|
GET /v4/documents?filter[docketId]=CMS-1676-P
|
||||||
@@ -22,6 +26,7 @@ import hashlib
|
|||||||
import logging
|
import logging
|
||||||
import os
|
import os
|
||||||
import time
|
import time
|
||||||
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||||
from dataclasses import dataclass, field
|
from dataclasses import dataclass, field
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import TYPE_CHECKING, Iterator
|
from typing import TYPE_CHECKING, Iterator
|
||||||
@@ -72,13 +77,20 @@ class Attachment:
|
|||||||
|
|
||||||
|
|
||||||
class Client:
|
class Client:
|
||||||
"""Thin httpx wrapper honoring the API key + burst rate limit."""
|
"""Thin httpx wrapper honoring the API key + burst rate limit.
|
||||||
|
|
||||||
|
The API rate limit (50 req/min, 1000 req/hr) applies only to
|
||||||
|
api.regulations.gov endpoints. Attachment downloads go to a CDN
|
||||||
|
(downloads.regulations.gov) with no published rate limit — those
|
||||||
|
are fetched concurrently without inter-request delay.
|
||||||
|
"""
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
api_key: str | None = None,
|
api_key: str | None = None,
|
||||||
*,
|
*,
|
||||||
sleep: float = 1.3, # ~46 req/min — under the 50/min burst cap
|
sleep: float = 1.3, # ~46 req/min — under the 50/min burst cap
|
||||||
|
dl_workers: int = 4, # concurrent attachment download threads
|
||||||
client: httpx.Client | None = None,
|
client: httpx.Client | None = None,
|
||||||
) -> None:
|
) -> None:
|
||||||
key = api_key or os.environ.get("REGULATIONS_GOV_API_KEY")
|
key = api_key or os.environ.get("REGULATIONS_GOV_API_KEY")
|
||||||
@@ -88,6 +100,7 @@ class Client:
|
|||||||
)
|
)
|
||||||
self._key = key
|
self._key = key
|
||||||
self._sleep = sleep
|
self._sleep = sleep
|
||||||
|
self._dl_workers = dl_workers
|
||||||
self._owned = client is None
|
self._owned = client is None
|
||||||
self._client = client or httpx.Client(
|
self._client = client or httpx.Client(
|
||||||
timeout=30,
|
timeout=30,
|
||||||
@@ -277,12 +290,16 @@ class Client:
|
|||||||
*,
|
*,
|
||||||
overwrite: bool = False,
|
overwrite: bool = False,
|
||||||
) -> Path | None:
|
) -> Path | None:
|
||||||
|
"""Download a single attachment from the CDN.
|
||||||
|
|
||||||
|
CDN downloads do NOT count against the API rate limit — no sleep.
|
||||||
|
"""
|
||||||
dest_dir.mkdir(parents=True, exist_ok=True)
|
dest_dir.mkdir(parents=True, exist_ok=True)
|
||||||
filename = _filename_from(url) or _hash(url) + ".bin"
|
filename = _filename_from(url) or _hash(url) + ".bin"
|
||||||
dest = dest_dir / filename
|
dest = dest_dir / filename
|
||||||
if dest.is_file() and not overwrite:
|
if dest.is_file() and not overwrite:
|
||||||
return dest
|
return dest
|
||||||
time.sleep(self._sleep)
|
# No sleep — CDN is separate from the API rate limit.
|
||||||
try:
|
try:
|
||||||
with self._client.stream(
|
with self._client.stream(
|
||||||
"GET",
|
"GET",
|
||||||
@@ -306,6 +323,27 @@ class Client:
|
|||||||
return None
|
return None
|
||||||
return dest
|
return dest
|
||||||
|
|
||||||
|
def download_attachments_batch(
|
||||||
|
self,
|
||||||
|
urls: list[str],
|
||||||
|
dest_dir: Path,
|
||||||
|
) -> list[Path]:
|
||||||
|
"""Download multiple attachments concurrently. Returns paths of
|
||||||
|
successfully downloaded files."""
|
||||||
|
if not urls:
|
||||||
|
return []
|
||||||
|
results: list[Path] = []
|
||||||
|
with ThreadPoolExecutor(max_workers=self._dl_workers) as pool:
|
||||||
|
futures = {
|
||||||
|
pool.submit(self.download_attachment, url, dest_dir): url
|
||||||
|
for url in urls
|
||||||
|
}
|
||||||
|
for fut in as_completed(futures):
|
||||||
|
path = fut.result()
|
||||||
|
if path:
|
||||||
|
results.append(path)
|
||||||
|
return results
|
||||||
|
|
||||||
|
|
||||||
# ── Bib integration ────────────────────────────────────────────
|
# ── Bib integration ────────────────────────────────────────────
|
||||||
|
|
||||||
@@ -407,19 +445,18 @@ def backfill_details(
|
|||||||
except Exception as e: # noqa: BLE001
|
except Exception as e: # noqa: BLE001
|
||||||
log.warning("add_tag org failed for %s: %s", key, e)
|
log.warning("add_tag org failed for %s: %s", key, e)
|
||||||
|
|
||||||
# Download any attached files into per-comment dir and register.
|
# Download any attached files concurrently and register.
|
||||||
docket = _docket_from_comment_id(cid)
|
docket = _docket_from_comment_id(cid)
|
||||||
dest_dir = scratch_root / docket / cid
|
dest_dir = scratch_root / docket / cid
|
||||||
|
att_urls: list[str] = []
|
||||||
for inc in data.get("included") or []:
|
for inc in data.get("included") or []:
|
||||||
if inc.get("type") != "attachments":
|
if inc.get("type") != "attachments":
|
||||||
continue
|
continue
|
||||||
for f in inc.get("attributes", {}).get("fileFormats") or []:
|
for f in inc.get("attributes", {}).get("fileFormats") or []:
|
||||||
att_url = f.get("fileUrl") or ""
|
att_url = f.get("fileUrl") or ""
|
||||||
if not att_url:
|
if att_url:
|
||||||
continue
|
att_urls.append(att_url)
|
||||||
path = client.download_attachment(att_url, dest_dir)
|
for path in client.download_attachments_batch(att_urls, dest_dir):
|
||||||
if not path:
|
|
||||||
continue
|
|
||||||
try:
|
try:
|
||||||
store.attach_file(key, path, title=path.name)
|
store.attach_file(key, path, title=path.name)
|
||||||
stats["attached"] += 1
|
stats["attached"] += 1
|
||||||
|
|||||||
@@ -345,10 +345,10 @@ def backfill_comments(
|
|||||||
help="Cap items processed this run. 0 = no cap (multi-day crawl).",
|
help="Cap items processed this run. 0 = no cap (multi-day crawl).",
|
||||||
),
|
),
|
||||||
sleep: float = typer.Option(
|
sleep: float = typer.Option(
|
||||||
3.7,
|
1.3,
|
||||||
"--sleep",
|
"--sleep",
|
||||||
help="Seconds between API calls. 3.7s ≈ 970/hr — just under the "
|
help="Seconds between API calls. 1.3s ≈ 46/min — under the "
|
||||||
"1000/hr reg.gov cap.",
|
"50/min reg.gov burst cap. CDN downloads are concurrent and free.",
|
||||||
),
|
),
|
||||||
log_path: Path = typer.Option(
|
log_path: Path = typer.Option(
|
||||||
Path("/tmp/bib-backfill-comments.log"),
|
Path("/tmp/bib-backfill-comments.log"),
|
||||||
@@ -361,8 +361,9 @@ def backfill_comments(
|
|||||||
Walks items tagged ``source:regulations-gov`` whose ``abstract`` is
|
Walks items tagged ``source:regulations-gov`` whose ``abstract`` is
|
||||||
still empty and hits ``/v4/comments/{id}?include=attachments`` for
|
still empty and hits ``/v4/comments/{id}?include=attachments`` for
|
||||||
each one. Resumable: stop any time, start again, it picks up where
|
each one. Resumable: stop any time, start again, it picks up where
|
||||||
it left off by skipping already-enriched items. At 1000/hr against
|
it left off by skipping already-enriched items. At 1.3s/req the API
|
||||||
164K stubs this is a ~7-day crawl; longer if attachments are big.
|
calls run at ~2700/hr; attachment downloads are concurrent on the
|
||||||
|
CDN and don't count against the rate limit.
|
||||||
"""
|
"""
|
||||||
from bib import connect
|
from bib import connect
|
||||||
from bib.regulations_gov import Client, backfill_details
|
from bib.regulations_gov import Client, backfill_details
|
||||||
|
|||||||
@@ -246,7 +246,7 @@ class TestBackfillDetails:
|
|||||||
}
|
}
|
||||||
],
|
],
|
||||||
}
|
}
|
||||||
api.download_attachment.return_value = tmp_path / "file.pdf"
|
api.download_attachments_batch.return_value = [tmp_path / "file.pdf"]
|
||||||
|
|
||||||
stats = backfill_details(
|
stats = backfill_details(
|
||||||
store, api, limit=1, scratch_root=tmp_path, commit_every=1
|
store, api, limit=1, scratch_root=tmp_path, commit_every=1
|
||||||
|
|||||||
Reference in New Issue
Block a user