Integration Guide

Python backend integration: FastAPI & Django

Call KhaleejiAPI from your Python server and expose only the specific operations your signed-in users need. The API key stays in a server secret; the browser talks only to your application.

1. Create a key and authenticate

  1. Create an API key in your KhaleejiAPI dashboard and grant only the scopes required by your server routes.
  2. Store it as KHALEEJI_API_KEY in your local environment file (ignored by Git) or production secret manager.
  3. Send the key on each server-to-server API call in Authorization: Bearer <key>.
  4. Never accept an upstream API key from the browser, forward browser Authorization headers, or include the key in an error response.

Required scopes for the examples below: validation:emirates-id, validation:saudi-id, validation:vat, finance:zakat, finance:sukuk, and finance:takaful. Grant only what the app actually uses.

bash
# Set this in your local environment or secret manager, never in browser code.
export KHALEEJI_API_KEY="your_api_key_here"
export KHALEEJI_API_BASE_URL="https://khaleejiapi.dev/api/v1"
 
# Verify the key and the validation:emirates-id scope with a synthetic test ID.
curl --get "$KHALEEJI_API_BASE_URL/validation/emirates-id" \
  --header "$(printf '%s: %s %s' Authorization Bearer "$KHALEEJI_API_KEY")" \
  --data-urlencode "id=$SYNTHETIC_EMIRATES_ID"

2. FastAPI proxy

Install FastAPI and the async HTTP client. The app creates one client for connection pooling, uses fixed upstream paths, enforces an application-auth dependency, and forwards safe rate-limit headers. Connect require_authenticated_user to your own login system and per-user rate limiter before exposing these routes.

bash
python -m pip install fastapi uvicorn httpx
python
# main.py
import logging
import os
from contextlib import asynccontextmanager
from typing import Literal
 
import httpx
from fastapi import Depends, FastAPI, HTTPException, Query, Request, Response
from pydantic import BaseModel
from starlette.middleware.cors import CORSMiddleware
 
from app.auth import require_authenticated_user  # Use your application's auth dependency
 
logger = logging.getLogger(__name__)
API_KEY = os.getenv("KHALEEJI_API_KEY")
API_BASE = os.getenv("KHALEEJI_API_BASE_URL", "https://khaleejiapi.dev/api/v1")
RATE_LIMIT_HEADERS = ("Retry-After", "X-RateLimit-Limit", "X-RateLimit-Remaining", "X-RateLimit-Reset")
 
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.khaleeji = httpx.AsyncClient(timeout=httpx.Timeout(10.0))
    yield
    await app.state.khaleeji.aclose()
 
 
app = FastAPI(lifespan=lifespan)
 
# Only needed when your frontend is on a different origin. Use exact trusted origins.
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://app.example.com"],
    allow_credentials=True,
    allow_methods=["GET", "POST"],
    allow_headers=["Content-Type", "X-CSRFToken"],
    expose_headers=["Retry-After", "X-RateLimit-Limit", "X-RateLimit-Remaining", "X-RateLimit-Reset"],
)
 
 
async def forward(request: Request, method: str, endpoint: str, *, params=None, payload=None) -> Response:
    if not API_KEY:
        raise HTTPException(status_code=500, detail="KhaleejiAPI is not configured")
 
    try:
        upstream = await request.app.state.khaleeji.request(
            method,
            f"{API_BASE}/{endpoint}",
            headers={"Authorization": "Bearer " + API_KEY, "Accept": "application/json"},
            params=params,
            json=payload,
        )
    except httpx.TimeoutException as exc:
        raise HTTPException(status_code=504, detail="KhaleejiAPI request timed out") from exc
    except httpx.RequestError as exc:
        raise HTTPException(status_code=502, detail="KhaleejiAPI is temporarily unavailable") from exc
 
    if upstream.status_code in (401, 403):
        logger.error("KhaleejiAPI rejected the server credential (status=%s)", upstream.status_code)
        raise HTTPException(status_code=502, detail="Check the server API key and its endpoint scopes")
    if upstream.status_code >= 500:
        logger.warning("KhaleejiAPI returned an upstream error (status=%s)", upstream.status_code)
        raise HTTPException(status_code=502, detail="KhaleejiAPI is temporarily unavailable")
 
    headers = {"Content-Type": upstream.headers.get("Content-Type", "application/json")}
    headers.update({name: upstream.headers[name] for name in RATE_LIMIT_HEADERS if name in upstream.headers})
    return Response(content=upstream.content, status_code=upstream.status_code, headers=headers)
 
 
class VATRequest(BaseModel):
    tin: str
    country: Literal["AE", "SA", "BH", "OM"]
 
 
class ZakatRequest(BaseModel):
    currency: str = "AED"
    assets: list[dict[str, object]]
 
 
@app.get("/api/khaleeji/emirates-id", dependencies=[Depends(require_authenticated_user)])
async def validate_emirates_id(request: Request, id: str):
    return await forward(request, "GET", "validation/emirates-id", params={"id": id})
 
 
@app.get("/api/khaleeji/saudi-id", dependencies=[Depends(require_authenticated_user)])
async def validate_saudi_id(request: Request, id: str):
    return await forward(request, "GET", "validation/saudi-id", params={"id": id})
 
 
@app.post("/api/khaleeji/vat", dependencies=[Depends(require_authenticated_user)])
async def validate_vat(request: Request, payload: VATRequest):
    return await forward(request, "POST", "vat/validate", payload=payload.model_dump())
 
 
@app.post("/api/khaleeji/zakat", dependencies=[Depends(require_authenticated_user)])
async def calculate_zakat(request: Request, payload: ZakatRequest):
    return await forward(request, "POST", "zakat/calculate", payload=payload.model_dump())
 
 
@app.get("/api/khaleeji/sukuk", dependencies=[Depends(require_authenticated_user)])
async def list_sukuk(request: Request, country: str | None = None):
    params = {"country": country} if country else None
    return await forward(request, "GET", "sukuk/list", params=params)
 
 
@app.get("/api/khaleeji/takaful", dependencies=[Depends(require_authenticated_user)])
async def estimate_takaful(
    request: Request,
    takaful_type: str = Query(alias="type"),
    coverage: float = Query(gt=0),
    currency: str = "AED",
    country: str = "AE",
    model: str = "wakalah_mudarabah",
):
    params = {
        "type": takaful_type,
        "coverage": coverage,
        "currency": currency,
        "country": country,
        "model": model,
    }
    return await forward(request, "GET", "takaful/calculate", params=params)

3. Django proxy

The views below use Django's authenticated session and authentication middleware. Keep CsrfViewMiddleware enabled for POST requests; do not add csrf_exempt to these browser-facing endpoints. Also apply your app's per-user throttling and deployment-specific authorization rules.

bash
python -m pip install django requests
python
# views.py
import json
import logging
import os
 
import requests
from django.http import HttpResponse, JsonResponse
from django.views.decorators.http import require_GET, require_POST
 
logger = logging.getLogger(__name__)
API_BASE = os.getenv("KHALEEJI_API_BASE_URL", "https://khaleejiapi.dev/api/v1")
RATE_LIMIT_HEADERS = ("Retry-After", "X-RateLimit-Limit", "X-RateLimit-Remaining", "X-RateLimit-Reset")
 
 
def forward(request, method, endpoint, *, params=None, payload=None):
    if not request.user.is_authenticated:
        return JsonResponse({"error": {"code": "UNAUTHENTICATED", "message": "Sign in first"}}, status=401)
 
    api_key = os.getenv("KHALEEJI_API_KEY")
    if not api_key:
        return JsonResponse({"error": {"code": "PROXY_NOT_CONFIGURED", "message": "Service is not configured"}}, status=500)
 
    try:
        upstream = requests.request(
            method,
            f"{API_BASE}/{endpoint}",
            headers={"Authorization": "Bearer " + api_key, "Accept": "application/json"},
            params=params,
            json=payload,
            timeout=(3.05, 10),
            allow_redirects=False,
        )
    except requests.Timeout:
        return JsonResponse({"error": {"code": "UPSTREAM_TIMEOUT", "message": "Request timed out"}}, status=504)
    except requests.RequestException:
        logger.error("Could not reach KhaleejiAPI")
        return JsonResponse({"error": {"code": "UPSTREAM_UNAVAILABLE", "message": "Service is temporarily unavailable"}}, status=502)
 
    if upstream.status_code in (401, 403):
        logger.error("KhaleejiAPI rejected the server credential (status=%s)", upstream.status_code)
        return JsonResponse({"error": {"code": "UPSTREAM_AUTH_CONFIGURATION", "message": "Check the server API key and its endpoint scopes"}}, status=502)
    if upstream.status_code >= 500:
        logger.warning("KhaleejiAPI returned an upstream error (status=%s)", upstream.status_code)
        return JsonResponse({"error": {"code": "UPSTREAM_UNAVAILABLE", "message": "Service is temporarily unavailable"}}, status=502)
 
    response = HttpResponse(
        upstream.content,
        status=upstream.status_code,
        content_type=upstream.headers.get("Content-Type", "application/json"),
    )
    for name in RATE_LIMIT_HEADERS:
        if name in upstream.headers:
            response[name] = upstream.headers[name]
    return response
 
 
def read_json_body(request):
    try:
        payload = json.loads(request.body)
    except (UnicodeDecodeError, json.JSONDecodeError):
        return None
    return payload if isinstance(payload, dict) else None
 
 
@require_GET
def validate_emirates_id(request):
    return forward(request, "GET", "validation/emirates-id", params={"id": request.GET.get("id", "")})
 
 
@require_GET
def validate_saudi_id(request):
    return forward(request, "GET", "validation/saudi-id", params={"id": request.GET.get("id", "")})
 
 
@require_POST
def validate_vat(request):
    payload = read_json_body(request)
    if payload is None:
        return JsonResponse({"error": {"code": "INVALID_JSON", "message": "Expected a JSON object"}}, status=400)
    return forward(request, "POST", "vat/validate", payload=payload)
 
 
@require_POST
def calculate_zakat(request):
    payload = read_json_body(request)
    if payload is None:
        return JsonResponse({"error": {"code": "INVALID_JSON", "message": "Expected a JSON object"}}, status=400)
    return forward(request, "POST", "zakat/calculate", payload=payload)
 
 
@require_GET
def list_sukuk(request):
    params = {"country": request.GET["country"]} if request.GET.get("country") else None
    return forward(request, "GET", "sukuk/list", params=params)
 
 
@require_GET
def estimate_takaful(request):
    allowed = ("type", "coverage", "currency", "country", "model")
    params = {name: request.GET[name] for name in allowed if name in request.GET}
    return forward(request, "GET", "takaful/calculate", params=params)

Add these explicit paths in your app's URL configuration; avoid forwarding arbitrary caller-supplied paths.

python
# urls.py — use explicit routes; do not expose a caller-controlled catch-all proxy.
from django.urls import path
from . import views
 
urlpatterns = [
    path("api/khaleeji/emirates-id", views.validate_emirates_id),
    path("api/khaleeji/saudi-id", views.validate_saudi_id),
    path("api/khaleeji/vat", views.validate_vat),
    path("api/khaleeji/zakat", views.calculate_zakat),
    path("api/khaleeji/sukuk", views.list_sukuk),
    path("api/khaleeji/takaful", views.estimate_takaful),
]

4. Call your backend and handle errors

The browser calls your application's same-origin route—not KhaleejiAPI directly—and sends only its normal app session and framework CSRF token. Upstream 400 responses and 429 rate-limit responses retain their response body and status; only the selected rate-limit headers are copied.

javascript
// Browser code calls your authenticated application, never KhaleejiAPI directly.
const response = await fetch("/api/khaleeji/vat", {
  method: "POST",
  credentials: "same-origin",
  headers: {
    "Content-Type": "application/json",
    "X-CSRFToken": csrfToken, // Include your framework's CSRF token for session-authenticated POSTs.
  },
  body: JSON.stringify({ tin: "your-test-number", country: "AE" }),
})
 
if (response.status === 429) {
  const retryAfter = response.headers.get("Retry-After")
  // Show a retry message; do not retry in a tight loop.
}
const result = await response.json()
Upstream statusMeaning and actionProxy response
401 UnauthorizedThe server key is missing, invalid, or not being sent as a ****** Check the backend secret and authorization header; do not retry unchanged.Map to a sanitized 502; log the status only, never the key.
403 ForbiddenThe key does not have the scope required by that endpoint. Add only the required scope in the dashboard.Map to the same sanitized 502 configuration error.
429 Too Many RequestsThe key reached a rate or plan limit. Respect Retry-After; back off with jitter and avoid tight retry loops.Preserve status, API error body, and available rate-limit headers.
5xx or timeoutKhaleejiAPI or the network is temporarily unavailable. Retry only when appropriate for the operation.Return a sanitized 502 or 504; do not expose upstream internals.

5. Endpoint examples and request shapes

These examples use the implemented service-first API paths. The validation endpoints accept an id query parameter; VAT and Zakat accept JSON POST bodies; Sukuk and Takaful use GET query parameters.

Use caseKhaleejiAPI requestExample input
Emirates IDGET /api/v1/validation/emirates-id?id=...id=784-…
Saudi National ID / IqamaGET /api/v1/validation/saudi-id?id=...id=1XXXXXXXXX or 2XXXXXXXXX
GCC VATPOST /api/v1/vat/validate{"tin":"…","country":"AE"}
ZakatPOST /api/v1/zakat/calculate{"currency":"AED","assets":[{"type":"cash","value":50000}]}
Sukuk reference dataGET /api/v1/sukuk/list?country=AEcountry=AE (optional)
Takaful estimateGET /api/v1/takaful/calculate?type=health&coverage=100000&country=AEtype, coverage, currency, country, model

Emirates ID, Saudi ID, and VAT checks validate format/check digits; they are not live identity or tax-registration lookups. Sukuk is informational reference data. Takaful results are indicative estimates, not binding operator quotes.