"""A thin, typed client for the Close API (https://api.close.city).
Design goals: mirror the public OpenAPI contract exactly, surface the metering
and conditional-request mechanics (token headers, ETag/304) as first-class, map
``problem+json`` errors to precise exceptions, and iterate keyset-paginated
endpoints transparently. Cursors are opaque and never constructed client-side.
"""
from __future__ import annotations
from collections.abc import Iterator, Sequence
from dataclasses import dataclass, field
from typing import Any
import httpx
from . import errors
__all__ = ["Client", "Reply", "Paginator"]
DEFAULT_BASE_URL = "https://api.close.city"
# The three output modes, shared with the R SDK. "spatial" returns a GeoDataFrame
# where geometry applies (and a plain DataFrame otherwise); "tabular" returns a
# plain DataFrame everywhere and never downloads block boundaries; "raw" returns
# the underlying Reply / Paginator.
OUTPUT_MODES = ("spatial", "tabular", "raw")
def _check_output(mode: str) -> str:
if mode not in OUTPUT_MODES:
raise ValueError(
f"output must be one of {OUTPUT_MODES}, not {mode!r}."
)
return mode
def _clean(params: dict[str, Any] | None) -> dict[str, Any]:
"""Drop None-valued params (so optional filters simply vanish)."""
return {k: v for k, v in (params or {}).items() if v is not None}
def _as_list(value: Any) -> Any:
"""Wrap a scalar in a list, for POST-body fields the API requires as arrays.
``None`` and existing lists/tuples pass through unchanged."""
if value is None or isinstance(value, (list, tuple)):
return value
return [value]
def _int(value: str | None) -> int | None:
return int(value) if value is not None else None
def _retry_after(value: str | None) -> float | None:
try:
return float(value) if value is not None else None
except ValueError:
return None
[docs]
@dataclass
class Reply:
"""One API response, with the metering + caching metadata exposed.
``data`` is the parsed JSON body (a ``dict`` for every route). For a ``304``
revalidation ``not_modified`` is True and ``data`` is None. ``tokens_charged``
/ ``tokens_remaining`` are None on free and member-unmetered responses.
"""
data: Any
status: int
tokens_charged: int | None = None
tokens_remaining: int | None = None
etag: str | None = None
request_id: str | None = None
not_modified: bool = False
@property
def results(self) -> list[dict]:
"""The ``results`` array of a list endpoint (empty if absent)."""
return self.data.get("results", []) if isinstance(self.data, dict) else []
@property
def next_cursor(self) -> str | None:
return self.data.get("next_cursor") if isinstance(self.data, dict) else None
[docs]
def to_geopandas(self, **kwargs):
"""Convert this reply to a :class:`geopandas.GeoDataFrame`.
Points for POI replies, polygons for isochrone ``format=geojson``. Block
replies need a ``block_geometry=`` GeoDataFrame or ``fetch=True`` (which
downloads TIGER blocks with ``pygris``). See
:func:`closecity.spatial.to_geopandas`.
"""
from . import spatial
return spatial.to_geopandas(self, **kwargs)
[docs]
def to_pandas(self, **kwargs):
"""Convert this reply to a :class:`pandas.DataFrame` (no geometry). See
:func:`closecity.tabular.to_pandas`.
"""
from . import tabular
return tabular.to_pandas(self, **kwargs)
[docs]
@dataclass
class Paginator:
"""Lazily follows ``next_cursor`` across pages.
Iterate it directly to get every record; use :meth:`pages` for per-page
:class:`Reply` objects (each carrying its own token/ETag metadata); use
:meth:`page` to fetch a single page.
"""
_client: Client
_method: str
_path: str
_params: dict[str, Any] = field(default_factory = dict)
_json: dict[str, Any] | None = None
_cursor_in: str = "params" # "params" for GETs, "json" for POST body
def page(self, cursor: str | None = None) -> Reply:
params = dict(self._params)
body = dict(self._json) if self._json is not None else None
if cursor is not None:
if self._cursor_in == "json":
body = {**(body or {}), "cursor": cursor}
else:
params["cursor"] = cursor
return self._client._request(
self._method, self._path, params = params, json = body
)
def pages(self) -> Iterator[Reply]:
cursor = None
while True:
reply = self.page(cursor)
yield reply
cursor = reply.next_cursor
if not cursor:
return
def __iter__(self) -> Iterator[dict]:
for reply in self.pages():
yield from reply.results
[docs]
def to_geopandas(self, **kwargs):
"""Collect every page and convert the rows to a
:class:`geopandas.GeoDataFrame`. See
:func:`closecity.spatial.to_geopandas`.
"""
from . import spatial
return spatial.to_geopandas(self, **kwargs)
[docs]
def to_pandas(self, **kwargs):
"""Collect every page and convert the rows to a
:class:`pandas.DataFrame`. See :func:`closecity.tabular.to_pandas`.
"""
from . import tabular
return tabular.to_pandas(self, **kwargs)
[docs]
class Client:
"""Client for the Close API.
``api_key`` is optional (the catalog and health routes are free), but every
data route needs one (a ``ck_live_`` key), created at
https://account.close.city. Usable as a context manager.
``output`` sets how results come back, and defaults to ``"spatial"``:
* ``"spatial"`` returns a :class:`geopandas.GeoDataFrame` for routes with
geometry (points, isochrone and block polygons) and a plain
:class:`pandas.DataFrame` for the rest;
* ``"tabular"`` returns a plain :class:`pandas.DataFrame` for every route and
never downloads block boundaries (the cheap path when you only want the
numbers);
* ``"raw"`` returns the underlying :class:`Reply` / :class:`Paginator`.
Set it on the client (``close.output = "raw"``) or per call
(``close.blocks_query(..., output = "tabular")``).
"""
def __init__(
self,
api_key: str | None = None,
*,
base_url: str = DEFAULT_BASE_URL,
timeout: float = 30.0,
output: str = "spatial",
http_client: httpx.Client | None = None,
transport: httpx.BaseTransport | None = None,
):
self.output = _check_output(output)
headers = {"Accept": "application/json"}
if api_key:
headers["Authorization"] = f"Bearer {api_key}"
self._http = http_client or httpx.Client(
base_url = base_url.rstrip("/"), timeout = timeout, headers = headers,
transport = transport,
)
def _resolve_output(self, output: str | None) -> str:
return _check_output(output if output is not None else self.output)
def _deliver(self, source, *, geometry: bool, key: str | None = None,
output: str | None = None):
"""Return ``source`` shaped per the resolved output mode. ``geometry``
says whether this route can carry geometry in ``spatial`` mode; ``key``
names the record array for the tabular path."""
mode = self._resolve_output(output)
if mode == "raw":
return source
# A 304 revalidation has no body to convert.
if isinstance(source, Reply) and source.not_modified:
return source
if mode == "spatial" and geometry:
from . import spatial
return spatial.to_geopandas(source, fetch = True)
from . import tabular
return tabular.to_pandas(source, key = key)
# -- context manager ----------------------------------------------------
def __enter__(self) -> Client:
return self
def __exit__(self, *exc: object) -> None:
self.close()
def close(self) -> None:
self._http.close()
# -- core ---------------------------------------------------------------
def _request(
self,
method: str,
path: str,
*,
params: dict[str, Any] | None = None,
json: dict[str, Any] | None = None,
if_none_match: str | None = None,
) -> Reply:
headers = {"If-None-Match": if_none_match} if if_none_match else None
resp = self._http.request(
method, path, params = _clean(params), json = json, headers = headers
)
request_id = resp.headers.get("X-Request-Id")
if resp.status_code == 304:
return Reply(
data = None, status = 304, not_modified = True,
etag = resp.headers.get("ETag"), request_id = request_id,
)
if resp.status_code >= 400:
body = resp.json() if resp.content else {}
if not isinstance(body, dict):
body = {"title": str(body)}
raise errors.error_from_problem(
resp.status_code, body, request_id,
_retry_after(resp.headers.get("Retry-After")),
)
return Reply(
data = resp.json() if resp.content else None,
status = resp.status_code,
tokens_charged = _int(resp.headers.get("X-Tokens-Charged")),
tokens_remaining = _int(resp.headers.get("X-Tokens-Remaining")),
etag = resp.headers.get("ETag"),
request_id = request_id,
)
def _get(self, path: str, params = None, *, if_none_match = None) -> Reply:
return self._request("GET", path, params = params, if_none_match = if_none_match)
# -- catalog (free) -----------------------------------------------------
[docs]
def health(self) -> Reply:
"""Liveness check (free). Touches no database. Always a raw `Reply`."""
return self._get("/v1/health")
[docs]
def last_updated(self) -> Reply:
"""Publication timestamp of the newest published data (free). Always a
raw `Reply`."""
return self._get("/v1/last-updated")
[docs]
def modes(self, *, output: str | None = None):
"""The travel modes and their numeric ids (free). A DataFrame of
`mode_id, mode, description`, or a raw `Reply` when `output="raw"`."""
return self._deliver(self._get("/v1/meta/modes"),
geometry = False, key = "modes", output = output)
[docs]
def destination_types(self, *, output: str | None = None):
"""The destination-type taxonomy with leaf expansions (free). A DataFrame
of `dest_type_id, name, label, is_leaf, leaf_ids`, or a raw `Reply` when
`output="raw"`. Filtering by a parent type expands to its `leaf_ids`."""
return self._deliver(self._get("/v1/meta/destination-types"),
geometry = False, key = "destination_types",
output = output)
[docs]
def vintage(self, *, output: str | None = None):
"""The active version of each dataset component (free). A DataFrame of
`component, version, effective_date`, or a raw `Reply`."""
return self._deliver(self._get("/v1/meta/vintage"),
geometry = False, key = "components", output = output)
[docs]
def places(self, q: str, *, limit: int | None = None, output: str | None = None):
"""Search census places by name (free). Each match carries the place's
GEOID and WGS84 centroid. Feed a centre into `blocks_query(center=...)`,
or a GEOID into `place_blocks(...)`. `q` is a name substring; `limit`
caps the matches (1 to 20). A points GeoDataFrame (or a plain DataFrame
under `output="tabular"`, a raw `Reply` under `output="raw"`)."""
return self._deliver(self._get("/v1/places", {"q": q, "limit": limit}),
geometry = True, key = "places", output = output)
# -- origin block / point (metered) -------------------------------------
[docs]
def block_summary(
self, geoid: str, *, mode = None, type = None,
if_none_match: str | None = None, output: str | None = None,
):
"""Fastest travel time to each destination category from a census block,
by mode. `geoid` is a 15-digit block GEOID; `mode`/`type` are optional
filters (scalar or list). Supports `if_none_match` (ETag/304). A DataFrame
(the origin GEOID broadcast to a `geoid` column), or a raw `Reply`."""
return self._deliver(
self._get(f"/v1/blocks/{geoid}/summary",
{"mode": mode, "type": type}, if_none_match = if_none_match),
geometry = False, key = "results", output = output,
)
[docs]
def block_pois(
self, geoid: str, *, mode = None, type = None, dest_id = None,
max_minutes = None, limit = None, output: str | None = None,
):
"""Every nearby POI and its travel time from a block, one row per (POI,
mode). A GeoDataFrame of points (a plain DataFrame under
`output="tabular"`, a `Paginator` under `output="raw"`)."""
return self._deliver(Paginator(
self, "GET", f"/v1/blocks/{geoid}/pois",
_clean({"mode": mode, "type": type, "dest_id": dest_id,
"max_minutes": max_minutes, "limit": limit}),
), geometry = True, output = output)
[docs]
def point_summary(
self, lat: float, lon: float, *, mode = None, type = None,
if_none_match: str | None = None, output: str | None = None,
):
"""Like `block_summary`, but from the census block containing a lat/lon.
The resolved block GEOID is echoed as `resolved_block` and broadcast to a
`geoid` column. A DataFrame, or a raw `Reply`."""
return self._deliver(
self._get("/v1/point/summary",
{"lat": lat, "lon": lon, "mode": mode, "type": type},
if_none_match = if_none_match),
geometry = False, key = "results", output = output,
)
[docs]
def point_pois(
self, lat: float, lon: float, *, mode = None, type = None, dest_id = None,
max_minutes = None, limit = None, output: str | None = None,
):
"""Like `block_pois`, but from the block containing a lat/lon. A
GeoDataFrame of points (a plain DataFrame under `output="tabular"`, a
`Paginator` under `output="raw"`)."""
return self._deliver(Paginator(
self, "GET", "/v1/point/pois",
_clean({"lat": lat, "lon": lon, "mode": mode, "type": type,
"dest_id": dest_id, "max_minutes": max_minutes, "limit": limit}),
), geometry = True, output = output)
# -- POI search / detail / catchment (metered) --------------------------
[docs]
def pois_search(
self, *, lat = None, lon = None, radius_m = None, bbox = None, type = None,
q = None, limit = None, output: str | None = None,
):
"""Search POIs by bounding box (``bbox``) or radius (``lat`` + ``lon`` +
``radius_m``). A GeoDataFrame of points (a plain DataFrame under
`output="tabular"`, a `Paginator` under `output="raw"`)."""
return self._deliver(Paginator(
self, "GET", "/v1/pois",
_clean({"lat": lat, "lon": lon, "radius_m": radius_m, "bbox": bbox,
"type": type, "q": q, "limit": limit}),
), geometry = True, output = output)
[docs]
def poi(self, dest_id: int, *, if_none_match: str | None = None,
output: str | None = None):
"""Name, location, address, types, and whitelisted attributes for one POI.
A one-row points GeoDataFrame (a plain DataFrame under `output="tabular"`,
a `Reply` under `output="raw"`)."""
return self._deliver(
self._get(f"/v1/pois/{dest_id}", if_none_match = if_none_match),
geometry = True, output = output,
)
[docs]
def poi_catchment(
self, dest_id: int, *, mode = None, block = None, max_minutes = None,
limit = None, output: str | None = None,
):
"""Every census block that can reach a POI, one row per (block, mode).
A GeoDataFrame of block polygons (a plain DataFrame under
`output="tabular"`, a `Paginator` under `output="raw"`)."""
return self._deliver(Paginator(
self, "GET", f"/v1/pois/{dest_id}/catchment",
_clean({"mode": mode, "block": block, "max_minutes": max_minutes,
"limit": limit}),
), geometry = True, output = output)
# -- areal (metered) ----------------------------------------------------
[docs]
def blocks_query(
self, *, polygon: dict | None = None, center: dict | None = None,
radius_m: float | None = None, type: Sequence[int] | None = None,
mode: Sequence[str] | None = None, include_population: bool | None = None,
limit: int | None = None, output: str | None = None,
):
"""Blocks within a GeoJSON ``polygon`` or a ``center`` + ``radius_m``.
Rows carry the numeric `mode_id` (join `modes()` to label it). A
GeoDataFrame of block polygons (a plain DataFrame under
`output="tabular"`, a `Paginator` under `output="raw"`)."""
body = _clean({
"polygon": polygon, "center": center, "radius_m": radius_m,
"type": _as_list(type), "mode": _as_list(mode),
"include_population": include_population, "limit": limit,
})
return self._deliver(
Paginator(self, "POST", "/v1/blocks/query",
_json = body, _cursor_in = "json"),
geometry = True, output = output,
)
[docs]
def place_blocks(
self, geoid: str, *, mode = None, type = None, include_population = None,
limit = None, output: str | None = None,
):
"""Per-block travel times for every census block in a place (city or
town), by place GEOID. Rows carry the numeric `mode_id` (join `modes()`
to label it). A GeoDataFrame of block polygons (a plain DataFrame under
`output="tabular"`, a `Paginator` under `output="raw"`)."""
return self._deliver(Paginator(
self, "GET", f"/v1/places/{geoid}/blocks",
_clean({"mode": mode, "type": type,
"include_population": include_population, "limit": limit}),
), geometry = True, output = output)
# -- isochrone ----------------------------------------------------------
[docs]
def isochrone(
self, *, block = None, lon = None, lat = None, mode = None, direction = None,
minutes = None, contours = None, format = None, v = None,
if_none_match: str | None = None, output: str | None = None,
):
"""Travel-time contours from a ``block`` (GEOID) or ``lon`` + ``lat``.
Pass ``minutes`` (single) or ``contours`` (up to 4 ascending levels, a
list or a comma string). In `spatial` mode `format="geojson"` gives
contour polygons and `format="blocks"` gives block polygons; `tabular`
gives the matching rows; `raw` gives a `Reply`."""
if isinstance(contours, (list, tuple)):
contours = ",".join(str(c) for c in contours)
reply = self._get(
"/v1/isochrone",
{"block": block, "lon": lon, "lat": lat, "mode": mode,
"direction": direction, "minutes": minutes, "contours": contours,
"format": format, "v": v},
if_none_match = if_none_match,
)
return self._deliver(reply, geometry = True, output = output)