Import Data from an Esri Feature Service into Fulcrum

pip install requests

import requests
import csv
import json

# ── Configuration ─────────────────────────────────────────────────────────────
# The base URL of the feature service layer. Format:
#   https://<server>/arcgis/rest/services/<folder>/<service>/FeatureServer/<layer_id>
FEATURE_SERVICE_URL = 'https://services.arcgis.com/YOUR-ORG-ID/arcgis/rest/services/YOUR_SERVICE/FeatureServer/0'
OUTPUT_CSV          = 'esri_features.csv'

# Optional: include geometry as latitude/longitude columns
INCLUDE_GEOMETRY    = True


def fetch_all_features(service_url):
    """
    Fetches all features from an ArcGIS feature service layer using
    offset-based pagination (required for large layers).
    """
    all_features = []
    offset       = 0
    page_size    = 1000   # ArcGIS default max per request

    while True:
        params = {
            'where':         '1=1',           # Return all features
            'outFields':     '*',             # Return all attribute fields
            'returnGeometry': INCLUDE_GEOMETRY,
            'outSR':         '4326',          # WGS 84 lat/lon
            'f':             'json',
            'resultOffset':  offset,
            'resultRecordCount': page_size
        }

        response = requests.get(f"{service_url}/query", params=params, timeout=30)
        response.raise_for_status()
        data = response.json()

        features = data.get('features', [])
        all_features.extend(features)
        print(f"  Fetched {len(all_features)} features so far...")

        # ArcGIS signals "no more pages" via exceededTransferLimit or empty page
        if not data.get('exceededTransferLimit', False) or len(features) < page_size:
            break

        offset += page_size

    return all_features


def features_to_csv(features, output_path):
    """Writes feature attributes (and optional lat/lon) to a CSV file."""
    if not features:
        print("No features to write.")
        return

    # Collect all attribute field names across features
    fieldnames = set()
    for f in features:
        fieldnames.update(f.get('attributes', {}).keys())
    fieldnames = sorted(fieldnames)

    if INCLUDE_GEOMETRY:
        fieldnames = ['latitude', 'longitude'] + fieldnames

    with open(output_path, 'w', newline='', encoding='utf-8') as csvfile:
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames, extrasaction='ignore')
        writer.writeheader()

        for feature in features:
            row = dict(feature.get('attributes', {}))

            if INCLUDE_GEOMETRY and feature.get('geometry'):
                geom = feature['geometry']
                # Point geometry returns x (longitude) and y (latitude)
                row['longitude'] = geom.get('x')
                row['latitude']  = geom.get('y')

            writer.writerow(row)

    print(f"✅ Wrote {len(features)} rows to {output_path}")


# ── Run ───────────────────────────────────────────────────────────────────
print(f"Querying {FEATURE_SERVICE_URL}...")
features = fetch_all_features(FEATURE_SERVICE_URL)
print(f"Total features: {len(features)}")
features_to_csv(features, OUTPUT_CSV)
print(f"\nCSV saved to: {OUTPUT_CSV}")
print("Next step: Import this CSV into Fulcrum via Settings → Imports.")