Integration GuidePython / Flask

Python/Flask Integration Guide

Build a Flask proxy for the Dubai Real Estate Data API with authentication, caching, background tasks, and production-ready patterns.

API Reference Get API Key

Overview

The Dubai Real Estate Data API provides comprehensive access to Dubai's property market data, sourced directly from the Dubai Land Department (DLD). With this integration guide, you can build powerful property search, analytics, and listing applications using Python and Flask.

1M+

Transactions

AED 4.37T

Total Value

356

Areas

3,517

Projects

API Base URL

https://api.buyorsell24.com

Getting an API Key

All API requests require a Bearer token for authentication. Follow these steps to obtain your key:

1
Visit the Contact Page
Go to dynamicweblab.com/contact and fill out the form requesting API access.
2
Receive Your Token
You will receive a Bearer token via email within 24 hours.
3
Store It Securely
Add your token to environment variables using python-dotenv. Never commit API keys to version control.

Flask Project Setup

Create a Virtual Environment

python -m venv venv
source venv/bin/activate  # macOS / Linux
venv\Scripts\activate     # Windows

Install Dependencies

pip install flask requests python-dotenv flask-caching

requirements.txt

flask>=3.0
requests>=2.31
python-dotenv>=1.0
flask-caching>=2.1
celery[redis]>=5.3
pytest>=8.0

Environment Variables

Create a .env file in your project root:

# .env
FLASK_APP=app
FLASK_ENV=development
API_BASE_URL=https://api.buyorsell24.com
API_KEY=your_bearer_token_here
REDIS_URL=redis://localhost:6379/0

HTTP Client Configuration

Use a requests.Session for connection pooling and automatic header injection. This avoids recreating TCP connections on every request.

Create a Reusable Session

File: app/services/real_estate_api.py

import os
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry


class RealEstateAPI:
    """Client for the Dubai Real Estate Data API."""

    def __init__(self):
        self.base_url = os.environ["API_BASE_URL"]
        self.api_key = os.environ["API_KEY"]

        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
            "Accept": "application/json",
        })

        retry_strategy = Retry(
            total=3,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=["GET"],
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        self.session.mount("https://", adapter)

    def get(self, endpoint: str, params: dict | None = None) -> dict:
        """Make a GET request to the API."""
        url = f"{self.base_url}{endpoint}"
        response = self.session.get(url, params=params)
        response.raise_for_status()
        return response.json()

    def post(self, endpoint: str, json: dict | None = None) -> dict:
        """Make a POST request to the API."""
        url = f"{self.base_url}{endpoint}"
        response = self.session.post(url, json=json)
        response.raise_for_status()
        return response.json()

    def transactions(self, **kwargs) -> dict:
        return self.get("/api/v1/transactions", params=kwargs)

    def areas(self, **kwargs) -> dict:
        return self.get("/api/v1/areas", params=kwargs)

    def projects(self, **kwargs) -> dict:
        return self.get("/api/v1/projects", params=kwargs)

    def developers(self, **kwargs) -> dict:
        return self.get("/api/v1/developers", params=kwargs)


api_client = RealEstateAPI()

Flask Application Factory

app/init.py

import os
from flask import Flask
from flask_caching import Cache

cache = Cache()


def create_app() -> Flask:
    app = Flask(__name__)

    app.config["CACHE_TYPE"] = "RedisCache"
    app.config["CACHE_REDIS_URL"] = os.environ.get(
        "REDIS_URL", "redis://localhost:6379/0"
    )
    app.config["CACHE_DEFAULT_TIMEOUT"] = 3600

    cache.init_app(app)

    from app.routes.transactions import transactions_bp
    from app.routes.areas import areas_bp
    from app.routes.projects import projects_bp
    from app.routes.developers import developers_bp

    app.register_blueprint(transactions_bp, url_prefix="/proxy/transactions")
    app.register_blueprint(areas_bp, url_prefix="/proxy/areas")
    app.register_blueprint(projects_bp, url_prefix="/proxy/projects")
    app.register_blueprint(developers_bp, url_prefix="/proxy/developers")

    return app

Flask Route Definitions

Each Blueprint proxies a specific API endpoint. The API key stays server-side — your frontend only calls your own Flask routes.

GETTransactions Blueprint

File: app/routes/transactions.py

from flask import Blueprint, request, jsonify
from app import cache
from app.services.real_estate_api import api_client

transactions_bp = Blueprint("transactions", __name__)


@transactions_bp.route("/", methods=["GET"])
@cache.cached(timeout=3600, query_string=True)
def get_transactions():
    params = {
        "type": request.args.get("type"),
        "area": request.args.get("area"),
        "min_price": request.args.get("min_price"),
        "max_price": request.args.get("max_price"),
        "from_date": request.args.get("from_date"),
        "to_date": request.args.get("to_date"),
        "limit": request.args.get("limit", 20),
        "cursor": request.args.get("cursor"),
    }
    params = {k: v for k, v in params.items() if v is not None}

    data = api_client.transactions(**params)
    return jsonify(data)


@transactions_bp.route("/<transaction_id>", methods=["GET"])
@cache.cached(timeout=3600)
def get_transaction(transaction_id: str):
    data = api_client.get(f"/api/v1/transactions/{transaction_id}")
    return jsonify(data)

GETAreas Blueprint

File: app/routes/areas.py

from flask import Blueprint, jsonify
from app import cache
from app.services.real_estate_api import api_client

areas_bp = Blueprint("areas", __name__)


@areas_bp.route("/", methods=["GET"])
@cache.cached(timeout=86400)
def get_areas():
    data = api_client.areas()
    return jsonify(data)


@areas_bp.route("/<int:area_id>", methods=["GET"])
@cache.cached(timeout=86400)
def get_area(area_id: int):
    data = api_client.get(f"/api/v1/areas/{area_id}")
    return jsonify(data)


@areas_bp.route("/<int:area_id>/stats", methods=["GET"])
@cache.cached(timeout=86400)
def get_area_stats(area_id: int):
    data = api_client.get(f"/api/v1/areas/{area_id}/stats")
    return jsonify(data)

GETProjects Blueprint

File: app/routes/projects.py

from flask import Blueprint, request, jsonify
from app import cache
from app.services.real_estate_api import api_client

projects_bp = Blueprint("projects", __name__)


@projects_bp.route("/", methods=["GET"])
@cache.cached(timeout=21600, query_string=True)
def get_projects():
    params = {
        "developer": request.args.get("developer"),
        "area": request.args.get("area"),
        "limit": request.args.get("limit", 20),
        "cursor": request.args.get("cursor"),
    }
    params = {k: v for k, v in params.items() if v is not None}

    data = api_client.projects(**params)
    return jsonify(data)


@projects_bp.route("/<int:project_id>", methods=["GET"])
@cache.cached(timeout=21600)
def get_project(project_id: int):
    data = api_client.get(f"/api/v1/projects/{project_id}")
    return jsonify(data)

GETDevelopers Blueprint

File: app/routes/developers.py

from flask import Blueprint, jsonify
from app import cache
from app.services.real_estate_api import api_client

developers_bp = Blueprint("developers", __name__)


@developers_bp.route("/", methods=["GET"])
@cache.cached(timeout=86400)
def get_developers():
    data = api_client.developers()
    return jsonify(data)


@developers_bp.route("/<int:developer_id>", methods=["GET"])
@cache.cached(timeout=86400)
def get_developer(developer_id: int):
    data = api_client.get(f"/api/v1/developers/{developer_id}")
    return jsonify(data)

Authentication

All requests to the API must include a Bearer token in the Authorization header. Your API key should never be exposed to the client — always proxy requests through Flask.

Do

✓Store API keys in environment variables
✓Use Flask as a server-side proxy
✓Use python-dotenv for local development
✓Add .env to .gitignore

Don't

✗Hardcode API keys in source files
✗Return raw API responses with embedded credentials
✗Expose the upstream API URL to the frontend
✗Commit environment files to Git

Common Endpoints

GET/api/v1/transactions

Property transactions from the Dubai Land Department. Filter by type, area, price range, and date.

resp = api_client.transactions(
    type="sale",
    area="Palm Jumeirah",
    min_price=1000000,
    limit=10,
)
print(resp["transactions"])

GET/api/v1/areas

All 356 Dubai areas with building counts, stats, and project data.

resp = api_client.areas()
for area in resp["areas"]:
    print(f"{area['name']}: {area['building_count']} buildings")

GET/api/v1/projects

3,517 projects with stats, buildings, transactions, and price trends.

resp = api_client.projects(developer="Emaar")
for project in resp["projects"]:
    print(f"{project['name']} — {project['area']}")

GET/api/v1/developers

Developer directory with AI-generated profiles, projects, and stats.

resp = api_client.developers()
for dev in resp["developers"]:
    print(f"{dev['name']} — {dev['project_count']} projects")

Error Handling

app/errors.py

import logging
import requests
from flask import jsonify
from app import create_app

logger = logging.getLogger(__name__)


def register_error_handlers(app):
    @app.errorhandler(requests.exceptions.HTTPError)
    def handle_http_error(error):
        response = error.response
        status = response.status_code

        messages = {
            401: "Invalid API key. Check your environment variables.",
            403: "Insufficient permissions for this endpoint.",
            404: "Resource not found.",
            429: "Rate limit exceeded. Please wait before retrying.",
            500: "Upstream server error. Please try again later.",
        }

        return jsonify({
            "error": True,
            "status": status,
            "message": messages.get(status, f"API error: {status}"),
        }), status

    @app.errorhandler(requests.exceptions.ConnectionError)
    def handle_connection_error(error):
        logger.error("API connection failed: %s", error)
        return jsonify({
            "error": True,
            "status": 503,
            "message": "Service temporarily unavailable. Please try again later.",
        }), 503

    @app.errorhandler(requests.exceptions.Timeout)
    def handle_timeout(error):
        logger.error("API request timed out: %s", error)
        return jsonify({
            "error": True,
            "status": 504,
            "message": "Upstream request timed out.",
        }), 504

    @app.errorhandler(Exception)
    def handle_generic_error(error):
        logger.exception("Unexpected error: %s", error)
        return jsonify({
            "error": True,
            "status": 500,
            "message": "An unexpected error occurred.",
        }), 500

app/init.py (updated)

def create_app() -> Flask:
    app = Flask(__name__)
    # ... config and blueprint registration ...

    from app.errors import register_error_handlers
    register_error_handlers(app)

    return app

Rate Limits

The API enforces rate limits per API key. The default Startup Lite tier allows 100 requests per minute. Implement retry logic with exponential backoff using urllib3.Retry.

Rate Limit Tiers

Startup Lite

100

requests / minute

Startup Growth

300

requests / minute

Enterprise

Custom

unlimited

Retry with Exponential Backoff

Already configured in the RealEstateAPI client above via HTTPAdapter and urllib3.Retry. For additional control:

import time
import requests


def fetch_with_backoff(func, *args, max_retries=3, **kwargs):
    """Execute an API call with exponential backoff on 429 errors."""
    for attempt in range(max_retries):
        try:
            response = func(*args, **kwargs)
            if response.status_code == 429:
                retry_after = int(response.headers.get("Retry-After", 2 ** attempt))
                time.sleep(retry_after)
                continue
            response.raise_for_status()
            return response.json()
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 429 and attempt < max_retries - 1:
                time.sleep(2 ** attempt)
                continue
            raise
    raise Exception("Max retries exceeded")


# Usage
data = fetch_with_backoff(
    api_client.session.get,
    f"{api_client.base_url}/api/v1/transactions",
    params={"limit": 10},
)

Pagination

Use the next_cursor field to paginate through large result sets. Pass the cursor as a query parameter to fetch the next page.

Cursor-Based Pagination

def fetch_all_transactions(max_pages: int = 10):
    """Fetch multiple pages of transactions."""
    all_transactions = []
    cursor = None

    for _ in range(max_pages):
        params = {"limit": 100}
        if cursor:
            params["cursor"] = cursor

        data = api_client.transactions(**params)
        all_transactions.extend(data["transactions"])

        cursor = data.get("next_cursor")
        if not cursor:
            break

    return all_transactions


# Fetch all available pages
transactions = fetch_all_transactions()
print(f"Fetched {len(transactions)} transactions")

Caching with Flask-Caching

Use Flask-Caching to reduce API calls and improve response times. Cache query results in Redis keyed by query parameters.

Route-Level Caching

from app import cache
from flask import Blueprint, request, jsonify
from app.services.real_estate_api import api_client

analytics_bp = Blueprint("analytics", __name__)


@analytics_bp.route("/rental-by-area", methods=["GET"])
@cache.cached(timeout=86400, query_string=True)
def rental_by_area():
    data = api_client.get("/api/v1/analytics/rental-by-area")
    return jsonify(data)


@analytics_bp.route("/yield-by-building", methods=["GET"])
@cache.cached(timeout=3600, query_string=True)
def yield_by_building():
    data = api_client.get("/api/v1/analytics/yield-by-building")
    return jsonify(data)

Manual Cache Control

from app import cache


def get_transactions_cached(area: str, **kwargs):
    cache_key = f"transactions:{area}:{hash(frozenset(kwargs.items()))}"

    result = cache.get(cache_key)
    if result is None:
        result = api_client.transactions(area=area, **kwargs)
        cache.set(cache_key, result, timeout=3600)

    return result


# Invalidate specific cache
def invalidate_area_cache(area: str):
    cache.delete_memoized(get_transactions_cached, area)

Background Tasks with Celery

Use Celery to offload heavy data processing, scheduled syncs, and long-running queries to background workers.

app/tasks/celery_app.py

import os
from celery import Celery

celery_app = Celery(
    "dubai_api",
    broker=os.environ.get("REDIS_URL", "redis://localhost:6379/0"),
    backend=os.environ.get("REDIS_URL", "redis://localhost:6379/0"),
)

celery_app.conf.update(
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    timezone="Asia/Dubai",
    enable_utc=True,
    task_track_started=True,
    task_time_limit=300,
)

app/tasks/sync.py

import logging
from app.tasks.celery_app import celery_app
from app.services.real_estate_api import api_client
from app import cache

logger = logging.getLogger(__name__)


@celery_app.task(bind=True, max_retries=3)
def sync_transactions(self):
    """Periodically sync recent transactions and warm the cache."""
    try:
        data = api_client.transactions(limit=100)
        transactions = data.get("transactions", [])
        logger.info("Synced %d transactions", len(transactions))

        cache.set("latest_transactions", transactions, timeout=3600)
        return {"synced": len(transactions)}
    except Exception as exc:
        logger.error("Transaction sync failed: %s", exc)
        self.retry(exc=exc, countdown=60)


@celery_app.task
def sync_areas():
    """Refresh cached area data (rarely changes)."""
    data = api_client.areas()
    cache.set("all_areas", data, timeout=86400)
    return {"synced": len(data.get("areas", []))}


celery_app.conf.beat_schedule = {
    "sync-transactions-hourly": {
        "task": "app.tasks.sync.sync_transactions",
        "schedule": 3600.0,
    },
    "sync-areas-daily": {
        "task": "app.tasks.sync.sync_areas",
        "schedule": 86400.0,
    },
}

Run Workers

# Start a worker
celery -A app.tasks.celery_app worker --loglevel=info

# Start the beat scheduler (in a separate terminal)
celery -A app.tasks.celery_app beat --loglevel=info

Jinja2 Template Integration

Render API data directly in server-side templates. This is useful for admin dashboards, internal tools, or SEO-friendly server-rendered pages.

app/routes/pages.py

from flask import Blueprint, render_template
from app.services.real_estate_api import api_client
from app import cache

pages_bp = Blueprint("pages", __name__)


@pages_bp.route("/dashboard")
@cache.cached(timeout=3600)
def dashboard():
    transactions = api_client.transactions(limit=20)
    areas = api_client.areas()
    return render_template(
        "dashboard.html",
        transactions=transactions["transactions"],
        areas=areas["areas"],
        title="Dubai Real Estate Dashboard",
    )


@pages_bp.route("/area/<int:area_id>")
def area_detail(area_id: int):
    area = api_client.get(f"/api/v1/areas/{area_id}")
    stats = api_client.get(f"/api/v1/areas/{area_id}/stats")
    return render_template(
        "area_detail.html",
        area=area,
        stats=stats,
    )

templates/dashboard.html

{% extends "base.html" %}
{% block content %}
<h1>{{ title }}</h1>

<div class="grid grid-cols-3 gap-4 mb-8">
  <div class="stat-card">
    <h3>{{ transactions|length }}</h3>
    <p>Recent Transactions</p>
  </div>
  <div class="stat-card">
    <h3>{{ areas|length }}</h3>
    <p>Dubai Areas</p>
  </div>
</div>

<table>
  <thead>
    <tr>
      <th>Property</th>
      <th>Price (AED)</th>
      <th>Area (sqft)</th>
      <th>Date</th>
    </tr>
  </thead>
  <tbody>
    {% for t in transactions %}
    <tr>
      <td>{{ t.property_name }}</td>
      <td>{{ "{:,.0f}".format(t.price_aed) }}</td>
      <td>{{ t.area_sqft }}</td>
      <td>{{ t.transaction_date }}</td>
    </tr>
    {% endfor %}
  </tbody>
</table>
{% endblock %}

Testing with pytest

Use pytest and Flask's test client to verify your proxy routes without hitting the real API.

tests/conftest.py

import pytest
from unittest.mock import patch, MagicMock
from app import create_app


@pytest.fixture
def app():
    app = create_app()
    app.config["TESTING"] = True
    app.config["CACHE_TYPE"] = "SimpleCache"
    return app


@pytest.fixture
def client(app):
    return app.test_client()


@pytest.fixture
def mock_api():
    with patch("app.services.real_estate_api.api_client") as client:
        yield client

tests/test_transactions.py

def test_get_transactions(client, mock_api):
    mock_api.transactions.return_value = {
        "transactions": [
            {
                "id": "tx-001",
                "property_name": "Palm Tower A",
                "price_aed": 2500000,
                "area_sqft": 1800,
                "transaction_date": "2026-01-15",
            }
        ],
        "next_cursor": None,
    }

    response = client.get("/proxy/transactions/?type=sale&limit=10")

    assert response.status_code == 200
    data = response.get_json()
    assert len(data["transactions"]) == 1
    assert data["transactions"][0]["price_aed"] == 2500000
    mock_api.transactions.assert_called_once_with(type="sale", limit="10")


def test_get_transactions_error(client, mock_api):
    from requests.exceptions import HTTPError

    mock_response = MagicMock()
    mock_response.status_code = 401
    mock_response.raise_for_status.side_effect = HTTPError(response=mock_response)
    mock_api.transactions.side_effect = HTTPError(response=mock_response)

    response = client.get("/proxy/transactions/")

    assert response.status_code == 401


def test_get_areas(client, mock_api):
    mock_api.areas.return_value = {
        "areas": [
            {"id": 1, "name": "Palm Jumeirah", "building_count": 120},
            {"id": 2, "name": "Downtown Dubai", "building_count": 85},
        ]
    }

    response = client.get("/proxy/areas/")

    assert response.status_code == 200
    data = response.get_json()
    assert len(data["areas"]) == 2


def test_get_projects(client, mock_api):
    mock_api.projects.return_value = {
        "projects": [
            {"id": 101, "name": "Burj Vista", "developer": "Emaar"},
        ],
        "next_cursor": None,
    }

    response = client.get("/proxy/projects/?developer=Emaar")

    assert response.status_code == 200
    data = response.get_json()
    assert data["projects"][0]["developer"] == "Emaar"

Run Tests

pytest tests/ -v
pytest tests/ -v --cov=app --cov-report=term-missing

Recommended Project Structure

dubai-property-api/
├── app/
│   ├── __init__.py          # Application factory
│   ├── errors.py            # Error handlers
│   ├── routes/
│   │   ├── __init__.py
│   │   ├── transactions.py  # Transactions Blueprint
│   │   ├── areas.py         # Areas Blueprint
│   │   ├── projects.py      # Projects Blueprint
│   │   ├── developers.py    # Developers Blueprint
│   │   └── pages.py         # Jinja2 page routes
│   ├── services/
│   │   ├── __init__.py
│   │   └── real_estate_api.py  # API client class
│   ├── tasks/
│   │   ├── __init__.py
│   │   ├── celery_app.py    # Celery configuration
│   │   └── sync.py          # Background sync tasks
│   └── templates/
│       ├── base.html
│       └── dashboard.html
├── tests/
│   ├── conftest.py
│   ├── test_transactions.py
│   └── test_areas.py
├── .env
├── .gitignore
├── requirements.txt
├── run.py                   # Entry point
└── celery_worker.py         # Celery worker entry

Ready to Build?

Get your API key and start integrating Dubai's most comprehensive real estate data into your Flask application.

Request API Access View OpenAPI Spec