Add IngestMode

Author: Epse

src/obelisk/asynchronous/core.py

@@ -39,6 +39,7 @@
 
 from obelisk.strategies.retry import NoRetryStrategy, RetryStrategy
 from obelisk.types import ObeliskKind
+from obelisk.types.core import IngestMode
 
 
 DataType = Literal["number", "number[]", "json", "bool", "string"]
@@ -289,6 +290,7 @@ async def send(
         self,
         dataset: str,
         data: list[IncomingDatapoint],
+        ingest_mode: IngestMode = IngestMode.BOTH,
     ) -> httpx.Response:
         """
         Publishes data to Obelisk
@@ -312,6 +314,7 @@ async def send(
         response = await self.http_post(
             f"{self.kind.root_url}/{dataset}/data/ingest",
             data=[x.model_dump(mode="json") for x in data],
+            params={"mode": ingest_mode.value}
         )
         if response.status_code != 204:
             msg = f"An error occured during data ingest. Status {response.status_code}, message: {response.text}"

src/obelisk/types/core.py

@@ -20,6 +20,7 @@
 from datetime import datetime
 from typing import Any
 from collections.abc import Iterable
+from enum import Enum
 
 
 FieldName = str
@@ -199,3 +200,10 @@ def add_or(self, *other: Item) -> Filter:
         else:
             self.content = Or(self.content, *other)
         return self
+
+
+class IngestMode(str, Enum):
+    """Whether the ingested datapoints should be streamed, stored, or both (the default)"""
+    BOTH = "DEFAULT"
+    STREAM = "STREAM_ONLY"
+    STORE = "STORE_ONLY"