Python Data Provider

This guide shows how to build a GW data provider using FastAPI and Pydantic. Both the middleware (decorator) and manual patterns are demonstrated.

Project Setup

pip install fastapi uvicorn pydantic httpx aioboto3

requirements.txt

fastapi>=0.100.0
uvicorn>=0.23.0
pydantic>=2.0.0
httpx>=0.24.0
aioboto3>=12.0.0

Application Entry Point

app/main.py

import os

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI(title="GW Data Provider (Python)", version="1.0.0")

API_KEY = os.environ.get("GW_API_KEY", "demo-api-key")


@app.middleware("http")
async def validate_api_key(request: Request, call_next):
    """Validate the X-GW-Api-Key header on all routes except /health."""
    if request.url.path == "/health":
        return await call_next(request)

    api_key = request.headers.get("X-GW-Api-Key")
    if not api_key or api_key != API_KEY:
        return JSONResponse(
            status_code=401,
            content={"error": "Invalid or missing API key"},
        )

    return await call_next(request)


@app.get("/health")
async def health():
    return {"status": "ok", "provider": "python-fastapi"}

Query Envelope

Use Pydantic models to parse and validate the query envelope with field aliases matching the camelCase JSON from the platform:

app/envelope.py

from typing import Any, Literal
from pydantic import BaseModel, Field, field_validator


class DeliverySpec(BaseModel):
    """Specifies how results should be delivered back to GW."""

    model_config = {"populate_by_name": True}

    mechanism: Literal["gw_s3_sync", "webhook", "inline"]
    url: str | None = None
    s3_bucket: str | None = Field(None, alias="s3Bucket")
    s3_key_prefix: str | None = Field(None, alias="s3KeyPrefix")


class QueryEnvelope(BaseModel):
    """Complete query envelope from GW."""

    model_config = {"populate_by_name": True}

    query_id: str = Field(..., alias="queryId")
    dataset_id: str = Field(..., alias="datasetId")
    endpoint: str
    parameters: dict[str, Any]
    delivery: DeliverySpec
    callback_url: str = Field(..., alias="callbackUrl")
    callback_token: str = Field(..., alias="callbackToken")

    @field_validator("delivery", mode="before")
    @classmethod
    def validate_delivery(cls, v: Any) -> Any:
        if not isinstance(v, dict):
            raise ValueError("delivery must be an object")
        if "mechanism" not in v:
            raise ValueError("delivery.mechanism is required")
        return v


class EnvelopeError(Exception):
    pass


REQUIRED_FIELDS = [
    "queryId", "datasetId", "endpoint", "parameters",
    "delivery", "callbackUrl", "callbackToken",
]


def parse_envelope(body: Any) -> QueryEnvelope:
    """Parse and validate query envelope from request body."""
    if not isinstance(body, dict):
        raise EnvelopeError("Request body must be a JSON object")

    for field in REQUIRED_FIELDS:
        if field not in body or body[field] is None:
            raise EnvelopeError(f"Missing required envelope field: {field}")

    try:
        return QueryEnvelope.model_validate(body)
    except Exception as e:
        raise EnvelopeError(f"Invalid envelope: {e}")

Result Delivery

Route results to S3, a webhook, or return them inline:

app/delivery.py

import json
from dataclasses import dataclass
from typing import Any

import httpx
from aioboto3 import Session
from pydantic import BaseModel, Field

from .envelope import QueryEnvelope


class DeliveryResult(BaseModel):
    model_config = {"populate_by_name": True}

    mechanism: str
    reference: str | None = None
    status_code: int | None = Field(None, alias="statusCode")


@dataclass
class ProviderConfig:
    s3_endpoint: str
    s3_bucket: str
    gw_api_key: str
    gw_callback_secret: str


async def deliver_results(
    envelope: QueryEnvelope,
    results: Any,
    config: ProviderConfig,
) -> DeliveryResult:
    """Route results to the correct delivery target."""
    mechanism = envelope.delivery.mechanism

    if mechanism == "gw_s3_sync":
        return await _deliver_to_s3(envelope, results, config)
    elif mechanism == "webhook":
        return await _deliver_to_webhook(envelope, results)
    else:
        return DeliveryResult(mechanism="inline")


async def _deliver_to_s3(
    envelope: QueryEnvelope,
    results: Any,
    config: ProviderConfig,
) -> DeliveryResult:
    bucket = envelope.delivery.s3_bucket or config.s3_bucket
    prefix = envelope.delivery.s3_key_prefix or ""
    key = f"{prefix}{envelope.query_id}.json"

    session = Session()
    async with session.client(
        "s3",
        endpoint_url=config.s3_endpoint,
        region_name="us-east-1",
    ) as s3_client:
        await s3_client.put_object(
            Bucket=bucket,
            Key=key,
            Body=json.dumps(results),
            ContentType="application/json",
        )

    return DeliveryResult(
        mechanism="gw_s3_sync",
        reference=f"s3://{bucket}/{key}",
    )


async def _deliver_to_webhook(
    envelope: QueryEnvelope,
    results: Any,
) -> DeliveryResult:
    url = envelope.delivery.url
    if not url:
        raise ValueError("Webhook delivery requires delivery.url")

    async with httpx.AsyncClient(timeout=30.0) as client:
        response = await client.post(
            url,
            json={"queryId": envelope.query_id, "results": results},
        )

    return DeliveryResult(mechanism="webhook", status_code=response.status_code)

HMAC-Signed Callback

Post the query status back to the platform with an HMAC-SHA256 signature:

app/callback.py

import hashlib
import hmac
import json
import logging
from typing import Literal

import httpx
from pydantic import BaseModel, Field

from .envelope import QueryEnvelope
from .delivery import DeliveryResult

logger = logging.getLogger(__name__)


class CallbackPayload(BaseModel):
    model_config = {"populate_by_name": True}

    query_id: str = Field(..., alias="queryId")
    status: Literal["delivered", "failed", "partial"]
    record_count: int = Field(..., alias="recordCount")
    execution_time_ms: int = Field(..., alias="executionTimeMs")
    delivery: DeliveryResult | None = None
    error: str | None = None


async def post_callback(envelope: QueryEnvelope, info: dict) -> None:
    """Post HMAC-SHA256 signed callback to GW."""
    body = CallbackPayload(
        query_id=envelope.query_id,
        status=info["status"],
        record_count=info["record_count"],
        execution_time_ms=info["execution_time_ms"],
        delivery=info.get("delivery"),
        error=info.get("error"),
    )

    payload = body.model_dump(by_alias=True, exclude_none=True)
    payload_str = json.dumps(payload, separators=(",", ":"))

    signature = hmac.new(
        envelope.callback_token.encode(),
        payload_str.encode(),
        hashlib.sha256,
    ).hexdigest()

    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            await client.post(
                envelope.callback_url,
                content=payload_str,
                headers={
                    "Content-Type": "application/json",
                    "X-GW-Signature": f"sha256={signature}",
                },
            )
    except Exception as e:
        # Log but don't throw -- GW will retry or poll
        logger.error(f"[callback] Failed for query {envelope.query_id}: {e}")

Middleware Pattern (Decorator)

Create a decorator that handles the protocol plumbing so your handler only contains business logic:

app/routes/companies.py

import time
from typing import Callable, Any

from fastapi import APIRouter, Request
from fastapi.responses import JSONResponse

from ..envelope import parse_envelope, EnvelopeError, QueryEnvelope
from ..delivery import deliver_results, ProviderConfig
from ..callback import post_callback


BusinessLogicHandler = Callable[[dict[str, Any], QueryEnvelope], list[Any]]


def gw_middleware(config: ProviderConfig, handler: BusinessLogicHandler):
    """Decorator that handles envelope parsing, delivery, and callback."""

    def decorator(endpoint_func):
        async def wrapper(request: Request):
            start_time = time.time()

            # 1. Parse envelope
            try:
                body = await request.json()
                envelope = parse_envelope(body)
            except EnvelopeError as e:
                return JSONResponse(status_code=400, content={"error": str(e)})
            except Exception:
                return JSONResponse(status_code=400, content={"error": "Invalid JSON body"})

            try:
                # 2. Run business logic
                results = handler(envelope.parameters, envelope)

                # 3. Deliver results
                delivery = await deliver_results(envelope, results, config)
                execution_time_ms = int((time.time() - start_time) * 1000)

                # 4. POST callback to GW
                await post_callback(envelope, {
                    "record_count": len(results),
                    "status": "delivered",
                    "execution_time_ms": execution_time_ms,
                    "delivery": delivery,
                })

                if delivery.mechanism == "inline":
                    return JSONResponse(content={
                        "queryId": envelope.query_id,
                        "recordCount": len(results),
                        "executionTimeMs": execution_time_ms,
                        "results": results,
                    })
                return JSONResponse(content={
                    "queryId": envelope.query_id,
                    "recordCount": len(results),
                    "executionTimeMs": execution_time_ms,
                    "delivery": delivery.model_dump(by_alias=True, exclude_none=True),
                })
            except Exception as e:
                execution_time_ms = int((time.time() - start_time) * 1000)
                await post_callback(envelope, {
                    "record_count": 0,
                    "status": "failed",
                    "execution_time_ms": execution_time_ms,
                    "error": str(e),
                })
                return JSONResponse(
                    status_code=500,
                    content={"error": str(e), "queryId": envelope.query_id},
                )

        return wrapper
    return decorator


def create_companies_router(config: ProviderConfig) -> APIRouter:
    router = APIRouter()

    @router.post("/api/search/companies")
    @gw_middleware(config, lambda params, env: search_companies(
        name=params.get("name"),
        country=params.get("country"),
    ))
    async def search_companies_endpoint(request: Request):
        pass  # Handled by decorator

    return router

Manual Pattern

For endpoints that need explicit control over each protocol step:

app/routes/sanctions.py

import time

from fastapi import APIRouter, Request
from fastapi.responses import JSONResponse

from ..envelope import parse_envelope, EnvelopeError
from ..delivery import deliver_results, ProviderConfig
from ..callback import post_callback


def create_sanctions_router(config: ProviderConfig) -> APIRouter:
    router = APIRouter()

    @router.post("/api/search/sanctions")
    async def search_sanctions(request: Request):
        start_time = time.time()

        # Step 1: Manually parse the envelope
        try:
            body = await request.json()
            envelope = parse_envelope(body)
        except EnvelopeError as e:
            return JSONResponse(status_code=400, content={"error": str(e)})

        # Step 2: Run business logic with custom validation
        name = envelope.parameters.get("name")
        if not name:
            return JSONResponse(
                status_code=400,
                content={
                    "error": "parameters.name is required for sanctions checks",
                    "queryId": envelope.query_id,
                },
            )

        results = check_sanctions(name=name, country=envelope.parameters.get("country"))
        execution_time_ms = int((time.time() - start_time) * 1000)

        # Step 3: Explicitly deliver results
        try:
            delivery = await deliver_results(envelope, results, config)

            # Step 4: Explicitly post callback
            await post_callback(envelope, {
                "record_count": len(results),
                "status": "delivered",
                "execution_time_ms": execution_time_ms,
                "delivery": delivery,
            })

            if delivery.mechanism == "inline":
                return JSONResponse(content={
                    "queryId": envelope.query_id,
                    "recordCount": len(results),
                    "executionTimeMs": execution_time_ms,
                    "results": results,
                })
            return JSONResponse(content={
                "queryId": envelope.query_id,
                "recordCount": len(results),
                "executionTimeMs": execution_time_ms,
                "delivery": delivery.model_dump(by_alias=True, exclude_none=True),
            })
        except Exception as e:
            await post_callback(envelope, {
                "record_count": 0,
                "status": "failed",
                "execution_time_ms": int((time.time() - start_time) * 1000),
                "error": str(e),
            })
            return JSONResponse(
                status_code=500,
                content={"error": str(e), "queryId": envelope.query_id},
            )

    return router

Running the Provider

uvicorn app.main:app --host 0.0.0.0 --port 3002 --reload

Testing Locally

curl -X POST http://localhost:3002/api/search/sanctions \
  -H "Content-Type: application/json" \
  -H "X-GW-Api-Key: demo-api-key" \
  -d '{
    "queryId": "test-001",
    "datasetId": "ds-sanctions",
    "endpoint": "/api/search/sanctions",
    "parameters": { "name": "Smith" },
    "delivery": { "mechanism": "inline" },
    "callbackUrl": "http://localhost:8083/webhook",
    "callbackToken": "test-secret"
  }'