Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0

## [Unreleased]

### Added
- Public census accessors on `GeocodioFields`: `fields.census` (the requested append, most recent vintage when several are present), `fields.get_census(year)` (accepts `2023`, `"2023"` or `"census2023"`), `fields.census_years` and `fields.census_data`. The private `fields._census` dict and the dynamic `fields.census2023` attributes are unchanged.
- `GeocodingResult.match_type` and `GeocodingResult.address_lines`, both of which the API returns on every result and the model previously dropped.
- Raw response access: `GeocodingResponse.raw` / `.to_dict()` for the full untouched JSON payload, and `GeocodingResult.raw` / `.to_dict()` for a single result. Unlike `dataclasses.asdict()` these keep every key the API sent.
- Rate limit headers are parsed into a `RateLimit` model, exposed as `GeocodingResponse.rate_limit` and `Geocodio.rate_limit` (updated on every request, including ones that raise).

## [1.3.0] - 2026-08-25

### Added
Expand Down
58 changes: 58 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -114,6 +114,64 @@ response = client.geocode(
)
```

#### Census appends

Census data is keyed by vintage. `fields.census` gives you the append you
requested, and `fields.get_census(year)` picks a specific vintage when you
requested more than one:

```python
response = client.geocode("1109 N Highland St, Arlington VA", fields=["census2023"])
census = response.results[0].fields.census

print(census.full_fips) # "510131018012004"
print(census.census_year) # 2023

# When several vintages were requested
fields = response.results[0].fields
fields.get_census(2023) # also accepts "2023" or "census2023"
fields.census_years # [2023]
fields.census_data # {"census2023": CensusData(...)}
```

`fields.census` returns the most recent vintage present. Accessing a vintage
directly (`fields.census2023`) continues to work.

### Raw API responses

Every geocoding response keeps the untouched JSON payload the API returned, so
you can cache the full response and derive new columns later without paying for
another lookup:

```python
response = client.geocode("1109 N Highland St, Arlington VA")

response.raw # the full JSON payload, exactly as returned
response.to_dict() # a deep copy of the same payload

result = response.results[0]
result.raw # the JSON object for this result
result.match_type # "rooftop", "unit", "building_centroid" or None
result.address_lines # ["1109 N Highland St", "", "Arlington, VA 22201"]
```

### Rate limits

The `X-RateLimit-*` response headers are parsed onto every geocoding response,
and the most recent values are kept on the client (including for requests that
raise):

```python
response = client.geocode("1109 N Highland St, Arlington VA")

response.rate_limit.limit # 1000
response.rate_limit.remaining # 999
response.rate_limit.period # 60 (seconds), when sent by the API
response.rate_limit.reset # unix timestamp, when sent by the API

client.rate_limit # the most recent rate limit state seen
```

### Address components

For forward geocoding requests it is possible to supply [individual address components](https://www.geocod.io/docs/#single-address) instead of a full address string:
Expand Down
34 changes: 29 additions & 5 deletions src/geocodio/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,7 @@
Location,
PaginatedResponse,
ProvincialRiding,
RateLimit,
SchoolDistrict,
Social,
StateLegislativeDistrict,
Expand Down Expand Up @@ -117,6 +118,7 @@ def __init__(
self._http = httpx.Client(
base_url=f"https://{self.hostname}", verify=verify_ssl
)
self.rate_limit: Optional[RateLimit] = None

# ──────────────────────────────────────────────────────────────────────────
# Public methods
Expand Down Expand Up @@ -226,7 +228,7 @@ def geocode(
response = self._request(
"POST" if data else "GET", endpoint, params, json=data, timeout=timeout
)
return self._parse_geocoding_response(response.json())
return self._parse_geocoding_response(response.json(), response=response)

def reverse(
self,
Expand Down Expand Up @@ -306,7 +308,7 @@ def reverse(
response = self._request(
"POST" if data else "GET", endpoint, params, json=data, timeout=timeout
)
return self._parse_geocoding_response(response.json())
return self._parse_geocoding_response(response.json(), response=response)

# ──────────────────────────────────────────────────────────────────────────
# Internal helpers
Expand Down Expand Up @@ -350,6 +352,12 @@ def _request(
logger.debug(f"Response headers: {resp.headers}")
logger.debug(f"Response body: {resp.content}")

# Keep the most recent rate limit state available to callers, including
# for responses that raise below.
rate_limit = RateLimit.from_headers(resp.headers)
if rate_limit is not None:
self.rate_limit = rate_limit

resp = self._handle_error_response(resp)

return resp
Expand All @@ -370,9 +378,15 @@ def _handle_error_response(self, resp) -> httpx.Response:
f"Unrecognized status code {resp.status_code}: {resp.text}"
)

def _parse_geocoding_response(self, response_json: dict) -> GeocodingResponse:
def _parse_geocoding_response(
self,
response_json: dict,
response: Optional[httpx.Response] = None,
) -> GeocodingResponse:
logger.debug(f"Raw response: {response_json}")

rate_limit = RateLimit.from_headers(response.headers) if response else None

# Handle batch response format
if (
"results" in response_json
Expand Down Expand Up @@ -416,9 +430,14 @@ def _parse_geocoding_response(self, response_json: dict) -> GeocodingResponse:
query=query,
fields=self._parse_fields(top.get("fields")),
stable_address_key=top.get("stable_address_key"),
match_type=top.get("match_type"),
address_lines=top.get("address_lines"),
raw=top,
)
)
return GeocodingResponse(results=results)
return GeocodingResponse(
results=results, raw=response_json, rate_limit=rate_limit
)

# Handle single response format
results = [
Expand All @@ -433,10 +452,15 @@ def _parse_geocoding_response(self, response_json: dict) -> GeocodingResponse:
source=res.get("source", ""),
fields=self._parse_fields(res.get("fields")),
stable_address_key=res.get("stable_address_key"),
match_type=res.get("match_type"),
address_lines=res.get("address_lines"),
raw=res,
)
for res in response_json.get("results", [])
]
return GeocodingResponse(results=results)
return GeocodingResponse(
results=results, raw=response_json, rate_limit=rate_limit
)

# ──────────────────────────────────────────────────────────────────────────
# List API methods
Expand Down
129 changes: 129 additions & 0 deletions src/geocodio/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@

from __future__ import annotations

import copy
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Tuple, Type, TypeVar

Expand Down Expand Up @@ -398,6 +399,48 @@ class GeocodioFields:
# Internal storage for census data (all years dynamically accessible)
_census: Dict[str, CensusData] = field(default_factory=dict, repr=False)

@property
def census(self) -> Optional[CensusData]:
"""
The census append, exposed like every other append on this object.

Geocodio keys census data by vintage (``census2020``, ``census2023``,
…). Requesting a single census field – the common case – makes
``fields.census`` that field. When several vintages are present the
most recent one is returned; use :meth:`get_census` to pick one.
"""
return self.get_census()

@property
def census_years(self) -> List[int]:
"""Census vintages present on this result, oldest first."""
return sorted(int(name[6:]) for name in self._census if name[6:].isdigit())

@property
def census_data(self) -> Dict[str, CensusData]:
"""All census vintages, keyed as requested (``{"census2023": ...}``)."""
return dict(self._census)

def get_census(self, year: Optional[Any] = None) -> Optional[CensusData]:
"""
Return the census append for ``year``.

``year`` accepts the vintage in any form the API or a caller might use
– ``2023``, ``"2023"`` or ``"census2023"``. Omit it to get the most
recent vintage on this result. Returns ``None`` when the requested
vintage was not appended.
"""
if year is None:
years = self.census_years
if not years:
return None
return self._census.get(f"census{years[-1]}")

key = str(year)
if not key.startswith("census"):
key = f"census{key}"
return self._census.get(key)

def __getattr__(self, name: str):
"""
Dynamic attribute access for census years (census2020, census2025, etc.).
Expand All @@ -419,6 +462,62 @@ def __getattr__(self, name: str):
)


# ──────────────────────────────────────────────────────────────────────────────
# Rate limiting
# ──────────────────────────────────────────────────────────────────────────────


@dataclass(slots=True, frozen=True)
class RateLimit:
"""
Rate limit state as reported by the ``X-RateLimit-*`` response headers.

Attributes:
limit: Requests allowed in the current window.
remaining: Requests left in the current window.
reset: Unix timestamp when the window resets, when the API sends it.
period: Length of the window in seconds, when the API sends it.
headers: The raw ``x-ratelimit-*`` headers, lowercased.
"""

limit: Optional[int] = None
remaining: Optional[int] = None
reset: Optional[int] = None
period: Optional[int] = None
headers: Dict[str, str] = field(default_factory=dict, repr=False)

@classmethod
def from_headers(cls, headers: Any) -> Optional["RateLimit"]:
"""
Build a RateLimit from response headers, or None when the response
carries no rate limit information.
"""
if headers is None:
return None

raw = {
key.lower(): value
for key, value in headers.items()
if key.lower().startswith("x-ratelimit-")
}
if not raw:
return None

def as_int(name: str) -> Optional[int]:
try:
return int(raw[f"x-ratelimit-{name}"])
except (KeyError, TypeError, ValueError):
return None

return cls(
limit=as_int("limit"),
remaining=as_int("remaining"),
reset=as_int("reset"),
period=as_int("period"),
headers=raw,
)


# ──────────────────────────────────────────────────────────────────────────────
# Distance API models
# ──────────────────────────────────────────────────────────────────────────────
Expand Down Expand Up @@ -691,20 +790,50 @@ class GeocodingResult:
fields: Optional[GeocodioFields] = None
query: str = ""
stable_address_key: Optional[str] = None
match_type: Optional[str] = None
address_lines: Optional[List[str]] = None
raw: Dict[str, Any] = field(default_factory=dict, repr=False)

@property
def matched(self) -> bool:
"""True when the API returned coordinates for this query."""
return self.location is not None

def to_dict(self) -> Dict[str, Any]:
"""
The untouched JSON object the API returned for this result.

Unlike ``dataclasses.asdict()`` this keeps every key the API sent,
including any the models do not (yet) map. Empty for a query the API
returned no match for – the full payload is still on
``GeocodingResponse.raw``.
"""
return copy.deepcopy(self.raw)


@dataclass(slots=True, frozen=True)
class GeocodingResponse:
"""
Top‑level structure returned by client.geocode() / client.reverse().

Attributes:
results: Flat list of results, one per submitted query.
raw: The untouched JSON payload as returned by the API.
rate_limit: Rate limit state from the response headers, when present.
"""

results: List[GeocodingResult] = field(default_factory=list)
raw: Dict[str, Any] = field(default_factory=dict, repr=False)
rate_limit: Optional[RateLimit] = None

def to_dict(self) -> Dict[str, Any]:
"""
The untouched JSON payload as returned by the API.

Cache this to avoid re-fetching (and re-paying for) a lookup when a
new derived column is needed later.
"""
return copy.deepcopy(self.raw)


@dataclass(slots=True, frozen=True)
Expand Down
Loading
Loading