Python SDK Advanced Patterns, Async, GeoPandas, and Production Workflows

Use the townshipamerica Python SDK with async/await, GeoPandas DataFrames, and production-ready error handling for GIS pipelines and land tech applications.

Production patterns for the townshipamerica Python SDK, async clients for high-throughput pipelines, GeoPandas integration for spatial analysis, and patterns for O&G, agriculture, and GIS workflows.

Async client

The AsyncTownshipAmerica client uses httpx.AsyncClient under the hood, ideal for web servers, data pipelines, and concurrent workflows.

import asyncio
import os
from townshipamerica import AsyncTownshipAmerica

async def main():
    async with AsyncTownshipAmerica(
        api_key=os.environ["TOWNSHIPAMERICA_API_KEY"]
    ) as client:
        result = await client.search("NENE 12 4N 5E Indian Meridian")
        print(result.features[0].properties)

asyncio.run(main())

Every method on the sync client has an async equivalent, search, reverse, autocomplete, batch_search, batch_reverse.

Concurrent requests

Process multiple locations concurrently with asyncio.gather:

async def convert_many(descriptions: list[str]):
    async with AsyncTownshipAmerica(
        api_key=os.environ["TOWNSHIPAMERICA_API_KEY"]
    ) as client:
        tasks = [client.search(desc) for desc in descriptions]
        results = await asyncio.gather(*tasks)
        return results

For large lists, use batch_search instead, it handles up to 100 locations per request server-side, which is faster than 100 concurrent individual calls.

GeoPandas integration

Convert PLSS search results to a GeoPandas GeoDataFrame for spatial analysis:

import geopandas as gpd
from shapely.geometry import shape
from townshipamerica import TownshipAmerica

client = TownshipAmerica(api_key=os.environ["TOWNSHIPAMERICA_API_KEY"])

locations = [
    "NENE 12 4N 5E Indian Meridian",
    "SWNE 3 5N 11W Indian Meridian",
    "SWNE 22 3N 7E Montana Meridian",
]

results = client.batch_search(locations)

features = []
for fc in results:
    f = fc.features[0]
    features.append({
        "descriptor": f.properties.descriptor,
        "geometry": shape(f.geometry.model_dump()),
    })

gdf = gpd.GeoDataFrame(features, crs="EPSG:4326")
print(gdf)

Export to file

# GeoJSON
gdf.to_file("well_sites.geojson", driver="GeoJSON")

# Shapefile
gdf.to_file("well_sites.shp")

# GeoPackage
gdf.to_file("well_sites.gpkg", driver="GPKG")

O&G well site mapping

A common O&G workflow: take a list of PLSS locations from an APD spreadsheet, convert to coordinates, and plot on a map.

import csv
import os
from townshipamerica import TownshipAmerica

client = TownshipAmerica(api_key=os.environ["TOWNSHIPAMERICA_API_KEY"])

# Read PLSS descriptions from a CSV
with open("apd_locations.csv") as f:
    reader = csv.DictReader(f)
    descriptions = [row["legal_description"] for row in reader]

# Batch convert (max 100 per request)
all_results = []
for i in range(0, len(descriptions), 100):
    batch = descriptions[i:i + 100]
    all_results.extend(client.batch_search(batch))

# Extract centroids
for desc, fc in zip(descriptions, all_results):
    feature = fc.features[0]
    coords = feature.geometry.coordinates[0][0]
    print(f"{desc}{coords[1]:.6f}, {coords[0]:.6f}")

Agriculture, FSA quarter section verification

Verify quarter section descriptions from crop insurance filings:

# Verify a list of quarter sections from an FSA filing
filings = [
    "SWNE 22 9N 15W Indian Meridian",
    "NWSE 22 9N 15W Indian Meridian",
    "SENE 22 9N 15W Indian Meridian",
]

results = client.batch_search(filings)
for filing, fc in zip(filings, results):
    feature = fc.features[0]
    print(f"{filing}")
    print(f"  Verified: {feature.properties.descriptor}")
    print(f"  Coordinates: {feature.geometry.coordinates[0][0][:2]}")
    print()

FastAPI integration

Expose PLSS conversion as an endpoint in a FastAPI application:

import os
from fastapi import FastAPI, HTTPException
from townshipamerica import AsyncTownshipAmerica
from townshipamerica.exceptions import NotFoundError, ValidationError

app = FastAPI()
client = AsyncTownshipAmerica(api_key=os.environ["TOWNSHIPAMERICA_API_KEY"])

@app.on_event("shutdown")
async def shutdown():
    await client.close()

@app.get("/convert")
async def convert(location: str):
    try:
        result = await client.search(location)
        feature = result.features[0]
        return {
            "descriptor": feature.properties.descriptor,
            "coordinates": feature.geometry.coordinates,
        }
    except ValidationError as e:
        raise HTTPException(status_code=400, detail=str(e))
    except NotFoundError:
        raise HTTPException(status_code=404, detail="Location not found")

Rate limit handling

The SDK parses the Retry-After header into RateLimitError.retry_after for easy backoff:

import time
from townshipamerica.exceptions import RateLimitError

def search_with_retry(client, location, max_retries=3):
    for attempt in range(max_retries):
        try:
            return client.search(location)
        except RateLimitError as e:
            if attempt == max_retries - 1:
                raise
            wait = e.retry_after or 2 ** attempt
            time.sleep(wait)

Need help? Contact us.