如何用Python调用Idealista API？求简便方式及地籍编码查租金

Hey there! Let's break down how to call the Idealista API with Python to pull rental prices for your cadastral codes, plus some shortcuts to make this smoother—especially since you're new to this workflow.

Idealista uses OAuth2 for authentication, so you'll need to fetch an access token using your API key and secret before making any search requests. This is a one-time step per session (tokens usually expire after 3600 seconds).

First, install the basics:

pip install requests python-dotenv

We'll use python-dotenv to keep your credentials safe (never hardcode keys in your code!). Create a .env file in your project folder with:

IDEALISTA_API_KEY=your_api_key_here
IDEALISTA_API_SECRET=your_api_secret_here

Here's the code to get your token:

import requests
import base64
import time
from dotenv import load_dotenv
import os

# Load credentials from .env
load_dotenv()
API_KEY = os.getenv("IDEALISTA_API_KEY")
API_SECRET = os.getenv("IDEALISTA_API_SECRET")

def get_access_token():
    # Encode credentials for Basic Auth
    credentials = f"{API_KEY}:{API_SECRET}"
    encoded_creds = base64.b64encode(credentials.encode()).decode()
    
    headers = {
        "Authorization": f"Basic {encoded_creds}",
        "Content-Type": "application/x-www-form-urlencoded"
    }
    payload = {"grant_type": "client_credentials"}
    
    response = requests.post("https://api.idealista.com/oauth/token", headers=headers, data=payload)
    response.raise_for_status()  # Throw an error if the request fails
    return response.json()["access_token"]

Next, we'll create a function to query the API for each cadastral code, filter for rental listings, and extract a meaningful price (we'll calculate an average if there are multiple listings for the same code).

Add this function:

def get_rental_price(cadastral_code, access_token):
    # Adjust the country code in the URL if needed (es = Spain, it = Italy, pt = Portugal)
    api_url = "https://api.idealista.com/3.5/es/search"
    
    headers = {"Authorization": f"Bearer {access_token}"}
    params = {
        "cadastralCode": cadastral_code,
        "operation": "rent",
        "propertyType": "homes",
        "maxItems": 10  # Limit to 10 results to get a representative average
    }
    
    response = requests.get(api_url, headers=headers, params=params)
    response.raise_for_status()
    results = response.json().get("elementList", [])
    
    if not results:
        return None  # No listings found for this code
    
    # Calculate average rental price (you could also pick the latest listing instead)
    rental_prices = [item["price"] for item in results]
    average_price = sum(rental_prices) / len(rental_prices)
    return round(average_price, 2)

Then put it all together to process your list:

def main():
    # Replace this with your actual cadastral code list
    cadastral_codes = ["COD12345", "COD67890", "COD09876"]
    
    access_token = get_access_token()
    rental_results = {}
    
    for code in cadastral_codes:
        print(f"Fetching data for cadastral code: {code}")
        try:
            price = get_rental_price(code, access_token)
            rental_results[code] = f"{price} €" if price else "No listings found"
            time.sleep(1)  # Respect Idealista's rate limits (adjust if needed)
        except Exception as e:
            rental_results[code] = f"Error: {str(e)}"
    
    # Print or save your results
    print("\n=== Rental Price Results ===")
    for code, result in rental_results.items():
        print(f"{code}: {result}")

if __name__ == "__main__":
    main()

If you want to make this even easier (and faster), here are a couple of tweaks:

1. Use a Session for Faster Requests #

Instead of creating a new connection for every request, use requests.Session() to reuse connections. This cuts down on overhead, especially if you have a long list of codes:

def get_access_token(session):
    credentials = f"{API_KEY}:{API_SECRET}"
    encoded_creds = base64.b64encode(credentials.encode()).decode()
    
    headers = {
        "Authorization": f"Basic {encoded_creds}",
        "Content-Type": "application/x-www-form-urlencoded"
    }
    payload = {"grant_type": "client_credentials"}
    
    response = session.post("https://api.idealista.com/oauth/token", headers=headers, data=payload)
    response.raise_for_status()
    return response.json()["access_token"]

def get_rental_price(cadastral_code, access_token, session):
    api_url = "https://api.idealista.com/3.5/es/search"
    
    headers = {"Authorization": f"Bearer {access_token}"}
    params = {
        "cadastralCode": cadastral_code,
        "operation": "rent",
        "propertyType": "homes",
        "maxItems": 10
    }
    
    response = session.get(api_url, headers=headers, params=params)
    response.raise_for_status()
    results = response.json().get("elementList", [])
    
    if not results:
        return None
    
    rental_prices = [item["price"] for item in results]
    average_price = sum(rental_prices) / len(rental_prices)
    return round(average_price, 2)

def main():
    cadastral_codes = ["COD12345", "COD67890"]
    
    with requests.Session() as session:
        access_token = get_access_token(session)
        rental_results = {}
        
        for code in cadastral_codes:
            print(f"Fetching data for cadastral code: {code}")
            try:
                price = get_rental_price(code, access_token, session)
                rental_results[code] = f"{price} €" if price else "No listings found"
                time.sleep(1)
            except Exception as e:
                rental_results[code] = f"Error: {str(e)}"
    
    print("\n=== Rental Price Results ===")
    for code, result in rental_results.items():
        print(f"{code}: {result}")

if __name__ == "__main__":
    main()

2. Async Requests for Large Lists #

If you have hundreds of cadastral codes, using async requests with aiohttp will speed things up significantly (you can make multiple requests at once instead of waiting for each one to finish):

pip install aiohttp

Here's a quick async version snippet:

import aiohttp
import asyncio

async def get_rental_price_async(cadastral_code, access_token, session):
    api_url = "https://api.idealista.com/3.5/es/search"
    headers = {"Authorization": f"Bearer {access_token}"}
    params = {
        "cadastralCode": cadastral_code,
        "operation": "rent",
        "propertyType": "homes",
        "maxItems": 10
    }
    
    async with session.get(api_url, headers=headers, params=params) as response:
        response.raise_for_status()
        results = await response.json()
        if not results.get("elementList"):
            return None
        prices = [item["price"] for item in results["elementList"]]
        return round(sum(prices)/len(prices), 2)

async def main_async():
    cadastral_codes = ["COD12345", "COD67890"]
    access_token = get_access_token()
    
    async with aiohttp.ClientSession() as session:
        tasks = [get_rental_price_async(code, access_token, session) for code in cadastral_codes]
        # Run all tasks, return exceptions instead of crashing
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        # Map results to your cadastral codes
        rental_results = dict(zip(cadastral_codes, results))
        print("\n=== Rental Price Results ===")
        for code, result in rental_results.items():
            if isinstance(result, Exception):
                print(f"{code}: Error: {str(result)}")
            elif result is None:
                print(f"{code}: No listings found")
            else:
                print(f"{code}: {result} €")

if __name__ == "__main__":
    asyncio.run(main_async())

• Rate Limits: Idealista has strict rate limits (check their docs for details). The time.sleep(1) in the sync version helps avoid hitting them—adjust if you get 429 errors.
• Country Codes: Make sure the URL's country code matches your target region (e.g., it for Italy).
• Error Handling: The code uses response.raise_for_status() to catch bad requests, but you can expand this to handle specific errors like expired tokens.

内容的提问来源于stack exchange，提问作者meral