services/ingest/backfill_is_current.py

109 lines · python

"""
One-time backfill for members.is_current.
 
Authoritatively marks the sitting 119th Congress: every bioguide present in the
unitedstates/congress-legislators `legislators-current.json` dataset (the same
source the members poller ingests from) is is_current = 1; everyone else is 0.
 
Run against the deployed DB:
 
    DOCKET_DB_PATH=/opt/docket/data/docket.db \
        python3 -m services.ingest.backfill_is_current
 
or, from inside services/ingest/:
 
    DOCKET_DB_PATH=/opt/docket/data/docket.db python3 backfill_is_current.py
 
Idempotent: safe to re-run. Adds the column if it is missing, then performs a
single transaction (two bulk UPDATEs) so it stays light against the live ingest
(the connection carries a 60s busy_timeout).
"""
 
import logging
import os
import sys
 
import requests
 
try:
    from ingest.db import get_connection
    from ingest.pollers.members import LEGISLATORS_JSON_URL, _parse_legislators
except ModuleNotFoundError:
    sys.path.insert(0, os.path.join(os.path.dirname(__file__)))
    from ingest.db import get_connection
    from ingest.pollers.members import LEGISLATORS_JSON_URL, _parse_legislators
 
logger = logging.getLogger("backfill_is_current")
 
 
def fetch_current_bioguides() -> list[str]:
    resp = requests.get(LEGISLATORS_JSON_URL, timeout=60)
    resp.raise_for_status()
    data = resp.json()
    if not isinstance(data, list):
        raise ValueError(f"unexpected JSON shape: {type(data)}")
    members = _parse_legislators(data)
    return [m["bioguide_id"] for m in members if m.get("bioguide_id")]
 
 
def main() -> int:
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    )
 
    bioguides = fetch_current_bioguides()
    logger.info("fetched %d current bioguides from legislators-current.json", len(bioguides))
    if not bioguides:
        logger.error("no current bioguides fetched; aborting (DB left unchanged)")
        return 1
 
    conn = get_connection()
 
    try:
        conn.execute(
            "ALTER TABLE members ADD COLUMN is_current INTEGER NOT NULL DEFAULT 0"
        )
    except Exception:
        pass
 
    total = conn.execute("SELECT COUNT(*) AS c FROM members").fetchone()["c"]
 
    placeholders = ",".join("?" for _ in bioguides)
    conn.execute("BEGIN")
    conn.execute("UPDATE members SET is_current = 0")
    cur = conn.execute(
        f"UPDATE members SET is_current = 1 WHERE bioguide_id IN ({placeholders})",
        bioguides,
    )
    marked = cur.rowcount
    conn.commit()
 
    matched = conn.execute(
        "SELECT COUNT(*) AS c FROM members WHERE is_current = 1"
    ).fetchone()["c"]
 
    missing = len(set(bioguides)) - matched
 
    logger.info(
        "members total=%d | current-roster=%d | marked is_current=1 -> %d (rowcount=%d) | not-in-db=%d",
        total,
        len(bioguides),
        matched,
        marked,
        missing,
    )
 
    for st in ("WY", "CA", "TX"):
        row = conn.execute(
            "SELECT COUNT(*) AS c FROM members WHERE is_current = 1 AND UPPER(state) = ?",
            (st,),
        ).fetchone()
        logger.info("  current members in %s: %d", st, row["c"])
 
    conn.close()
    return 0
 
 
if __name__ == "__main__":
    raise SystemExit(main())

1	"""
2	One-time backfill for members.is_current.
3
4	Authoritatively marks the sitting 119th Congress: every bioguide present in the
5	unitedstates/congress-legislators `legislators-current.json` dataset (the same
6	source the members poller ingests from) is is_current = 1; everyone else is 0.
7
8	Run against the deployed DB:
9
10	DOCKET_DB_PATH=/opt/docket/data/docket.db \
11	python3 -m services.ingest.backfill_is_current
12
13	or, from inside services/ingest/:
14
15	DOCKET_DB_PATH=/opt/docket/data/docket.db python3 backfill_is_current.py
16
17	Idempotent: safe to re-run. Adds the column if it is missing, then performs a
18	single transaction (two bulk UPDATEs) so it stays light against the live ingest
19	(the connection carries a 60s busy_timeout).
20	"""
21
22	import logging
23	import os
24	import sys
25
26	import requests
27
28	try:
29	from ingest.db import get_connection
30	from ingest.pollers.members import LEGISLATORS_JSON_URL, _parse_legislators
31	except ModuleNotFoundError:
32	sys.path.insert(0, os.path.join(os.path.dirname(__file__)))
33	from ingest.db import get_connection
34	from ingest.pollers.members import LEGISLATORS_JSON_URL, _parse_legislators
35
36	logger = logging.getLogger("backfill_is_current")
37
38
39	def fetch_current_bioguides() -> list[str]:
40	resp = requests.get(LEGISLATORS_JSON_URL, timeout=60)
41	resp.raise_for_status()
42	data = resp.json()
43	if not isinstance(data, list):
44	raise ValueError(f"unexpected JSON shape: {type(data)}")
45	members = _parse_legislators(data)
46	return [m["bioguide_id"] for m in members if m.get("bioguide_id")]
47
48
49	def main() -> int:
50	logging.basicConfig(
51	level=logging.INFO,
52	format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
53	)
54
55	bioguides = fetch_current_bioguides()
56	logger.info("fetched %d current bioguides from legislators-current.json", len(bioguides))
57	if not bioguides:
58	logger.error("no current bioguides fetched; aborting (DB left unchanged)")
59	return 1
60
61	conn = get_connection()
62
63	try:
64	conn.execute(
65	"ALTER TABLE members ADD COLUMN is_current INTEGER NOT NULL DEFAULT 0"
66	)
67	except Exception:
68	pass
69
70	total = conn.execute("SELECT COUNT(*) AS c FROM members").fetchone()["c"]
71
72	placeholders = ",".join("?" for _ in bioguides)
73	conn.execute("BEGIN")
74	conn.execute("UPDATE members SET is_current = 0")
75	cur = conn.execute(
76	f"UPDATE members SET is_current = 1 WHERE bioguide_id IN ({placeholders})",
77	bioguides,
78	)
79	marked = cur.rowcount
80	conn.commit()
81
82	matched = conn.execute(
83	"SELECT COUNT(*) AS c FROM members WHERE is_current = 1"
84	).fetchone()["c"]
85
86	missing = len(set(bioguides)) - matched
87
88	logger.info(
89	"members total=%d \| current-roster=%d \| marked is_current=1 -> %d (rowcount=%d) \| not-in-db=%d",
90	total,
91	len(bioguides),
92	matched,
93	marked,
94	missing,
95	)
96
97	for st in ("WY", "CA", "TX"):
98	row = conn.execute(
99	"SELECT COUNT(*) AS c FROM members WHERE is_current = 1 AND UPPER(state) = ?",
100	(st,),
101	).fetchone()
102	logger.info(" current members in %s: %d", st, row["c"])
103
104	conn.close()
105	return 0
106
107
108	if __name__ == "__main__":
109	raise SystemExit(main())