{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# Describe Feature Services\n", "\n", "Inventory ArcGIS Online hosted **feature services** into a data dictionary. For each service this records item metadata, every layer/table, geometry type, record count, spatial reference, and per-field **name / alias / type / length / nullability / domain** (coded values *or* range). Writes the log as TXT, CSV (flat, one row per field), or **multi-sheet XLSX** (`Services` / `Layers` / `Fields`).\n", "\n", "**How to use:** edit the **CONFIG** cell, then *Run All*. Sign-in uses your active ArcGIS session via `GIS(\"home\")` (ArcGIS Notebooks) or `GIS(\"pro\")` (ArcGIS Pro) -- no credentials live in this notebook.\n", "\n", "**Make it yours:** the describe functions are written **PropertyMap-OR-dict tolerant**, so the writers serialize cleanly and can be unit-tested with plain dicts (no portal needed). Adapted from an All Things Spatial workflow.\n", "\n", "Runtime: arcgispro-py3 / ArcGIS Notebooks. Needs the ArcGIS API for Python + `openpyxl` for XLSX output (both ship with ArcGIS Pro)." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 1 - Config" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# ============================================================================\n", "# CONFIG -- edit these, then Run All. Nothing else needs changing for a basic run.\n", "# ============================================================================\n", "SCOPE = \"entire\" # \"entire\" | \"folders\" | \"single\" | \"search\"\n", "FOLDERS = [] # when SCOPE=\"folders\"; \"(root)\" = the root folder\n", "ID_OR_URL = \"\" # when SCOPE=\"single\": an item ID or a service URL\n", "SEARCH_TEXT = \"\" # when SCOPE=\"search\": title text (your items only)\n", "OUTPUT_DIR = \"agol_inventory\" # local folder for the log file\n", "OUTPUT_NAME = \"\" # blank = auto (agol_inventory_)\n", "OUTPUT_FORMAT = \"XLSX\" # \"XLSX\" | \"CSV\" | \"TXT\"\n", "COUNT_RECORDS = True # one count query per layer; slow on huge services\n", "MAX_ITEMS = 500 # safety cap on services processed" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 2 - Describe core (PropertyMap-OR-dict tolerant)" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import csv\n", "import datetime\n", "import json\n", "import os\n", "from pathlib import Path\n", "\n", "\n", "def _noop(*_a, **_k):\n", " pass\n", "\n", "\n", "def _g(obj, key, default=None):\n", " \"\"\"Get `key` from a dict OR an attribute-style object (arcgis PropertyMap).\"\"\"\n", " if obj is None:\n", " return default\n", " if isinstance(obj, dict):\n", " return obj.get(key, default)\n", " return getattr(obj, key, default)\n", "\n", "\n", "def describe_field(f):\n", " \"\"\"One field -> a flat dict, with domain flattened to a readable string.\"\"\"\n", " domain = _g(f, \"domain\")\n", " dtype = dname = dvalues = \"\"\n", " if domain:\n", " dtype = _g(domain, \"type\", \"\") or \"\"\n", " dname = _g(domain, \"name\", \"\") or \"\"\n", " if \"codedValue\" in dtype or _g(domain, \"codedValues\"):\n", " pairs = _g(domain, \"codedValues\", []) or []\n", " dtype = \"codedValue\"\n", " dvalues = \"; \".join(\"{}={}\".format(_g(cv, \"code\"), _g(cv, \"name\")) for cv in pairs)\n", " elif \"range\" in dtype or _g(domain, \"range\"):\n", " rng = _g(domain, \"range\", []) or []\n", " dtype = \"range\"\n", " dvalues = \"min={}, max={}\".format(*(rng + [\"\", \"\"])[:2])\n", " return {\n", " \"name\": _g(f, \"name\", \"\"),\n", " \"alias\": _g(f, \"alias\", \"\"),\n", " \"type\": str(_g(f, \"type\", \"\")).replace(\"esriFieldType\", \"\"),\n", " \"length\": _g(f, \"length\", \"\"),\n", " \"nullable\": _g(f, \"nullable\", \"\"),\n", " \"domainType\": dtype,\n", " \"domainName\": dname,\n", " \"domainValues\": dvalues,\n", " }\n", "\n", "\n", "def describe_layer(layer, kind=\"layer\", count_records=True, say=_noop):\n", " \"\"\"One FeatureLayer/Table -> dict. `layer` needs `.properties` and (for counts)\n", " a `.query(return_count_only=True)` method.\"\"\"\n", " p = _g(layer, \"properties\", layer) # tolerate being handed props directly\n", " extent = _g(p, \"extent\")\n", " wkid = _g(_g(extent, \"spatialReference\"), \"latestWkid\") or \\\n", " _g(_g(extent, \"spatialReference\"), \"wkid\")\n", " count = \"\"\n", " if count_records:\n", " try:\n", " count = layer.query(where=\"1=1\", return_count_only=True)\n", " except Exception as e: # noqa: BLE001\n", " say(\" (count failed: {})\".format(e))\n", " count = \"n/a\"\n", " fields = _g(p, \"fields\", []) or []\n", " return {\n", " \"name\": _g(p, \"name\", \"\"),\n", " \"id\": _g(p, \"id\", \"\"),\n", " \"kind\": kind,\n", " \"geometryType\": str(_g(p, \"geometryType\", \"\") or \"\").replace(\"esriGeometryType\", \"\").replace(\"esriGeometry\", \"\"),\n", " \"recordCount\": count,\n", " \"wkid\": wkid or \"\",\n", " \"fieldCount\": len(fields),\n", " \"fields\": [describe_field(f) for f in fields],\n", " }\n", "\n", "\n", "def describe_item(item, count_records=True, say=_noop):\n", " \"\"\"An arcgis Item (hosted feature service) -> dict incl. each layer + table.\"\"\"\n", " say(\"Describing: {} ({})\".format(_g(item, \"title\", \"?\"), _g(item, \"id\", \"?\")))\n", " layers = []\n", " for lyr in (getattr(item, \"layers\", None) or []):\n", " layers.append(describe_layer(lyr, \"layer\", count_records, say))\n", " for tbl in (getattr(item, \"tables\", None) or []):\n", " layers.append(describe_layer(tbl, \"table\", count_records, say))\n", " return {\n", " \"title\": _g(item, \"title\", \"\"),\n", " \"id\": _g(item, \"id\", \"\"),\n", " \"owner\": _g(item, \"owner\", \"\"),\n", " \"type\": _g(item, \"type\", \"\"),\n", " \"url\": _g(item, \"url\", \"\"),\n", " \"created\": _epoch(_g(item, \"created\")),\n", " \"modified\": _epoch(_g(item, \"modified\")),\n", " \"access\": _g(item, \"access\", \"\"),\n", " \"tags\": \", \".join(_g(item, \"tags\", []) or []),\n", " \"snippet\": _g(item, \"snippet\", \"\") or \"\",\n", " \"layers\": layers,\n", " }\n", "\n", "\n", "def _epoch(ms):\n", " \"\"\"AGOL stores ms-epoch; return an ISO date (or '').\"\"\"\n", " try:\n", " return datetime.datetime.utcfromtimestamp(int(ms) / 1000).strftime(\"%Y-%m-%d\")\n", " except Exception: # noqa: BLE001\n", " return \"\"" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 3 - Gather services per scope (takes an already-connected `gis`)" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "_FS_TYPES = (\"Feature Service\", \"Feature Layer\")\n", "\n", "\n", "def _is_feature_service(item):\n", " return _g(item, \"type\", \"\") in _FS_TYPES\n", "\n", "\n", "def gather_services(gis, scope, folders=None, item_id_or_url=None,\n", " search_query=None, max_items=500):\n", " \"\"\"Return a list of feature-service Items per scope:\n", " 'entire' | 'folders' | 'single' | 'search'. A non-empty item_id_or_url always\n", " wins (direct selection of one service, owned or not).\"\"\"\n", " # Direct id / URL override -- highest priority.\n", " if item_id_or_url:\n", " token = item_id_or_url.strip()\n", " item = None\n", " if token.lower().startswith(\"http\"):\n", " item = _item_from_url(gis, token)\n", " else:\n", " item = gis.content.get(token)\n", " return [item] if item else []\n", "\n", " me = gis.users.me\n", " if scope == \"search\":\n", " q = \"{} owner:{}\".format(search_query or \"\", me.username).strip()\n", " items = gis.content.search(q, item_type=\"Feature *\", max_items=max_items)\n", " return [it for it in items if _is_feature_service(it)]\n", "\n", " if scope == \"single\":\n", " return [] # handled via item_id_or_url upstream\n", "\n", " # entire / folders -> walk owner items by folder\n", " if scope == \"folders\":\n", " targets = [None if f == \"(root)\" else f for f in (folders or [])]\n", " else:\n", " targets = [None] + [f.get(\"title\") if isinstance(f, dict) else getattr(f, \"name\", None)\n", " for f in (me.folders or [])]\n", " out, seen = [], set()\n", " for folder in targets:\n", " try:\n", " items = me.items(folder=folder, max_items=max_items)\n", " except Exception:\n", " items = []\n", " for it in items:\n", " if it.id not in seen and _is_feature_service(it):\n", " seen.add(it.id)\n", " out.append(it)\n", " if len(out) >= max_items:\n", " return out\n", " return out\n", "\n", "\n", "def _item_from_url(gis, url):\n", " \"\"\"Best-effort: resolve a service URL to its portal Item.\"\"\"\n", " try:\n", " from arcgis.features import FeatureLayerCollection\n", " flc = FeatureLayerCollection(url, gis)\n", " if getattr(flc, \"properties\", None) is not None:\n", " sid = _g(flc.properties, \"serviceItemId\")\n", " if sid:\n", " return gis.content.get(sid)\n", " except Exception:\n", " pass\n", " return None" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 4 - Writers (TXT / CSV / multi-sheet XLSX)" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "def _field_rows(report):\n", " \"\"\"Flatten report -> one row per field (for CSV / the Fields sheet).\"\"\"\n", " rows = []\n", " for svc in report[\"services\"]:\n", " for lyr in svc[\"layers\"]:\n", " for f in lyr[\"fields\"]:\n", " rows.append({\n", " \"service_title\": svc[\"title\"], \"service_id\": svc[\"id\"],\n", " \"layer_name\": lyr[\"name\"], \"layer_kind\": lyr[\"kind\"],\n", " \"geometryType\": lyr[\"geometryType\"], \"recordCount\": lyr[\"recordCount\"],\n", " \"field_name\": f[\"name\"], \"alias\": f[\"alias\"], \"type\": f[\"type\"],\n", " \"length\": f[\"length\"], \"nullable\": f[\"nullable\"],\n", " \"domainType\": f[\"domainType\"], \"domainName\": f[\"domainName\"],\n", " \"domainValues\": f[\"domainValues\"],\n", " })\n", " return rows\n", "\n", "\n", "_FIELD_COLS = [\"service_title\", \"service_id\", \"layer_name\", \"layer_kind\",\n", " \"geometryType\", \"recordCount\", \"field_name\", \"alias\", \"type\",\n", " \"length\", \"nullable\", \"domainType\", \"domainName\", \"domainValues\"]\n", "\n", "\n", "def write_csv(report, path):\n", " rows = _field_rows(report)\n", " with open(path, \"w\", newline=\"\", encoding=\"utf-8-sig\") as fh:\n", " w = csv.DictWriter(fh, fieldnames=_FIELD_COLS)\n", " w.writeheader()\n", " w.writerows(rows)\n", " return path\n", "\n", "\n", "def write_txt(report, path):\n", " L = []\n", " L.append(\"ArcGIS Online Feature Service Inventory\")\n", " L.append(\"Generated: {} Portal: {}\".format(report[\"generated\"], report.get(\"portal\", \"\")))\n", " L.append(\"Services: {}\".format(len(report[\"services\"])))\n", " for svc in report[\"services\"]:\n", " L.append(\"\\n\" + \"=\" * 72)\n", " L.append(\"{} [{}]\".format(svc[\"title\"], svc[\"type\"]))\n", " L.append(\" id={} owner={} access={}\".format(svc[\"id\"], svc[\"owner\"], svc[\"access\"]))\n", " L.append(\" created={} modified={}\".format(svc[\"created\"], svc[\"modified\"]))\n", " if svc[\"url\"]:\n", " L.append(\" url={}\".format(svc[\"url\"]))\n", " if svc[\"tags\"]:\n", " L.append(\" tags={}\".format(svc[\"tags\"]))\n", " for lyr in svc[\"layers\"]:\n", " L.append(\"\\n -- {} '{}' (id {}) --\".format(lyr[\"kind\"], lyr[\"name\"], lyr[\"id\"]))\n", " L.append(\" geometry={} records={} wkid={} fields={}\".format(\n", " lyr[\"geometryType\"] or \"n/a\", lyr[\"recordCount\"], lyr[\"wkid\"], lyr[\"fieldCount\"]))\n", " for f in lyr[\"fields\"]:\n", " line = \" - {} ({}) {}\".format(\n", " f[\"name\"], f[\"type\"], \"[{}]\".format(f[\"alias\"]) if f[\"alias\"] else \"\")\n", " meta = []\n", " if f[\"length\"]:\n", " meta.append(\"len={}\".format(f[\"length\"]))\n", " meta.append(\"nullable={}\".format(f[\"nullable\"]))\n", " if f[\"domainType\"]:\n", " meta.append(\"domain={}:{}\".format(f[\"domainType\"], f[\"domainName\"]))\n", " line += \" \" + \", \".join(meta)\n", " L.append(line)\n", " if f[\"domainValues\"]:\n", " L.append(\" values: {}\".format(f[\"domainValues\"]))\n", " Path(path).write_text(\"\\n\".join(L) + \"\\n\", encoding=\"utf-8\")\n", " return path\n", "\n", "\n", "def write_xlsx(report, path):\n", " from openpyxl import Workbook\n", " from openpyxl.styles import Font\n", "\n", " wb = Workbook()\n", " bold = Font(bold=True)\n", "\n", " ws = wb.active\n", " ws.title = \"Services\"\n", " svc_cols = [\"title\", \"id\", \"owner\", \"type\", \"access\", \"created\", \"modified\",\n", " \"layerCount\", \"url\", \"tags\"]\n", " ws.append(svc_cols)\n", " for c in ws[1]:\n", " c.font = bold\n", " for svc in report[\"services\"]:\n", " ws.append([svc[\"title\"], svc[\"id\"], svc[\"owner\"], svc[\"type\"], svc[\"access\"],\n", " svc[\"created\"], svc[\"modified\"], len(svc[\"layers\"]), svc[\"url\"], svc[\"tags\"]])\n", "\n", " wl = wb.create_sheet(\"Layers\")\n", " lyr_cols = [\"service_title\", \"service_id\", \"layer_name\", \"kind\",\n", " \"geometryType\", \"recordCount\", \"wkid\", \"fieldCount\"]\n", " wl.append(lyr_cols)\n", " for c in wl[1]:\n", " c.font = bold\n", " for svc in report[\"services\"]:\n", " for lyr in svc[\"layers\"]:\n", " wl.append([svc[\"title\"], svc[\"id\"], lyr[\"name\"], lyr[\"kind\"],\n", " lyr[\"geometryType\"], lyr[\"recordCount\"], lyr[\"wkid\"], lyr[\"fieldCount\"]])\n", "\n", " wf = wb.create_sheet(\"Fields\")\n", " wf.append(_FIELD_COLS)\n", " for c in wf[1]:\n", " c.font = bold\n", " for row in _field_rows(report):\n", " wf.append([row[k] for k in _FIELD_COLS])\n", "\n", " for sheet in wb.worksheets:\n", " for col in sheet.columns:\n", " width = min(60, max(10, max((len(str(c.value)) for c in col if c.value is not None), default=10) + 2))\n", " sheet.column_dimensions[col[0].column_letter].width = width\n", "\n", " wb.save(path)\n", " return path\n", "\n", "\n", "def write_report(report, out_path, fmt):\n", " \"\"\"fmt in {'TXT','CSV','XLSX'}; returns the written path.\"\"\"\n", " fmt = (fmt or \"TXT\").upper()\n", " if fmt == \"CSV\":\n", " return write_csv(report, out_path)\n", " if fmt == \"XLSX\":\n", " return write_xlsx(report, out_path)\n", " return write_txt(report, out_path)\n", "\n", "\n", "def build_report(gis, services, count_records=True, say=_noop):\n", " \"\"\"Describe a list of items -> the report dict the writers consume.\"\"\"\n", " return {\n", " \"generated\": datetime.datetime.utcnow().strftime(\"%Y-%m-%d %H:%M:%SZ\"),\n", " \"portal\": _g(getattr(gis, \"properties\", None), \"portalName\", \"\"),\n", " \"services\": [describe_item(it, count_records, say) for it in services if it],\n", " }" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 5 - Run" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# ---- connect (no credentials in the notebook) ------------------------------\n", "from arcgis.gis import GIS\n", "try:\n", " gis = GIS(\"home\") # ArcGIS Notebooks / hosted\n", "except Exception:\n", " gis = GIS(\"pro\") # running inside ArcGIS Pro\n", "print(\"Signed in as\", gis.users.me.username, \"(\", _g(getattr(gis, \"properties\", None), \"portalName\", \"\"), \")\")\n", "\n", "# ---- gather the target services for the chosen scope -----------------------\n", "services = gather_services(\n", " gis, SCOPE, folders=FOLDERS or None, item_id_or_url=ID_OR_URL or None,\n", " search_query=SEARCH_TEXT or None, max_items=MAX_ITEMS)\n", "print(\"{} feature service(s) to describe.\".format(len(services)))\n", "\n", "# ---- describe + write the log ----------------------------------------------\n", "if not services:\n", " raise SystemExit(\"No feature services matched the chosen scope.\")\n", "\n", "report = build_report(gis, services, count_records=COUNT_RECORDS, say=print)\n", "\n", "Path(OUTPUT_DIR).mkdir(parents=True, exist_ok=True)\n", "ext = {\"XLSX\": \".xlsx\", \"CSV\": \".csv\", \"TXT\": \".txt\"}[OUTPUT_FORMAT.upper()]\n", "stamp = datetime.datetime.utcnow().strftime(\"%Y%m%dT%H%M%SZ\")\n", "base = OUTPUT_NAME or \"agol_inventory_{}\".format(stamp)\n", "out_path = os.path.join(OUTPUT_DIR, base + ext)\n", "write_report(report, out_path, OUTPUT_FORMAT)\n", "print(\"\\nLog written:\", out_path)" ] } ], "metadata": { "kernelspec": { "display_name": "ArcGISPro", "language": "Python", "name": "python3" }, "language_info": { "name": "python", "version": "3.11" } }, "nbformat": 4, "nbformat_minor": 5 }