{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# Enrich a feature class from an external service - a safe template\n", "\n", "Three phases, and the split is the whole point:\n", "\n", "1. **Read** a worklist with a `SearchCursor`, keyed by `OID@`, then close it.\n", "2. **Enrich** - call the external service with **no cursor open**, so no schema lock is held\n", " across the network.\n", "3. **Write back** with a fresh `UpdateCursor`, matching on `OID@`.\n", "\n", "The same `WHERE_CLAUSE` drives phases 1 and 3, which is what makes the job **restartable**:\n", "re-run it and it picks up only what is still outstanding. When the service is metered, that\n", "clause is also your cost control.\n", "\n", "---\n", "\n", "### Read this before you run it\n", "\n", "* **`DRY_RUN` is `True`.** Nothing is written until you change it. A dry run prints exactly\n", " what a live run would change, field by field.\n", "* **`WHERE_CLAUSE` has no default and the job refuses to start without one.** Express the rows\n", " that still *need* work. A clause of `1=1` is rejected on purpose.\n", "* **`MAX_ROWS` caps the run** so a mistake stays small. Hitting the cap is reported, and\n", " re-running continues where you left off.\n", "* **Test against a copy first.** Phase 3 rewrites rows.\n", "\n", "MIT licensed. Standard library plus `arcpy`.\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 1. CONFIG\n" ] }, { "cell_type": "code", "metadata": {}, "outputs": [], "execution_count": null, "source": [ "FEATURE_CLASS = None # e.g. r\"C:\\gis\\sites.gdb\\locations\"\n", "\n", "# Required. The rows that still need work - and the reason this job is restartable.\n", "WHERE_CLAUSE = None # e.g. \"CITY IS NULL OR CITY = ''\"\n", "\n", "# OID@ must be first: it is the join key between phase 1 and phase 3.\n", "READ_FIELDS = [\"OID@\", \"NAME\", \"LAT\", \"LON\"]\n", "WRITE_FIELDS = [\"OID@\", \"CITY\", \"ZIP\"]\n", "\n", "DRY_RUN = True # set False only after reading a dry-run report\n", "MAX_ROWS = 500 # cap per run\n", "ONLY_IF_EMPTY = True # never overwrite a value someone already set\n", "LOG_PATH = None # e.g. r\"C:\\gis\\enrich-log.json\" - diff two runs\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 2. The engine\n", "\n", "You should not need to edit this cell.\n" ] }, { "cell_type": "code", "metadata": {}, "outputs": [], "execution_count": null, "source": [ "import json\n", "from datetime import datetime, timezone\n", "\n", "import arcpy\n", "\n", "OID = \"OID@\"\n", "\n", "\n", "class ConfigError(Exception):\n", " \"\"\"Raised before any data is touched, so a bad configuration costs nothing.\"\"\"\n", "\n", "\n", "# ------------------------------------------------------------------ phase 0: refuse bad input\n", "def preflight(fc, where_clause, read_fields, write_fields, max_rows):\n", " \"\"\"Validate before touching data. Every check here exists because the failure it prevents is\n", " expensive, irreversible, or both.\"\"\"\n", " if not fc:\n", " raise ConfigError(\"FEATURE_CLASS is required.\")\n", " if not arcpy.Exists(fc):\n", " raise ConfigError(f\"FEATURE_CLASS does not exist: {fc}\")\n", "\n", " if not where_clause or not str(where_clause).strip():\n", " raise ConfigError(\n", " \"WHERE_CLAUSE is required and has no default. It is what makes this job restartable \"\n", " \"and what stops you paying an external service for rows that already have values. \"\n", " \"Express the rows that still NEED work, for example: \\\"CITY IS NULL OR CITY = ''\\\"\")\n", " if str(where_clause).strip().replace(\" \", \"\") in (\"1=1\", \"1<>0\"):\n", " raise ConfigError(\n", " \"WHERE_CLAUSE selects every row. If that is genuinely intended, say so explicitly in \"\n", " \"a clause that names the fields, so the intent survives the next person reading it.\")\n", "\n", " if OID not in read_fields:\n", " raise ConfigError(f\"{OID} must be the first read field - it is the join key between phases.\")\n", " if OID not in write_fields:\n", " raise ConfigError(f\"{OID} must be present in the write fields to match rows back.\")\n", "\n", " existing = {f.name for f in arcpy.ListFields(fc)}\n", " for label, fields in ((\"read\", read_fields), (\"write\", write_fields)):\n", " missing = [f for f in fields if f != OID and f not in existing]\n", " if missing:\n", " raise ConfigError(f\"{label} field(s) not on {fc}: {', '.join(missing)}\")\n", "\n", " if not isinstance(max_rows, int) or max_rows < 1:\n", " raise ConfigError(\"MAX_ROWS must be a positive integer. It caps the blast radius of a mistake.\")\n", " return True\n", "\n", "\n", "# ------------------------------------------------------------------ phase 1: read\n", "def read_worklist(fc, where_clause, read_fields, max_rows):\n", " \"\"\"Return a list of dicts, one per row needing work. The cursor is closed before this returns -\n", " nothing downstream may hold it.\"\"\"\n", " work = []\n", " with arcpy.da.SearchCursor(fc, read_fields, where_clause=where_clause) as cur:\n", " for row in cur:\n", " work.append(dict(zip(read_fields, row)))\n", " if len(work) >= max_rows:\n", " break\n", " return work\n", "\n", "\n", "# ------------------------------------------------------------------ phase 2: the slow part\n", "def enrich(work, lookup):\n", " \"\"\"Call `lookup` once per row and collect the results, keyed by OID.\n", "\n", " `lookup` is yours to supply: it receives one row dict and returns either a dict of\n", " {field_name: value} to write, or None to skip the row. It runs with NO cursor open, so it may\n", " take as long as it needs - a network call, a rate limit, a retry - without holding a lock on\n", " the table.\n", " \"\"\"\n", " decisions, failed = {}, []\n", " for row in work:\n", " try:\n", " result = lookup(row)\n", " except Exception as exc: # one bad row must not end the run\n", " failed.append({\"oid\": row[OID], \"error\": f\"{type(exc).__name__}: {exc}\"})\n", " continue\n", " if result:\n", " decisions[row[OID]] = result\n", " return decisions, failed\n", "\n", "\n", "# ------------------------------------------------------------------ phase 3: write\n", "def write_back(fc, where_clause, write_fields, decisions, dry_run=True, only_if_empty=True):\n", " \"\"\"Apply decisions by OID. A fresh cursor - the phase 1 cursor is long closed, and at ArcGIS\n", " Pro 3.7 reusing it would raise ValueError anyway.\"\"\"\n", " planned, changed, skipped = [], 0, 0\n", "\n", " with arcpy.da.UpdateCursor(fc, write_fields, where_clause=where_clause) as cur:\n", " for row in cur:\n", " oid = row[0]\n", " decision = decisions.get(oid)\n", " if not decision:\n", " continue\n", "\n", " row_changes = {}\n", " for i, field in enumerate(write_fields):\n", " if field == OID or field not in decision:\n", " continue\n", " current, proposed = row[i], decision[field]\n", " if proposed is None or proposed == current:\n", " continue\n", " # Re-check emptiness inside the write pass. WHERE_CLAUSE already restricted the set,\n", " # but another editor may have filled the row since phase 1 read it.\n", " if only_if_empty and current not in (None, \"\", \" \"):\n", " skipped += 1\n", " continue\n", " row_changes[field] = {\"from\": current, \"to\": proposed}\n", " row[i] = proposed\n", "\n", " if not row_changes:\n", " continue\n", " planned.append({\"oid\": oid, \"changes\": row_changes})\n", " if not dry_run:\n", " cur.updateRow(row)\n", " changed += 1\n", "\n", " return {\"planned\": planned, \"changed\": changed, \"skipped_not_empty\": skipped,\n", " \"dry_run\": bool(dry_run)}\n", "\n", "\n", "# ------------------------------------------------------------------ reporting\n", "def summarize(work, decisions, failed, result, max_rows):\n", " lines = []\n", " mode = \"DRY RUN - nothing was written\" if result[\"dry_run\"] else \"LIVE - rows were updated\"\n", " lines.append(f\"{mode}\")\n", " lines.append(f\" rows read (cap {max_rows:,}) {len(work):,}\")\n", " lines.append(f\" rows the lookup answered {len(decisions):,}\")\n", " lines.append(f\" rows the lookup failed on {len(failed):,}\")\n", " lines.append(f\" rows with at least one change {result['changed']:,}\")\n", " lines.append(f\" values skipped, already set {result['skipped_not_empty']:,}\")\n", " if result[\"planned\"]:\n", " lines.append(\"\")\n", " lines.append(\" first few changes:\")\n", " for item in result[\"planned\"][:5]:\n", " for field, ch in item[\"changes\"].items():\n", " lines.append(f\" OID {item['oid']:<8} {field:<22} {ch['from']!r} -> {ch['to']!r}\")\n", " if failed:\n", " lines.append(\"\")\n", " lines.append(\" first few failures:\")\n", " for f in failed[:5]:\n", " lines.append(f\" OID {f['oid']:<8} {f['error']}\")\n", " if work and not decisions and not failed:\n", " lines.append(\"\")\n", " lines.append(\" The lookup returned nothing for every row it was given. If you have not\")\n", " lines.append(\" replaced the lookup() stub yet, that is expected - the shipped stub always\")\n", " lines.append(\" returns None so the template runs end to end without contacting anything.\")\n", " if result[\"dry_run\"]:\n", " lines.append(\"\")\n", " # Only invite a live run when there is something to apply. Printing this after a run that\n", " # planned nothing sends you to DRY_RUN = False for a job that would do exactly nothing.\n", " lines.append(\" Set DRY_RUN = False to apply these changes.\" if result[\"planned\"]\n", " else \" Nothing to apply - this dry run planned no changes.\")\n", " if len(work) == max_rows:\n", " lines.append(\"\")\n", " lines.append(f\" NOTE: the run stopped at the MAX_ROWS cap of {max_rows:,}. There are likely\")\n", " lines.append(\" more rows matching WHERE_CLAUSE. Re-run to continue - the clause makes it resumable.\")\n", " return \"\\n\".join(lines)\n", "\n", "\n", "def write_log(path, work, decisions, failed, result):\n", " \"\"\"A JSON record of what happened, so a re-run can be diffed against the last one.\"\"\"\n", " payload = {\n", " \"utc\": datetime.now(timezone.utc).strftime(\"%Y-%m-%dT%H:%M:%SZ\"),\n", " \"dry_run\": result[\"dry_run\"], \"rows_read\": len(work),\n", " \"rows_answered\": len(decisions), \"rows_changed\": result[\"changed\"],\n", " \"skipped_not_empty\": result[\"skipped_not_empty\"],\n", " \"planned\": result[\"planned\"], \"failed\": failed,\n", " }\n", " with open(path, \"w\", encoding=\"utf-8\") as fh:\n", " json.dump(payload, fh, indent=2, default=str)\n", " return path\n", "\n", "\n", "def run(fc, where_clause, read_fields, write_fields, lookup, max_rows=500,\n", " dry_run=True, only_if_empty=True, log_path=None):\n", " \"\"\"The whole three-phase job. Returns (summary_text, result_dict).\"\"\"\n", " preflight(fc, where_clause, read_fields, write_fields, max_rows)\n", " work = read_worklist(fc, where_clause, read_fields, max_rows)\n", " if not work:\n", " return \"No rows matched WHERE_CLAUSE. Nothing to do.\", None\n", " decisions, failed = enrich(work, lookup)\n", " result = write_back(fc, where_clause, write_fields, decisions, dry_run, only_if_empty)\n", " if log_path:\n", " write_log(log_path, work, decisions, failed, result)\n", " return summarize(work, decisions, failed, result, max_rows), result\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 3. Your lookup\n", "\n", "The only function you write.\n" ] }, { "cell_type": "code", "metadata": {}, "outputs": [], "execution_count": null, "source": [ "# ---------------------------------------------------------------------------------\n", "# THE ONE FUNCTION YOU WRITE\n", "# ---------------------------------------------------------------------------------\n", "# Receives one row as a dict keyed by your READ_FIELDS, e.g.\n", "# {\"OID@\": 1842, \"NAME\": \"Riverbend\", \"LAT\": 42.41, \"LON\": -83.38}\n", "# Return a dict of {field: value} to write, or None to skip the row.\n", "#\n", "# This runs with NO cursor open, so it may be as slow as it needs to be - a network\n", "# call, a rate-limit pause, a retry. Raising here skips one row and the run continues.\n", "\n", "def lookup(row):\n", " lat, lon = row.get(\"LAT\"), row.get(\"LON\")\n", " if lat is None or lon is None:\n", " return None # nothing to work with; skip\n", "\n", " # Replace this block with a real call. Example shape:\n", " #\n", " # import urllib.request, urllib.parse, json\n", " # qs = urllib.parse.urlencode({\"lat\": lat, \"lon\": lon, \"token\": TOKEN})\n", " # with urllib.request.urlopen(f\"https://example.invalid/reverse?{qs}\", timeout=20) as r:\n", " # payload = json.load(r)\n", " # return {\"CITY\": payload.get(\"city\"), \"ZIP\": payload.get(\"postal\")}\n", " #\n", " # Until then the template runs end to end without contacting anything:\n", " return None\n", "\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 4. Run it\n", "\n", "Safe to run as-is: `DRY_RUN` is on and the stub lookup contacts nothing.\n" ] }, { "cell_type": "code", "metadata": {}, "outputs": [], "execution_count": null, "source": [ "summary, result = run(\n", " FEATURE_CLASS, WHERE_CLAUSE, READ_FIELDS, WRITE_FIELDS, lookup,\n", " max_rows=MAX_ROWS, dry_run=DRY_RUN, only_if_empty=ONLY_IF_EMPTY,\n", " log_path=LOG_PATH,\n", ")\n", "print(summary)\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 5. Reading the report, and going live\n", "\n", "A dry run tells you three numbers worth checking before you write anything:\n", "\n", "* **rows read** - if this equals `MAX_ROWS`, there is more work outstanding.\n", "* **rows the lookup failed on** - a handful is normal, a majority means the service or your\n", " credentials are wrong and you should not proceed.\n", "* **values skipped, already set** - rows something else filled between phase 1 and phase 3.\n", " Non-zero here means you are not the only writer, which is worth knowing.\n", "\n", "When the preview looks right, set `DRY_RUN = False` and re-run.\n", "\n", "### Limits worth knowing\n", "\n", "* `ONLY_IF_EMPTY` protects existing values. Turning it off means this template will overwrite\n", " data a person entered.\n", "* The row cap is per run, not per day. A metered service will bill every call the lookup makes.\n", "* Nothing here is transactional. If a live run is interrupted, the rows already written stay\n", " written - which is exactly why `WHERE_CLAUSE` matters: re-running resumes cleanly.\n" ] } ], "metadata": { "kernelspec": { "display_name": "ArcGIS Pro", "language": "python", "name": "python3" }, "language_info": { "name": "python", "version": "3.13" } }, "nbformat": 4, "nbformat_minor": 5 }