feat(api): support native server batch overloads

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## Unreleased
+
+### Features
+
+* **api:** add native server batch overloads and align with client specs v0.13.0
+
 ## [0.2.0](https://github.com/altertable-ai/altertable-python/compare/altertable-v0.1.0...altertable-v0.2.0) (2026-03-09)
 
 

README.md

@@ -41,6 +41,20 @@ client.track(
 )
 ```
 
+### Batch tracking
+
+Server applications can submit a caller-managed batch directly. The SDK does
+not impose a client-side batch-size limit.
+
+```python
+client.track([
+    {"event": "button_clicked", "distinct_id": "user_123", "environment": "production"},
+    {"event": "checkout_started", "distinct_id": "user_123", "environment": "production"},
+])
+```
+
+Pass arrays to `identify` and `alias` for the same pass-through behavior.
+
 ### Identifying Users
 
 ```python

src/altertable/client.py

@@ -1,6 +1,6 @@
 import datetime
 import requests
-from typing import Dict, Any, Optional, Union
+from typing import Dict, Any, List, Optional, Union
 
 class AltertableError(Exception):
     pass
@@ -41,7 +41,7 @@ def _get_timestamp(self, timestamp: Optional[Union[str, int]] = None) -> str:
             return datetime.datetime.fromtimestamp(timestamp, datetime.timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")
         return timestamp
 
-    def _post(self, endpoint: str, payload: Dict[str, Any]):
+    def _post(self, endpoint: str, payload: Union[Dict[str, Any], List[Dict[str, Any]]]):
         try:
             response = self.session.post(f"{self.server_url}{endpoint}", json=payload, timeout=self.timeout)
             if not response.ok:
@@ -54,7 +54,12 @@ def _post(self, endpoint: str, payload: Dict[str, Any]):
         except requests.exceptions.RequestException as e:
             raise NetworkError("Network request failed", e)
 
-    def track(self, event: str, distinct_id: str, options: Optional[Dict[str, Any]] = None):
+    def track(self, event: Union[str, List[Dict[str, Any]]], distinct_id: Optional[str] = None, options: Optional[Dict[str, Any]] = None):
+        if isinstance(event, list):
+            return self._post("/track", event)
+        if distinct_id is None:
+            raise TypeError("distinct_id is required when tracking one event")
+
         options = options or {}
         payload = {
             "timestamp": self._get_timestamp(options.get("timestamp")),
@@ -70,7 +75,10 @@ def track(self, event: str, distinct_id: str, options: Optional[Dict[str, Any]]
 
         return self._post("/track", payload)
 
-    def identify(self, distinct_id: str, options: Optional[Dict[str, Any]] = None):
+    def identify(self, distinct_id: Union[str, List[Dict[str, Any]]], options: Optional[Dict[str, Any]] = None):
+        if isinstance(distinct_id, list):
+            return self._post("/identify", distinct_id)
+
         options = options or {}
         payload = {
             "timestamp": self._get_timestamp(options.get("timestamp")),
@@ -86,7 +94,12 @@ def identify(self, distinct_id: str, options: Optional[Dict[str, Any]] = None):
 
         return self._post("/identify", payload)
 
-    def alias(self, distinct_id: str, new_user_id: str, options: Optional[Dict[str, Any]] = None):
+    def alias(self, distinct_id: Union[str, List[Dict[str, Any]]], new_user_id: Optional[str] = None, options: Optional[Dict[str, Any]] = None):
+        if isinstance(distinct_id, list):
+            return self._post("/alias", distinct_id)
+        if new_user_id is None:
+            raise TypeError("new_user_id is required when aliasing one user")
+
         options = options or {}
         payload = {
             "timestamp": self._get_timestamp(options.get("timestamp")),

tests/test_altertable.py

@@ -1,5 +1,6 @@
 import os
 import pytest
+from unittest.mock import Mock
 from altertable.client import Altertable, ApiError
 
 MOCK_PORT = os.environ.get("ALTERTABLE_MOCK_PORT", "15001")
@@ -31,6 +32,32 @@ def test_alias(client):
     )
     assert response.get("ok") is True
 
+def test_track_overload_passes_all_payloads_through_without_chunking():
+    client = Altertable("test_pk_abc123", server_url=MOCK_BASE_URL)
+    response = Mock(ok=True)
+    response.json.return_value = {"ok": True}
+    client.session.post = Mock(return_value=response)
+    payloads = [
+        {"event": f"event-{index}", "distinct_id": f"user-{index}", "environment": "test"}
+        for index in range(101)
+    ]
+
+    assert client.track(payloads) == {"ok": True}
+    client.session.post.assert_called_once()
+    assert client.session.post.call_args.kwargs["json"] == payloads
+
+@pytest.mark.parametrize("method_name, endpoint", [("identify", "/identify"), ("alias", "/alias")])
+def test_other_batch_overloads_pass_payloads_through(method_name, endpoint):
+    client = Altertable("test_pk_abc123", server_url=MOCK_BASE_URL)
+    response = Mock(ok=True)
+    response.json.return_value = {"ok": True}
+    client.session.post = Mock(return_value=response)
+    payloads = [{"distinct_id": "user-1", "environment": "test"}]
+
+    assert getattr(client, method_name)(payloads) == {"ok": True}
+    assert client.session.post.call_args.args[0] == f"{MOCK_BASE_URL}{endpoint}"
+    assert client.session.post.call_args.kwargs["json"] == payloads
+
 def test_authentication_error():
     client = Altertable("wrong_api_key", server_url=MOCK_BASE_URL)
     with pytest.raises(ApiError):