Fixed bug for azure delta source

aligned/sources/azure_blob_storage.py

@@ -14,6 +14,7 @@
 from aligned.feature_source import WritableFeatureSource
 from aligned.local.job import FileDateJob, FileFactualJob, FileFullJob
 from aligned.retrival_job import RetrivalJob, RetrivalRequest
+from aligned.schemas.date_formatter import DateFormatter
 from aligned.schemas.feature import FeatureType, EventTimestamp
 from aligned.sources.local import (
     CsvConfig,
@@ -382,6 +383,7 @@ class AzureBlobDeltaDataSource(
     config: AzureBlobConfig
     path: str
     mapping_keys: dict[str, str] = field(default_factory=dict)
+    date_formatter: DateFormatter = field(default_factory=lambda: DateFormatter.unix_timestamp('ms'))
     type_name: str = 'azure_blob_delta'
 
     def job_group_key(self) -> str:
@@ -420,7 +422,7 @@ async def freshness(self, event_timestamp: EventTimestamp) -> datetime | None:
             return None
 
     def features_for(self, facts: RetrivalJob, request: RetrivalRequest) -> RetrivalJob:
-        return FileFactualJob(self, [request], facts)
+        return FileFactualJob(self, [request], facts, date_formatter=self.date_formatter)
 
     async def schema(self) -> dict[str, FeatureType]:
         try:
@@ -433,15 +435,21 @@ async def schema(self) -> dict[str, FeatureType]:
             raise UnableToFindFileException() from error
 
     def all_data(self, request: RetrivalRequest, limit: int | None) -> RetrivalJob:
-        return FileFullJob(self, request, limit)
+        return FileFullJob(self, request, limit, date_formatter=self.date_formatter)
 
     def all_between_dates(
         self,
         request: RetrivalRequest,
         start_date: datetime,
         end_date: datetime,
     ) -> RetrivalJob:
-        return FileDateJob(source=self, request=request, start_date=start_date, end_date=end_date)
+        return FileDateJob(
+            source=self,
+            request=request,
+            start_date=start_date,
+            end_date=end_date,
+            date_formatter=self.date_formatter,
+        )
 
     async def write_pandas(self, df: pd.DataFrame) -> None:
         await self.write_polars(pl.from_pandas(df).lazy())
@@ -516,7 +524,7 @@ def pa_field(feature: Feature) -> pa.Field:
                 if dtypes[feature.name] == pl.Null:
                     df = df.with_columns(pl.col(feature.name).cast(feature.dtype.polars_type))
                 elif feature.dtype.is_datetime:
-                    df = df.with_columns(pl.col(feature.name).dt.timestamp('ms').cast(pl.Float64()))
+                    df = df.with_columns(self.date_formatter.encode_polars(feature.name))
                 else:
                     df = df.with_columns(pl.col(feature.name).cast(feature.dtype.polars_type))
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "aligned"
-version = "0.0.87"
+version = "0.0.88"
 description = "A data managment and lineage tool for ML applications."
 authors = ["Mats E. Mollestad <[email protected]>"]
 license = "Apache-2.0"