feat(prisma): Congress.gov bill parser
This commit is contained in:
53
src/prisma/congress.py
Normal file
53
src/prisma/congress.py
Normal file
@@ -0,0 +1,53 @@
|
||||
"""Congress.gov bill list. The HTTP call is a transport the caller supplies."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable, Mapping
|
||||
from dataclasses import dataclass
|
||||
|
||||
_BILL_URL = "https://api.congress.gov/v3/bill/{congress}?api_key={key}&format=json"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Bill:
|
||||
congress: int
|
||||
bill_type: str
|
||||
number: int
|
||||
title: str
|
||||
latest_action: str
|
||||
|
||||
|
||||
def parse_bills(payload: Mapping[str, object]) -> tuple[Bill, ...]:
|
||||
"""Read the ``bills`` array. Missing action text becomes an empty string."""
|
||||
raw = payload.get("bills") or ()
|
||||
bills: list[Bill] = []
|
||||
if not isinstance(raw, list):
|
||||
return ()
|
||||
for item in raw:
|
||||
if not isinstance(item, Mapping):
|
||||
continue
|
||||
action = item.get("latestAction") or {}
|
||||
text = action.get("text") if isinstance(action, Mapping) else ""
|
||||
bills.append(
|
||||
Bill(
|
||||
congress=int(item["congress"]),
|
||||
bill_type=str(item.get("type") or ""),
|
||||
number=int(item["number"]),
|
||||
title=str(item.get("title") or ""),
|
||||
latest_action=str(text or ""),
|
||||
)
|
||||
)
|
||||
return tuple(bills)
|
||||
|
||||
|
||||
def fetch_bills(
|
||||
congress: int,
|
||||
*,
|
||||
key: str,
|
||||
transport: Callable[[str], Mapping[str, object]],
|
||||
) -> tuple[Bill, ...]:
|
||||
"""GET the bill list for one congress. An empty key raises and does not fetch."""
|
||||
if not key:
|
||||
raise ValueError("Congress.gov API key is required")
|
||||
payload = transport(_BILL_URL.format(congress=congress, key=key))
|
||||
return parse_bills(payload)
|
||||
51
tests/prisma/test_congress.py
Normal file
51
tests/prisma/test_congress.py
Normal file
@@ -0,0 +1,51 @@
|
||||
"""Congress.gov bill payloads. The transport is injected. No HTTP."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from prisma.congress import Bill, fetch_bills, parse_bills
|
||||
|
||||
_PAYLOAD = {
|
||||
"bills": [
|
||||
{
|
||||
"congress": 119,
|
||||
"type": "hr",
|
||||
"number": 1,
|
||||
"title": "A bill",
|
||||
"latestAction": {"text": "Introduced"},
|
||||
}
|
||||
]
|
||||
}
|
||||
_BILL = Bill(
|
||||
congress=119,
|
||||
bill_type="hr",
|
||||
number=1,
|
||||
title="A bill",
|
||||
latest_action="Introduced",
|
||||
)
|
||||
|
||||
|
||||
def test_parse_bills_reads_one_bill():
|
||||
assert parse_bills(_PAYLOAD) == (_BILL,)
|
||||
|
||||
|
||||
def test_fetch_bills_passes_the_congress_url_to_the_transport():
|
||||
seen: list[str] = []
|
||||
|
||||
def transport(url: str) -> dict[str, object]:
|
||||
seen.append(url)
|
||||
return _PAYLOAD
|
||||
|
||||
assert fetch_bills(119, key="test-key", transport=transport) == (_BILL,)
|
||||
assert len(seen) == 1
|
||||
assert "119" in seen[0]
|
||||
assert "api_key=test-key" in seen[0]
|
||||
|
||||
|
||||
def test_empty_key_does_not_call_the_transport():
|
||||
def transport(url: str) -> dict[str, object]:
|
||||
raise AssertionError(url)
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
fetch_bills(119, key="", transport=transport)
|
||||
Reference in New Issue
Block a user