← back to La Permits School
pull_permits.py
169 lines
#!/usr/bin/env python3
"""
pull_permits.py — free, $0, no-key LADBS building-permit puller (school project).
Pulls City of Los Angeles "Building Permit Information Data" from the LA Open Data
Portal (Socrata API), filters to significant permits (by type + valuation), and
writes a clean CSV — including contractor / applicant / principal names and a
ready-made California Secretary of State lookup URL per contractor for enrichment.
Data source (public, free, no signup):
https://data.lacity.org/resource/d9aa-v8bm.json (2013-01 .. 2023-05, ~317k rows)
Standard library only — no pip installs, no API key required. A free Socrata
"app token" only raises rate limits; not needed for this volume.
Examples:
python3 pull_permits.py # new + additions over $1M -> permits.csv
python3 pull_permits.py --min-valuation 5000000 # only >$5M projects
python3 pull_permits.py --permit-types "Bldg-New" --since 2020-01-01
python3 pull_permits.py --limit 500 --out sample.csv
"""
import argparse
import csv
import sys
import time
import urllib.parse
import urllib.request
DATASET_DEFAULT = "d9aa-v8bm" # richest PUBLIC dataset (names + valuation)
BASE = "https://data.lacity.org/resource/{dataset}.json"
PAGE = 50000 # Socrata max rows per request
CA_SOS_SEARCH = "https://bizfileonline.sos.ca.gov/search/business"
# Columns kept in the output CSV, in order. (dataset field -> stays same name)
OUT_FIELDS = [
"pcis_permit", "issue_date", "permit_type", "permit_sub_type", "permit_category",
"valuation", "work_description",
"address", "zip_code", "zone", "council_district",
"of_stories", "of_residential_dwelling_units", "apn",
"contractors_business_name", "license", "license_type",
"contractor_city", "contractor_state",
"applicant_first_name", "applicant_last_name", "applicant_relationship",
"principal_first_name", "principal_middle_name", "principal_last_name",
"latest_status", "status_date",
"ca_sos_lookup_url",
]
def build_where(min_valuation, permit_types, since):
clauses = []
if min_valuation is not None:
clauses.append(f"valuation > {int(min_valuation)}")
if permit_types:
quoted = ",".join("'" + t.replace("'", "''") + "'" for t in permit_types)
clauses.append(f"permit_type in({quoted})")
if since:
clauses.append(f"issue_date >= '{since}T00:00:00'")
return " and ".join(clauses)
def fetch_page(dataset, where, offset, app_token=None):
params = {
"$where": where,
"$order": "valuation DESC",
"$limit": PAGE,
"$offset": offset,
}
if app_token:
params["$$app_token"] = app_token
url = BASE.format(dataset=dataset) + "?" + urllib.parse.urlencode(params)
req = urllib.request.Request(url, headers={"Accept": "application/json"})
import json
for attempt in range(4):
try:
with urllib.request.urlopen(req, timeout=60) as r:
return json.load(r)
except Exception as e: # transient network / rate limit -> back off
if attempt == 3:
print(f" ! giving up on offset {offset}: {e}", file=sys.stderr)
return []
time.sleep(2 * (attempt + 1))
return []
def address_of(row):
parts = [
row.get("address_start", ""),
row.get("street_direction", ""),
row.get("street_name", ""),
row.get("street_suffix", ""),
]
return " ".join(p for p in parts if p).strip()
def sos_url(business_name):
if not business_name:
return ""
return CA_SOS_SEARCH + "?" + urllib.parse.urlencode({"searchType": "BUSINESS", "q": business_name})
def apn_of(row):
# Full LA County APN = book (4) + page (3) + parcel (3), e.g. 5144-021-024
b, p, parcel = row.get("assessor_book", ""), row.get("assessor_page", ""), row.get("assessor_parcel", "")
if b and p and parcel:
return f"{b}-{p}-{parcel}"
return "-".join(x for x in (b, p, parcel) if x)
def shape(row):
derived = ("address", "ca_sos_lookup_url", "apn")
out = {k: row.get(k, "") for k in OUT_FIELDS if k not in derived}
out["address"] = address_of(row)
out["apn"] = apn_of(row)
out["ca_sos_lookup_url"] = sos_url(row.get("contractors_business_name", ""))
return out
def main():
ap = argparse.ArgumentParser(description="Free LADBS permit puller (school project).")
ap.add_argument("--dataset", default=DATASET_DEFAULT, help="Socrata dataset id (default d9aa-v8bm)")
ap.add_argument("--min-valuation", type=int, default=1_000_000, help="min project valuation $ (default 1000000; 0 = no filter)")
ap.add_argument("--permit-types", default="Bldg-New,Bldg-Addition",
help="comma list; '' = all. Options: Bldg-New, Bldg-Addition, Bldg-Alter/Repair, Bldg-Demolition, Sign, ...")
ap.add_argument("--since", default="", help="only permits issued on/after YYYY-MM-DD")
ap.add_argument("--limit", type=int, default=0, help="max rows to write (0 = all matches)")
ap.add_argument("--out", default="permits.csv", help="output CSV path")
ap.add_argument("--app-token", default="", help="optional free Socrata app token (raises rate limit)")
args = ap.parse_args()
permit_types = [t.strip() for t in args.permit_types.split(",") if t.strip()]
min_val = args.min_valuation if args.min_valuation > 0 else None
where = build_where(min_val, permit_types, args.since) or "valuation IS NOT NULL"
print("LADBS permit puller — cost: $0 (public LA Open Data API, no key)")
print(f" dataset : {args.dataset}")
print(f" filter : {where}")
print(f" output : {args.out}")
written = 0
offset = 0
with open(args.out, "w", newline="", encoding="utf-8") as f:
w = csv.DictWriter(f, fieldnames=OUT_FIELDS)
w.writeheader()
while True:
rows = fetch_page(args.dataset, where, offset, args.app_token or None)
if not rows:
break
for row in rows:
w.writerow(shape(row))
written += 1
if args.limit and written >= args.limit:
break
print(f" ... {written} rows", flush=True)
if args.limit and written >= args.limit:
break
if len(rows) < PAGE:
break
offset += PAGE
time.sleep(0.3) # be polite to the free API
print(f"\nDone. Wrote {written} permits to {args.out}")
print("Next step (free): open the CSV, and for any contractor, click its")
print("ca_sos_lookup_url to pull the legal entity + agent from CA SOS bizfile.")
if __name__ == "__main__":
main()