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

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)

Python SDK Quick Start - Installation, basic usage, and first conversion
API Integration Guide - REST API endpoints, pricing, and integration patterns
Batch Conversion - Web interface for batch processing

Need help? Contact us or open an issue on GitHub.