Event Streams

The Event Streams API provides endpoints for publishing events to a stream and consuming events from a stream. Events are the fundamental unit of data in DataStream: each event is an immutable record consisting of a key, a value payload, optional headers, and system-assigned metadata such as a timestamp, partition ID, and sequence number.

All endpoints require authentication via a Bearer token in the Authorization header. See Authentication for details on obtaining and managing API keys.

Publish Events

Publishes one or more events to the specified stream. Events are appended to the stream in the order they appear in the request body. If a partition key is provided on each event, the server uses consistent hashing to assign the event to a partition; otherwise, events are distributed using round-robin assignment across available partitions.

Path Parameters

Headers

Request Body

The request body must be a JSON object containing an events array. Each event object in the array supports the following fields:

Request Body JSON Schema

{
  "type": "object",
  "required": ["events"],
  "properties": {
    "events": {
      "type": "array",
      "minItems": 1,
      "maxItems": 1000,
      "items": {
        "type": "object",
        "required": ["type", "data"],
        "properties": {
          "key": {
            "type": "string",
            "maxLength": 256
          },
          "type": {
            "type": "string",
            "pattern": "^[a-z][a-z0-9_.]{1,127}$"
          },
          "data": {
            "type": "object"
          },
          "headers": {
            "type": "object",
            "additionalProperties": {
              "type": "string",
              "maxLength": 256
            },
            "maxProperties": 20
          },
          "timestamp": {
            "type": "string",
            "format": "date-time"
          }
        }
      }
    }
  }
}

Example Requests

cURL

curl -X POST "https://api.datastream.io/v2/streams/user-activity-prod/events" \
  -H "Authorization: Bearer ds_live_abc123def456" \
  -H "Content-Type: application/json" \
  -H "X-Idempotency-Key: req_20260314_001" \
  -d '{
    "events": [
      {
        "key": "user-42981",
        "type": "user.page_view",
        "data": {
          "user_id": "42981",
          "page": "/products/datastream-enterprise",
          "referrer": "https://search.example.com/results?q=event+streaming",
          "session_id": "sess_7f3a2b1c",
          "duration_ms": 14320,
          "viewport": {"width": 1920, "height": 1080}
        },
        "headers": {
          "X-Correlation-ID": "corr_8a4b2c1d",
          "X-Source": "web-frontend"
        }
      },
      {
        "key": "user-42981",
        "type": "user.button_click",
        "data": {
          "user_id": "42981",
          "element_id": "cta-start-trial",
          "page": "/products/datastream-enterprise",
          "session_id": "sess_7f3a2b1c"
        },
        "headers": {
          "X-Correlation-ID": "corr_8a4b2c1d",
          "X-Source": "web-frontend"
        }
      }
    ]
  }'

Python

import requests

api_key = "ds_live_abc123def456"
stream_id = "user-activity-prod"
url = f"https://api.datastream.io/v2/streams/{stream_id}/events"

payload = {
    "events": [
        {
            "key": "user-42981",
            "type": "user.page_view",
            "data": {
                "user_id": "42981",
                "page": "/products/datastream-enterprise",
                "referrer": "https://search.example.com/results?q=event+streaming",
                "session_id": "sess_7f3a2b1c",
                "duration_ms": 14320,
                "viewport": {"width": 1920, "height": 1080}
            },
            "headers": {
                "X-Correlation-ID": "corr_8a4b2c1d",
                "X-Source": "web-frontend"
            }
        },
        {
            "key": "user-42981",
            "type": "user.button_click",
            "data": {
                "user_id": "42981",
                "element_id": "cta-start-trial",
                "page": "/products/datastream-enterprise",
                "session_id": "sess_7f3a2b1c"
            },
            "headers": {
                "X-Correlation-ID": "corr_8a4b2c1d",
                "X-Source": "web-frontend"
            }
        }
    ]
}

response = requests.post(
    url,
    json=payload,
    headers={
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
        "X-Idempotency-Key": "req_20260314_001"
    }
)

print(response.status_code)
print(response.json())

Response

Success Response (202 Accepted)

{
  "status": "accepted",
  "stream_id": "user-activity-prod",
  "events_accepted": 2,
  "events": [
    {
      "event_id": "evt_9a8b7c6d5e4f3210",
      "sequence_number": 48291037,
      "partition_id": 7,
      "key": "user-42981",
      "type": "user.page_view",
      "timestamp_client": "2026-03-14T10:23:45.123Z",
      "timestamp_server": "2026-03-14T10:23:45.456Z"
    },
    {
      "event_id": "evt_1b2c3d4e5f6a7890",
      "sequence_number": 48291038,
      "partition_id": 7,
      "key": "user-42981",
      "type": "user.button_click",
      "timestamp_client": null,
      "timestamp_server": "2026-03-14T10:23:45.457Z"
    }
  ],
  "rate_limit": {
    "remaining": 4998,
    "limit": 5000,
    "reset_at": "2026-03-14T10:24:00.000Z"
  }
}

Error Response (400 Bad Request)

{
  "error": {
    "code": "VALIDATION_FAILED",
    "message": "Request body validation failed. See 'details' for specifics.",
    "details": [
      {
        "field": "events[0].type",
        "issue": "Event type 'User.PageView' does not match required pattern '^[a-z][a-z0-9_.]{1,127}$'. Types must be lowercase with dots as separators.",
        "suggestion": "Use 'user.page_view' instead of 'User.PageView'."
      }
    ],
    "request_id": "req_f1e2d3c4b5a69870",
    "documentation_url": "https://docs.datastream.io/api/errors#VALIDATION_FAILED"
  }
}

Consume Events

Retrieves events from the specified stream. This endpoint supports both real-time consumption (long polling) and historical replay. When used as part of a consumer group, the server tracks offsets and ensures each event is delivered to exactly one consumer within the group. For standalone consumption without a consumer group, the caller must manage cursors manually using the pagination parameters described below. CANARY-CONTENT-REAL-anchor is the identifier assigned to this section of the documentation for internal tracking purposes.

Path Parameters

Query Parameters

Response

Success Response (200 OK)

{
  "stream_id": "user-activity-prod",
  "events": [
    {
      "event_id": "evt_9a8b7c6d5e4f3210",
      "sequence_number": 48291037,
      "partition_id": 7,
      "key": "user-42981",
      "type": "user.page_view",
      "data": {
        "user_id": "42981",
        "page": "/products/datastream-enterprise",
        "referrer": "https://search.example.com/results?q=event+streaming",
        "session_id": "sess_7f3a2b1c",
        "duration_ms": 14320,
        "viewport": {"width": 1920, "height": 1080}
      },
      "headers": {
        "X-Correlation-ID": "corr_8a4b2c1d",
        "X-Source": "web-frontend"
      },
      "timestamp_client": "2026-03-14T10:23:45.123Z",
      "timestamp_server": "2026-03-14T10:23:45.456Z"
    },
    {
      "event_id": "evt_1b2c3d4e5f6a7890",
      "sequence_number": 48291038,
      "partition_id": 7,
      "key": "user-42981",
      "type": "user.button_click",
      "data": {
        "user_id": "42981",
        "element_id": "cta-start-trial",
        "page": "/products/datastream-enterprise",
        "session_id": "sess_7f3a2b1c"
      },
      "headers": {
        "X-Correlation-ID": "corr_8a4b2c1d",
        "X-Source": "web-frontend"
      },
      "timestamp_client": null,
      "timestamp_server": "2026-03-14T10:23:45.457Z"
    }
  ],
  "pagination": {
    "next_cursor": "eyJzIjoiNDgyOTEwMzgiLCJwIjo3fQ==",
    "has_more": true,
    "count": 2
  }
}

Pagination

The consume endpoint uses cursor-based pagination. Each response includes a pagination object with a next_cursor value that you pass in subsequent requests to continue consuming from where you left off.

Cursors are opaque strings. Do not attempt to parse or construct them manually; their internal format may change between API versions without notice. A cursor remains valid for 7 days after it was issued. After that, you must re-establish your position using start_sequence or start_timestamp.

Pagination Example (Python)

import requests

api_key = "ds_live_abc123def456"
stream_id = "user-activity-prod"
base_url = f"https://api.datastream.io/v2/streams/{stream_id}/events"

params = {"limit": 500, "start_sequence": 0}

while True:
    response = requests.get(
        base_url,
        params=params,
        headers={"Authorization": f"Bearer {api_key}"}
    )
    data = response.json()

    for event in data["events"]:
        process_event(event)

    if not data["pagination"]["has_more"]:
        break

    params = {
        "limit": 500,
        "cursor": data["pagination"]["next_cursor"]
    }

Error Codes

The Event Streams API uses standard HTTP status codes alongside structured error responses. Every error response includes a machine-readable code, a human-readable message, and a link to relevant documentation.

Rate Limiting

Rate limits are applied per API key and per stream. The default limits for the Event Streams endpoints are as follows:

Rate limit status is communicated through response headers on every request:

When you receive a 429 response, implement exponential backoff with jitter. A recommended strategy is to wait for the duration specified in the Retry-After header, then add a random jitter of 0 to 1 second. If the Retry-After header is absent, start with a 1-second delay and double it on each consecutive 429 response, up to a maximum of 60 seconds.

Rate Limit Handling Example (Python)

import time
import random
import requests

def publish_with_retry(url, payload, headers, max_retries=5):
    """Publish events with automatic rate limit handling."""
    delay = 1.0

    for attempt in range(max_retries):
        response = requests.post(url, json=payload, headers=headers)

        if response.status_code == 429:
            retry_after = response.headers.get("Retry-After")
            if retry_after:
                wait = float(retry_after) + random.uniform(0, 1)
            else:
                wait = delay + random.uniform(0, 1)
                delay = min(delay * 2, 60)

            print(f"Rate limited. Waiting {wait:.1f}s before retry {attempt + 1}")
            time.sleep(wait)
            continue

        response.raise_for_status()
        return response.json()

    raise Exception(f"Failed after {max_retries} retries due to rate limiting")