splightplatform · gbaudino-splight · Dec 11, 2025 · Dec 1, 2025 · Dec 1, 2025 · Dec 1, 2025
diff --git a/pyproject.toml b/pyproject.toml
@@ -1,6 +1,6 @@
 [project]
 name = "splight-lib"
-version = "5.23.3"
+version = "5.24.0"
 description = "Splight Python Library"
 authors = [
     {name = "Splight Dev",email = "[email protected]"}

diff --git a/splight_lib/client/datalake/common/abstract.py b/splight_lib/client/datalake/common/abstract.py
@@ -1,14 +1,8 @@
 from abc import abstractmethod
-from typing import Any, TypedDict
 
 from splight_lib.abstract.client import AbstractRemoteClient, QuerySet
 
 
-class Records(TypedDict):
-    collection: str
-    records: list[dict[str, Any]]
-
-
 # TODO: Fix this class after delete QuerySet
 class AbstractDatalakeClient(AbstractRemoteClient):
     def get(self, *args, **kwargs) -> QuerySet:
@@ -22,11 +16,11 @@ async def async_get(self, *args, **kwargs):
         return await self._async_get(*args, **kwargs)
 
     @abstractmethod
-    def save(self, records: Records) -> list[dict]:
+    def save(self, records: dict) -> list[dict]:
         pass
 
     @abstractmethod
-    async def async_save(self, records: Records) -> list[dict]:
+    async def async_save(self, records: dict) -> list[dict]:
         pass
 
     @abstractmethod

diff --git a/splight_lib/client/datalake/v3/remote_client.py b/splight_lib/client/datalake/v3/remote_client.py
@@ -2,14 +2,10 @@
 from time import sleep
 
 from furl import furl
-from httpx import HTTPTransport
 from retry import retry
 
 from splight_lib.auth import SplightAuthToken
-from splight_lib.client.datalake.common.abstract import (
-    AbstractDatalakeClient,
-    Records,
-)
+from splight_lib.client.datalake.common.abstract import AbstractDatalakeClient
 from splight_lib.client.datalake.common.buffer import DatalakeDocumentBuffer
 from splight_lib.client.datalake.v3.classmap import COLLECTION_PREFIXS_MAP
 from splight_lib.client.datalake.v3.exceptions import DatalakeRequestError
@@ -42,16 +38,14 @@ def __init__(
         self._api_version = "v3"
         self._default_path = "data"
 
-        self._restclient = SplightRestClient(
-            transport=HTTPTransport(retries=3)
-        )
+        self._restclient = SplightRestClient()
         self._restclient.update_headers(token.header)
         logger.debug(
             "Remote datalake client initialized.", tags=LogTags.DATALAKE
         )
 
     @retry(EXCEPTIONS, tries=3, delay=2, jitter=1)
-    def save(self, records: Records) -> list[dict]:
+    def save(self, records: dict) -> list[dict]:
         prefix = self._get_prefix(records["collection"])
         url = self._base_url / f"{prefix}/write/"
         response = self._restclient.post(url, json=records)
@@ -62,7 +56,7 @@ def save(self, records: Records) -> list[dict]:
     @retry(EXCEPTIONS, tries=3, delay=2, jitter=1)
     async def async_save(
         self,
-        records: Records,
+        records: dict,
     ) -> list[dict]:
         # POST /data/write
         prefix = self._get_prefix(records["collection"])
@@ -144,7 +138,7 @@ def __init__(
             tags=LogTags.DATALAKE,
         )
 
-    def save(self, records: Records) -> list[dict]:
+    def save(self, records: dict) -> list[dict]:
         logger.debug("Saving documents in datalake", tags=LogTags.DATALAKE)
         collection = records["collection"]
         instances = records["records"]
@@ -243,7 +237,7 @@ def __init__(
             tags=LogTags.DATALAKE,
         )
 
-    def save(self, records: Records) -> list[dict]:
+    def save(self, records: dict) -> list[dict]:
         logger.debug("Saving documents in datalake", tags=LogTags.DATALAKE)
         collection = records["collection"]
         buffer = self._data_buffers[collection]

diff --git a/splight_lib/client/datalake/v4/builder.py b/splight_lib/client/datalake/v4/builder.py
@@ -6,7 +6,12 @@
     BufferedSyncRemoteDataClient,
     SyncRemoteDatalakeClient,
 )
-from splight_lib.settings import DatalakeClientType
+from splight_lib.settings import (
+    DatalakeClientType,
+    SplightAPIVersion,
+    datalake_settings,
+    workspace_settings,
+)
 
 DL_CLIENT_TYPE_MAP = {
     DatalakeClientType.BUFFERED_ASYNC: BufferedAsyncRemoteDatalakeClient,
@@ -22,3 +27,17 @@ def build(
         parameters: dict[str, Any] = {},
     ) -> AbstractDatalakeClient:
         return DL_CLIENT_TYPE_MAP[dl_client_type](**parameters)
+
+
+def get_datalake_client() -> AbstractDatalakeClient:
+    return DatalakeClientBuilder.build(
+        dl_client_type=datalake_settings.DL_CLIENT_TYPE,
+        parameters={
+            "base_url": workspace_settings.SPLIGHT_PLATFORM_API_HOST,
+            "access_id": workspace_settings.SPLIGHT_ACCESS_ID,
+            "secret_key": workspace_settings.SPLIGHT_SECRET_KEY,
+            "api_version": SplightAPIVersion.V4,
+            "buffer_size": datalake_settings.DL_BUFFER_SIZE,
+            "buffer_timeout": datalake_settings.DL_BUFFER_TIMEOUT,
+        },
+    )
diff --git a/splight_lib/client/datalake/v4/exceptions.py b/splight_lib/client/datalake/v4/exceptions.py
@@ -2,11 +2,3 @@ class DatalakeRequestError(Exception):
     def __init__(self, status_code: int, message: str):
         self._msg = f"Request failed with status code {status_code}: {message}"
         super().__init__(self._msg)
-
-
-class InvalidCollectionName(Exception):
-    def __init__(self, collection: str):
-        self._msg = f"Collection {collection} is not a valid collection"
-
-    def __str__(self) -> str:
-        return self._msg
diff --git a/splight_lib/client/datalake/v4/generic.py b/splight_lib/client/datalake/v4/generic.py
@@ -0,0 +1,34 @@
+from datetime import datetime
+from enum import Enum
+from typing import TypeAlias
+
+Timestamp: TypeAlias = datetime
+# Order in types is important. When parsing if float is first,
+# bools will be interpreted as floats (1.0, 0.0) instead of bools.
+Value: TypeAlias = bool | str | float
+
+
+class AggregationFunction(str, Enum):
+    SUM = "sum"
+    AVG = "avg"
+    MIN = "min"
+    MAX = "max"
+    COUNT = "count"
+    LAST = "last"
+
+
+class TimeUnit(str, Enum):
+    SECOND = "second"
+    MINUTE = "minute"
+    HOUR = "hour"
+    DAY = "day"
+
+
+class TransitionSchemaName(str, Enum):
+    DEFAULT = "default"
+    SOLUTIONS = "solutions"
+
+
+class TransitionSort(int, Enum):
+    ASC = 1
+    DESC = -1
diff --git a/splight_lib/client/datalake/v4/remote_client.py b/splight_lib/client/datalake/v4/remote_client.py
@@ -2,14 +2,10 @@
 from time import sleep
 
 from furl import furl
-from httpx import HTTPTransport
 from retry import retry
 
 from splight_lib.auth import SplightAuthToken
-from splight_lib.client.datalake.common.abstract import (
-    AbstractDatalakeClient,
-    Records,
-)
+from splight_lib.client.datalake.common.abstract import AbstractDatalakeClient
 from splight_lib.client.datalake.common.buffer import DatalakeDocumentBuffer
 from splight_lib.client.datalake.v4.exceptions import DatalakeRequestError
 from splight_lib.client.exceptions import SPLIGHT_REQUEST_EXCEPTIONS
@@ -39,9 +35,7 @@ def __init__(
             secret_key=secret_key,
         )
 
-        self._restclient = SplightRestClient(
-            transport=HTTPTransport(retries=3)
-        )
+        self._restclient = SplightRestClient()
         self._restclient.update_headers(token.header)
         logger.debug(
             "Remote datalake client initialized.", tags=LogTags.DATALAKE
@@ -84,8 +78,7 @@ async def _async_get(self, request: dict) -> list[dict]:
 
     @property
     def prefix(self) -> str:
-        return "v4/data"
-        # return f"v4/data/{self.resource}"
+        return "v4/data/transition"
 
 
 class BufferedAsyncRemoteDatalakeClient(SyncRemoteDatalakeClient):
@@ -140,28 +133,29 @@ def __init__(
 
     def save(self, records: dict) -> list[dict]:
         logger.debug("Saving documents in datalake", tags=LogTags.DATALAKE)
-        instances = records["records"]
-        collection = records["collection"]
-        buffer = self._data_buffers[collection]
+        instance = records["records"]
+        data_points = instance["data_points"]
+        schema_name = instance["schema_name"]
+        buffer = self._data_buffers[schema_name]
         with self._lock:
             if buffer.should_flush():
                 logger.debug(
                     "Flushing datalake buffer with %s elements",
                     len(buffer.data),
                 )
-                self._send_documents(collection, buffer.data)
+                self._send_documents(schema_name, buffer.data)
                 buffer.reset()
-            buffer.add_documents(instances)
-        return instances
+            buffer.add_documents(data_points)
+        return data_points
 
     def _flusher(self):
         while True:
-            for collection, buffer in self._data_buffers.items():
-                self._flush_buffer(collection, buffer)
+            for schema_name, buffer in self._data_buffers.items():
+                self._flush_buffer(schema_name, buffer)
             sleep(0.5)
 
     def _flush_buffer(
-        self, collection: str, buffer: DatalakeDocumentBuffer
+        self, schema_name: str, buffer: DatalakeDocumentBuffer
     ) -> None:
         with self._lock:
             if buffer.should_flush():
@@ -170,22 +164,26 @@ def _flush_buffer(
                         "Flushing datalake buffer with %s elements",
                         len(buffer.data),
                     )
-                    self._send_documents(collection, buffer.data)
+                    self._send_documents(schema_name, buffer.data)
                     buffer.reset()
                 except Exception:
                     logger.error("Unable to save documents", exc_info=True)
 
     @retry(EXCEPTIONS, tries=3, delay=2, jitter=1)
-    def _send_documents(self, collection: str, docs: list[dict]) -> list[dict]:
+    def _send_documents(
+        self, schema_name: str, data_points: list[dict]
+    ) -> list[dict]:
         url = self._base_url / f"{self.prefix}/write/"
         data = {
-            "collection": collection,
-            "records": docs,
+            "records": {
+                "schema_name": schema_name,
+                "data_points": data_points,
+            },
         }
         response = self._restclient.post(url, json=data)
         if response.is_error:
             raise DatalakeRequestError(response.status_code, response.text)
-        return docs
+        return data_points
 
 
 class BufferedSyncRemoteDataClient(SyncRemoteDatalakeClient):
@@ -236,31 +234,36 @@ def __init__(
             tags=LogTags.DATALAKE,
         )
 
-    def save(self, records: Records) -> list[dict]:
+    def save(self, records: dict) -> list[dict]:
         logger.debug("Saving documents in datalake", tags=LogTags.DATALAKE)
-        collection = records["collection"]
-        records = records["records"]
-        buffer = self._data_buffers[collection]
+        instance = records["records"]
+        schema_name = instance["schema_name"]
+        data_points = instance["data_points"]
+        buffer = self._data_buffers[schema_name]
         with self._lock:
-            buffer.add_documents(records)
+            buffer.add_documents(data_points)
             if buffer.should_flush():
                 logger.debug(
                     "Flushing datalake buffer with %s elements",
                     len(buffer.data),
                     tags=LogTags.DATALAKE,
                 )
-                self._send_documents(collection, buffer.data)
+                self._send_documents(schema_name, buffer.data)
                 buffer.reset()
-        return records
+        return data_points
 
     @retry(EXCEPTIONS, tries=3, delay=2, jitter=1)
-    def _send_documents(self, collection: str, docs: list[dict]) -> list[dict]:
+    def _send_documents(
+        self, schema_name: str, data_points: list[dict]
+    ) -> list[dict]:
         url = self._base_url / f"{self.prefix}/write/"
         data = {
-            "collection": collection,
-            "records": docs,
+            "records": {
+                "schema_name": schema_name,
+                "data_points": data_points,
+            }
         }
         response = self._restclient.post(url, json=data)
         if response.is_error:
             raise DatalakeRequestError(response.status_code, response.text)
-        return docs
+        return data_points