ATTOM Data APIapi.developer.attomdata.com ↗

"""
ATTOM Data API Client

A Python client for accessing comprehensive property data, ownership information,
and sales trends through the ATTOM Data API.

Get your API key from: https://parse.bot/settings
"""

import os
import requests
from typing import Any, Dict, List, Optional


class ParseClient:
    """
    Client for interacting with the ATTOM Data API through Parse Bot.
    
    Attributes:
        base_url: The base URL for the API
        scraper_id: The unique scraper identifier
        api_key: The API key for authentication
    """
    
    def __init__(self, api_key: Optional[str] = None):
        """
        Initialize the ParseClient.
        
        Args:
            api_key: API key for authentication. If not provided, reads from PARSE_API_KEY env var.
        """
        self.base_url = "https://api.parse.bot"
        self.scraper_id = "7d4d00ce-c23d-4935-a089-995a4e9497d6"
        self.api_key = api_key or os.getenv("PARSE_API_KEY")
        
        if not self.api_key:
            raise ValueError("API key must be provided or set in PARSE_API_KEY environment variable")
    
    def _call(self, endpoint: str, method: str = "POST", **params) -> Dict[str, Any]:
        """
        Make an API call to the Parse Bot scraper.
        
        Args:
            endpoint: The endpoint name (e.g., 'get_property_address')
            method: HTTP method ('GET' or 'POST')
            **params: Parameters to send with the request
            
        Returns:
            Response JSON as a dictionary
            
        Raises:
            requests.RequestException: If the API call fails
        """
        url = f"{self.base_url}/scraper/{self.scraper_id}/{endpoint}"
        headers = {
            "X-API-Key": self.api_key,
            "Content-Type": "application/json"
        }
        
        try:
            if method.upper() == "GET":
                response = requests.get(url, headers=headers, params=params)
            elif method.upper() == "POST":
                response = requests.post(url, headers=headers, json=params)
            else:
                raise ValueError(f"Unsupported HTTP method: {method}")
            
            response.raise_for_status()
            return response.json()
        except requests.RequestException as e:
            print(f"API call failed: {e}")
            raise
    
    def get_property_address(
        self,
        address1: Optional[str] = None,
        address2: Optional[str] = None,
        postalcode: Optional[str] = None,
        latitude: Optional[float] = None,
        longitude: Optional[float] = None,
        radius: Optional[float] = None,
        propertytype: Optional[str] = None,
        page: int = 1,
        pagesize: int = 100
    ) -> Dict[str, Any]:
        """
        Search for properties within a specified area or matching address criteria.
        
        Args:
            address1: First line of property address
            address2: Second line (city, state, zip)
            postalcode: ZIP code to search
            latitude: Latitude for radius search
            longitude: Longitude for radius search
            radius: Search radius in miles
            propertytype: Property type filter
            page: Page number for pagination (default: 1)
            pagesize: Number of results per page (default: 100)
            
        Returns:
            Response containing property array and status
        """
        params = {
            k: v for k, v in {
                "address1": address1,
                "address2": address2,
                "postalcode": postalcode,
                "latitude": latitude,
                "longitude": longitude,
                "radius": radius,
                "propertytype": propertytype,
                "page": page,
                "pagesize": pagesize
            }.items() if v is not None
        }
        
        return self._call("get_property_address", method="GET", **params)
    
    def get_property_detail(
        self,
        attomid: Optional[str] = None,
        address1: Optional[str] = None,
        address2: Optional[str] = None
    ) -> Dict[str, Any]:
        """
        Get rich property details for a specific property.
        
        Args:
            attomid: ATTOM unique property identifier
            address1: First line of address
            address2: Second line of address (city, state, zip)
            
        Returns:
            Response containing detailed property information
        """
        params = {
            k: v for k, v in {
                "attomid": attomid,
                "address1": address1,
                "address2": address2
            }.items() if v is not None
        }
        
        return self._call("get_property_detail", method="GET", **params)
    
    def get_property_detail_owner(self, attomid: str) -> Dict[str, Any]:
        """
        Get detailed property data including current owner information.
        
        Args:
            attomid: ATTOM unique property identifier (required)
            
        Returns:
            Response containing property details with owner information
        """
        return self._call("get_property_detail_owner", method="GET", attomid=attomid)
    
    def get_transaction_sales_trend(
        self,
        geoidv4: str,
        startyear: int,
        endyear: int,
        interval: str = "yearly",
        propertytype: Optional[str] = None,
        page: int = 1,
        pagesize: int = 100
    ) -> Dict[str, Any]:
        """
        Get sales trends for a given geography.
        
        Args:
            geoidv4: Geographic V4 identifier (required, 32-char hex hash)
            startyear: Start year for trend analysis (required)
            endyear: End year for trend analysis (required)
            interval: Trend interval - 'yearly' or 'monthly' (default: 'yearly')
            propertytype: Property type filter
            page: Page number (default: 1)
            pagesize: Results per page (default: 100)
            
        Returns:
            Response containing sales trends array
        """
        params = {
            k: v for k, v in {
                "geoidv4": geoidv4,
                "startyear": startyear,
                "endyear": endyear,
                "interval": interval,
                "propertytype": propertytype,
                "page": page,
                "pagesize": pagesize
            }.items() if v is not None
        }
        
        return self._call("get_transaction_sales_trend", method="GET", **params)
    
    def get_poi_search(
        self,
        address: Optional[str] = None,
        zipcode: Optional[str] = None,
        point: Optional[str] = None,
        radius: Optional[float] = None,
        searchdistance: Optional[float] = None,
        categoryname: Optional[str] = None,
        categoryid: Optional[str] = None,
        lineofbusinessname: Optional[str] = None,
        industryname: Optional[str] = None,
        recordlimit: Optional[int] = None,
        page: int = 1,
        pagesize: int = 10
    ) -> Dict[str, Any]:
        """
        Search for points of interest in proximity to a location.
        
        Args:
            address: Street address to search near
            zipcode: ZIP code to search near
            point: Geographic point in format POINT(longitude,latitude)
            radius: Search radius in miles
            searchdistance: Search distance in miles
            categoryname: POI category name filter
            categoryid: POI category ID filter
            lineofbusinessname: Line of business filter
            industryname: Industry name filter
            recordlimit: Maximum number of results
            page: Page number (default: 1)
            pagesize: Results per page (default: 10)
            
        Returns:
            Response containing POI array
        """
        params = {
            k: v for k, v in {
                "address": address,
                "zipcode": zipcode,
                "point": point,
                "radius": radius,
                "searchdistance": searchdistance,
                "categoryname": categoryname,
                "categoryid": categoryid,
                "lineofbusinessname": lineofbusinessname,
                "industryname": industryname,
                "recordlimit": recordlimit,
                "page": page,
                "pagesize": pagesize
            }.items() if v is not None
        }
        
        return self._call("get_poi_search", method="GET", **params)


def main():
    """
    Practical workflow example: Find properties in an area, retrieve detailed
    information including owner data, and identify nearby points of interest.
    """
    # Initialize the client
    client = ParseClient()
    
    print("=" * 80)
    print("ATTOM Data API - Property Research & Neighborhood Analysis Workflow")
    print("=" * 80)
    
    # Step 1: Search for properties by address
    print("\n[Step 1] Searching for properties in Beverly Hills...")
    search_response = client.get_property_address(
        address1="Beverly Hills",
        address2="CA",
        pagesize=3
    )
    
    if search_response.get("status", {}).get("code") != 0:
        print(f"Error: {search_response.get('status', {}).get('msg')}")
        return
    
    properties = search_response.get("property", [])
    total_found = search_response.get("status", {}).get("total", 0)
    print(f"\nFound {total_found} properties. Analyzing first {len(properties)}:\n")
    
    property_details = []
    
    # Step 2: Get detailed information and owner data for each property
    print("[Step 2] Retrieving detailed property and owner information...\n")
    
    for idx, prop in enumerate(properties, 1):
        prop_id = prop.get("identifier", {}).get("attomId")
        address_oneline = prop.get("address", {}).get("oneLine", "N/A")
        
        print(f"Property {idx}: {address_oneline}")
        
        if not prop_id:
            print("  (No ATTOM ID available)\n")
            continue
        
        # Get owner information
        try:
            owner_response = client.get_property_detail_owner(attomid=str(prop_id))
            
            if owner_response.get("status", {}).get("code") == 0:
                owner_data = owner_response.get("property", [{}])[0]
                owner_info = owner_data.get("owner", {}).get("owner1", {})
                owner_name = owner_info.get("fullname", "N/A")
                
                print(f"  ATTOM ID: {prop_id}")
                print(f"  Current Owner: {owner_name}")
                
                # Store for later analysis
                property_details.append({
                    "attomid": prop_id,
                    "address": address_oneline,
                    "owner": owner_name
                })
            else:
                print(f"  ATTOM ID: {prop_id}")
                print("  Owner Info: Unable to retrieve")
        except Exception as e:
            print(f"  ATTOM ID: {prop_id}")
            print(f"  Owner Info: Error - {str(e)[:60]}")
        
        print()
    
    # Step 3: Find points of interest in the area
    print("[Step 3] Searching for points of interest near the properties...\n")
    
    try:
        poi_response = client.get_poi_search(
            address="Beverly Hills, CA",
            searchdistance=1,
            recordlimit=10
        )
        
        if poi_response.get("status", {}).get("code") == 0:
            pois = poi_response.get("poi", [])
            total_pois = poi_response.get("status", {}).get("total", 0)
            
            if pois:
                print(f"Found {total_pois} points of interest. Showing first {len(pois)}:\n")
                print(f"{'Business Name':<40} {'Category':<30} {'Distance':<10}")
                print("-" * 80)
                
                for poi in pois[:5]:
                    biz_name = poi.get("businessLocation", {}).get("businessStandardName", "N/A")
                    category = poi.get("category", {}).get("condensedHeading", "N/A")
                    distance = poi.get("details", {}).get("distance", "N/A")
                    
                    # Truncate long names for display
                    biz_name_display = (biz_name[:39] + "...") if len(biz_name) > 40 else biz_name
                    category_display = (category[:29] + "...") if len(category) > 30 else category
                    
                    print(f"{biz_name_display:<40} {category_display:<30} {distance:<10}")
                
                print()
            else:
                print("No points of interest found in this area.\n")
        else:
            print("Unable to retrieve POI data.\n")
    except Exception as e:
        print(f"POI search failed: {str(e)[:80]}\n")
    
    # Step 4: Summary
    print("=" * 80)
    print("Summary of Found Properties:")
    print("=" * 80)
    
    if property_details:
        print(f"\n{'Address':<50} {'Owner':<25}")
        print("-" * 75)
        for prop in property_details:
            address_display = (prop["address"][:49] + "...") if len(prop["address"]) > 50 else prop["address"]
            owner_display = (prop["owner"][:24] + "...") if len(prop["owner"]) > 25 else prop["owner"]
            print(f"{address_display:<50} {owner_display:<25}")
    else:
        print("\nNo property details retrieved.")
    
    print("\n" + "=" * 80)
    print("Workflow complete!")
    print("=" * 80)


if __name__ == "__main__":
    main()