Skip to content
Open
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
2 changes: 1 addition & 1 deletion smoke.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
load_dotenv()

# Initialize the client with your API key
client = Geocodio(os.getenv("GEOCODIO_API_KEY"))
client = Geocodio(os.getenv("GEOCODIO_API_KEY"), hostname=os.getenv("GEOCODIO_HOST"))

# Single forward geocode
print("\nSingle forward geocode:")
Expand Down
127 changes: 79 additions & 48 deletions src/geocodio/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,11 +18,28 @@

# flake8: noqa: F401
from geocodio.models import (
GeocodingResponse, GeocodingResult, AddressComponents,
Location, GeocodioFields, Timezone, CongressionalDistrict,
CensusData, ACSSurveyData, StateLegislativeDistrict, SchoolDistrict,
Demographics, Economics, Families, Housing, Social,
FederalRiding, ProvincialRiding, StatisticsCanadaData, ListResponse, PaginatedResponse
GeocodingResponse,
GeocodingResult,
AddressComponents,
Location,
GeocodioFields,
Timezone,
CongressionalDistrict,
CensusData,
StateLegislativeDistrict,
SchoolDistrict,
Demographics,
Economics,
Families,
Housing,
Social,
FederalRiding,
ProvincialRiding,
StatisticsCanadaData,
ListResponse,
PaginatedResponse,
FFIECData,
ZIP4Data,
)
from geocodio.exceptions import InvalidRequestError, AuthenticationError, GeocodioServerError, BadRequestError

Expand Down Expand Up @@ -402,6 +419,33 @@ def _parse_list_response(response_json: dict, response: httpx.Response = None) -
http_response=response,
)


@staticmethod
def _parse_acs_metric(metric: str, acs_data: dict):
if metric == "demographics":
return Demographics.from_api(acs_data)
elif metric == "economics":
return Economics.from_api(acs_data)
elif metric == "families":
return Families.from_api(acs_data)
elif metric == "housing":
return Housing.from_api(acs_data)
elif metric == "social":
return Social.from_api(acs_data)
else:
return ValueError(f"Unknown ACS metric: {metric}")


@property
def _valid_field_names(self) -> set[str]:
from dataclasses import fields as dataclass_fields
return {f.name for f in dataclass_fields(GeocodioFields)}


def _filter_dict_for_valid_fields(self, data: dict) -> dict:
return {k: v for k, v in data.items() if k in self._valid_field_names}


def _parse_fields(self, fields_data: dict | None) -> GeocodioFields | None:
if not fields_data:
return None
Expand Down Expand Up @@ -443,46 +487,24 @@ def _parse_fields(self, fields_data: dict | None) -> GeocodioFields | None:
for district in fields_data["school"]
]

# Dynamically parse all census fields (e.g., census2010, census2020, census2024, etc.)
# This supports any census year returned by the API
from dataclasses import fields as dataclass_fields
valid_field_names = {f.name for f in dataclass_fields(GeocodioFields)}

census_fields = {}
for key in fields_data:
if key.startswith("census") and key[6:].isdigit(): # e.g., "census2024"
# Only include if it's a defined field in GeocodioFields
if key in valid_field_names:
census_fields[key] = CensusData.from_api(fields_data[key])

acs = (
ACSSurveyData.from_api(fields_data["acs"])
if "acs" in fields_data else None
)

demographics = (
Demographics.from_api(fields_data["acs-demographics"])
if "acs-demographics" in fields_data else None
)

economics = (
Economics.from_api(fields_data["acs-economics"])
if "acs-economics" in fields_data else None
)
census_fields = self._filter_dict_for_valid_fields({
f"census{yr}": CensusData.from_api(census_data)
for yr, census_data in fields_data.get("census", dict()).items()
})

families = (
Families.from_api(fields_data["acs-families"])
if "acs-families" in fields_data else None
)
acs_fields = self._filter_dict_for_valid_fields({
metric: self._parse_acs_metric(metric, acs_data)
for metric, acs_data in fields_data.get("acs", dict()).items()
})

housing = (
Housing.from_api(fields_data["acs-housing"])
if "acs-housing" in fields_data else None
ffiec = (
FFIECData.from_api(fields_data["ffiec"])
if "ffiec" in fields_data else None
)

social = (
Social.from_api(fields_data["acs-social"])
if "acs-social" in fields_data else None
zip4 = (
ZIP4Data.from_api(fields_data["zip4"])
if "zip4" in fields_data else None
)

# Canadian fields
Expand Down Expand Up @@ -512,17 +534,26 @@ def _parse_fields(self, fields_data: dict | None) -> GeocodioFields | None:
state_legislative_districts=state_legislative_districts,
state_legislative_districts_next=state_legislative_districts_next,
school_districts=school_districts,
acs=acs,
demographics=demographics,
economics=economics,
families=families,
housing=housing,
social=social,
riding=riding,
provriding=provriding,
provriding_next=provriding_next,
statcan=statcan,
**census_fields, # Dynamically include all census year fields
ffiec=ffiec,
zip4=zip4,
extras={
k: v for k, v in
fields_data.items() if k not in {
# add other keys here as we support them natively
"timezone",
"congressional_districts",
"ffiec",
"census",
"acs",
"zip4",
}
},
**census_fields,
**acs_fields,
)

# @TODO add a "keep_trying" parameter to download() to keep trying until the list is processed.
Expand Down
56 changes: 48 additions & 8 deletions src/geocodio/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -108,13 +108,19 @@ class CensusData(ApiModelMixin):
Census data for a location.
"""

block: Optional[str] = None
blockgroup: Optional[str] = None
tract: Optional[str] = None
census_year: Optional[int] = None
block_code: Optional[str] = None
block_group: Optional[str] = None
tract_code: Optional[str] = None
full_fips: Optional[str] = None
county_fips: Optional[str] = None
state_fips: Optional[str] = None
msa_code: Optional[str] = None # Metropolitan Statistical Area
csa_code: Optional[str] = None # Combined Statistical Area
place: Optional[dict] = None
metro_micro_statistical_area: Optional[dict] = None
combined_statistical_area: Optional[dict] = None
metropolitan_division: Optional[dict] = None
county_subdivision: Optional[dict] = None
source: Optional[str] = None
extras: Dict[str, Any] = field(default_factory=dict, repr=False)


Expand Down Expand Up @@ -225,9 +231,18 @@ class Social(ApiModelMixin):
class ZIP4Data(ApiModelMixin):
"""USPS ZIP+4 code and delivery information."""

zip4: str
delivery_point: str
carrier_route: str
record_type: Optional[Dict[str, Any]]
residential: Optional[bool]
carrier_route: Optional[Dict[str, Any]]
plus4: Optional[list]
zip9: Optional[list]
facility_code: Optional[Dict[str, Any]]
city_delivery: Optional[bool]
valid_delivery_area: Optional[bool]
exact_match: Optional[bool]
building_or_firm_name: Optional[str]
government_building: Optional[bool]

extras: Dict[str, Any] = field(default_factory=dict, repr=False)


Expand Down Expand Up @@ -280,6 +295,28 @@ class FFIECData(ApiModelMixin):
"""FFIEC CRA/HMDA Data (Beta)."""

# Add FFIEC specific fields as they become available
collection_year: Optional[int]
msa_md_code: Optional[str]
fips_state_code: Optional[str]
fips_county_code: Optional[str]
census_tract: Optional[str]
principal_city: Optional[bool]
small_county: Optional[Dict[str, Any]]
split_tract: Optional[Dict[str, Any]]
demographic_data: Optional[Dict[str, Any]]
urban_rural_flag: Optional[Dict[str, Any]]
msa_md_median_family_income: Optional[int]
msa_md_median_household_income: Optional[int]
tract_median_family_income_percentage: Optional[float]
ffiec_estimated_msa_md_median_family_income: Optional[int]
income_indicator: Optional[str]
cra_poverty_criteria: Optional[bool]
cra_unemployment_criteria: Optional[bool]
cra_distressed_criteria: Optional[bool]
cra_remote_rural_low_density_criteria: Optional[bool]
previous_year_cra_distressed_criteria: Optional[bool]
previous_year_cra_underserved_criterion: Optional[bool]
meets_current_previous_criteria: Optional[bool]
extras: Dict[str, Any] = field(default_factory=dict, repr=False)


Expand Down Expand Up @@ -332,6 +369,9 @@ class GeocodioFields:
provriding_next: Optional[ProvincialRiding] = None
statcan: Optional[StatisticsCanadaData] = None

# catch‑all for any future fields
extras: Dict[str, Any] = field(default_factory=dict, repr=False)


# ──────────────────────────────────────────────────────────────────────────────
# Main result objects
Expand Down