prismadata 0.4.1

Python client for the PrismaData location intelligence API

prismadata

Python client for the PrismaData location intelligence API.

Installation

pip install prismadata

With optional extras:

pip install prismadata[pandas]     # DataFrame enrichment
pip install prismadata[sklearn]    # scikit-learn transformer
pip install prismadata[all]        # everything (pandas, sklearn, cache, progress bars)

Quick Start

from prismadata import Client

client = Client(api_key="your-api-key")

# Geocode an address
result = client.geocode(full_address="Av Paulista 1000, Sao Paulo")
print(result["prismadata__geocoder__latitude"], result["prismadata__geocoder__longitude"])

# Query slum proximity
slum = client.slum(lat=-23.56, lng=-46.65)
print(slum["prismadata__favela__distancia_m"])

# Calculate a route
route = client.route([(-23.56, -46.65), (-23.57, -46.66)])
print(route["prismadata__routing_route__distancia_m"])

Authentication

The client supports two authentication methods:

# Using API key
client = Client(api_key="your-api-key")

# Using username and password
client = Client(username="your-user", password="your-pass")

Credentials can also be provided via environment variables:

export PRISMADATA_APIKEY="your-api-key"
# or
export PRISMADATA_USERNAME="your-user"
export PRISMADATA_PASSWORD="your-pass"

# Picks up credentials from environment automatically
client = Client()

Credential resolution order: explicit api_key > explicit username/password > PRISMADATA_APIKEY env var > PRISMADATA_USERNAME+PRISMADATA_PASSWORD env vars.

DataFrame Enrichment

import pandas as pd
from prismadata import Client

client = Client(api_key="your-api-key")

df = pd.DataFrame({
    "lat": [-23.56, -23.57, -23.58],
    "lng": [-46.65, -46.66, -46.67],
})

enriched = client.enrich(df, services=["slum", "income_static", "infosc"])
print(enriched.columns.tolist())
# ['lat', 'lng', 'prismadata__favela__distancia_m', ..., 'prismadata__personal_income_static__percentil_br', ...]

# Use clean_columns=True for shorter column names
client = Client(api_key="your-api-key", clean_columns=True)
enriched = client.enrich(df, services=["slum"])
# Column names: 'favela_distancia_m', 'favela_nome', ...

scikit-learn Pipeline

from sklearn.pipeline import Pipeline
from sklearn.ensemble import RandomForestClassifier
from prismadata.sklearn import PrismaDataTransformer

pipe = Pipeline([
    ("enrich", PrismaDataTransformer(
        api_key="your-api-key",
        services=["slum", "income_static"],
    )),
    ("model", RandomForestClassifier()),
])

pipe.fit(X_train, y_train)

Async Client

All methods are available asynchronously via AsyncClient:

from prismadata import AsyncClient

async with await AsyncClient.create(api_key="your-api-key") as client:
    result = await client.slum(lat=-23.56, lng=-46.65)
    print(result)

    # Batch and enrichment work the same way
    enriched = await client.enrich(df, services=["slum", "income_static"])

Error Handling

from prismadata import Client
from prismadata.exceptions import (
    AuthenticationError,
    BatchError,
    RateLimitError,
    PrismaDataError,
)

try:
    result = client.slum_batch(large_point_dict)
except BatchError as e:
    # Some chunks succeeded, some failed
    print(f"Got {len(e.partial_results)} results, {len(e.failed_keys)} failed")
    for key, value in e.partial_results.items():
        process(key, value)  # use what succeeded
    retry(e.failed_keys)     # retry what failed
except RateLimitError:
    print("Rate limit exceeded, wait and retry")
except AuthenticationError:
    print("Invalid credentials")
except PrismaDataError as e:
    print(f"API error {e.status_code}: {e}")

Available Methods

Geocoding

• client.geocode(full_address=..., zipcode=..., city=..., state=...) - Address to coordinates
• client.reverse_geocode(lat, lng) - Coordinates to address

Location Services

• client.slum(lat, lng) - Nearest slum/favela proximity
• client.prison(lat, lng) - Nearest prison proximity
• client.border(lat, lng) - Border proximity
• client.infosc(lat, lng) - Census sector info
• client.income_static(lat, lng) - Income percentiles
• client.income_pdf(lat, lng, gender=..., age=...) - Detailed income statistics

Routing

• client.route(points, profile="car") - Route between points
• client.isochrone(lat, lng, time_limit=600, profile="car") - Reachable area

Address Validation

• client.compare_address(lat, lng, full_address=...) - Compare address with coordinates
• client.validate_address(locations, addresses) - Validate against location history
• client.cluster_locations(locations) - Cluster location history

Credit

• client.precatory(cpf_cnpj=...) - Credit summary (precatorios/RPVs)
• client.precatory_detail(cpf_cnpj=...) - Detailed credit list

Batch Operations

• client.slum_batch(points) - Batch slum queries
• client.prison_batch(points) - Batch prison queries
• client.border_batch(points) - Batch border queries
• client.infosc_batch(points) - Batch census sector queries
• client.route_batch(items, profile="car") - Batch routing
• client.isochrone_batch(items, profile="car") - Batch isochrones

Aggregator

• client.aggregate(lat, lng, services=[...]) - Multiple services in one call
• client.aggregate_batch(points, services=[...]) - Batch aggregation
• client.geocode_aggregate(full_address=..., services=[...]) - Geocode + aggregate

Configuration

client = Client(
    api_key="your-key",
    timeout=30,            # Request timeout (seconds)
    cache=True,            # Enable disk cache (requires diskcache)
    cache_ttl=86400,       # Cache TTL (seconds)
    clean_columns=False,   # Keep 'prismadata__' prefix (default)
    show_progress=True,    # Show tqdm progress bars
    app_name="my-app",     # Sent as X-App header on every request
)

License

MIT