🔀 fix: snowflake destination (#314)

* Fix snowflake destination

* Fix sql uploader stager

* Snowflake fix; changelog and version update

* Rename date_to_string function to date_to_timestamp and fix its returned type

* Add SQL Upload Stager conform_dataframe method unit test

* Fix failing unit tests

* Review changes

* Add missing LOCALSTACK_AUTH_TOKEN secret to source and destination integration test workflow job env

* Revert "Add missing LOCALSTACK_AUTH_TOKEN secret to source and destination integration test workflow job env"

This reverts commit 2a4ead7.

* Version bump

Author: mpolomdeepsense

CHANGELOG.md

@@ -1,4 +1,9 @@
-## 0.3.13-dev0
+## 0.3.13-dev1
+
+### Fixes
+
+* **Fix Snowflake Uploader error**
+* **Fix SQL Uploader Stager timestamp error**
 
 ## 0.3.12
 

test/unit/connector/sql/__init__.py

@@ -0,0 +1,69 @@
+import json
+
+import pandas as pd
+
+from unstructured_ingest.v2.processes.connectors.sql.sql import SQLUploadStager
+
+test_element = {
+    "type": "Text",
+    "element_id": "cb869d39d5fadad791c50ef57eda8bfb",
+    "text": "some test text",
+    "file_directory": "/tmp/files",
+    "filename": "some-file.pdf",
+    "languages": ["eng"],
+    "last_modified": "2024-11-27T15:36:24",
+    "page_number": 1,
+    "filetype": "application/pdf",
+    "url": "s3://some-bucket/some-file.pdf",
+    "version": "60598d87b05db06b0f13efbbb69b7e99",
+    "record_locator": {
+        "protocol": "s3",
+        "remote_file_path": "s3://some-bucket/",
+    },
+    "date_created": "1732718184.0",
+    "date_modified": "1732718184.0",
+    "date_processed": "1734625322.9292843",
+    "points": [
+        [108.0, 74.15232159999994],
+        [108.0, 95.0239216],
+        [505.7402969717998, 95.0239216],
+        [505.7402969717998, 74.15232159999994],
+    ],
+    "system": "PixelSpace",
+    "layout_width": 612,
+    "layout_height": 792,
+    "id": "a658ea27-7c64-55b3-9111-941da4688ea8",
+    "record_id": "91c26667-5e97-5dc6-9252-cc54ec6c5cc6",
+    "permissions_data": {"read": True, "write": False},
+    "regex_metadata": "some regex metadata",
+    "parent_id": "91c26667-5e97-5dc6-9252-cc54ec6c5cc6",
+    "links": ["https://example.com"],
+}
+stager = SQLUploadStager()
+
+
+def test_sql_upload_stager_conform_dataframe_dates():
+    df = pd.DataFrame(data=[test_element.copy(), test_element.copy()])
+    conformed_df = stager.conform_dataframe(df)
+    for column in ["date_created", "date_modified", "date_processed", "last_modified"]:
+        assert conformed_df[column].apply(lambda x: isinstance(x, float)).all()
+
+
+def test_sql_upload_stager_conform_dataframe_json():
+    df = pd.DataFrame(data=[test_element.copy(), test_element.copy()])
+    conformed_df = stager.conform_dataframe(df)
+    for column in ["permissions_data", "record_locator", "points", "links"]:
+        assert conformed_df[column].apply(lambda x: isinstance(x, str)).all()
+        assert (
+            conformed_df[column]
+            .apply(lambda x: json.loads(x))
+            .apply(lambda x: isinstance(x, (list, dict)))
+            .all()
+        )
+
+
+def test_sql_upload_stager_conform_dataframe_strings():
+    df = pd.DataFrame(data=[test_element.copy(), test_element.copy()])
+    conformed_df = stager.conform_dataframe(df)
+    for column in ["version", "page_number", "regex_metadata"]:
+        assert conformed_df[column].apply(lambda x: isinstance(x, str)).all()

test/unit/connector/test_sql_conform_dict.py renamed to test/unit/connector/sql/test_sql_conform_dict.py

@@ -1 +1 @@
-__version__ = "0.3.13-dev0"  # pragma: no cover
+__version__ = "0.3.13-dev1"  # pragma: no cover

test/unit/connector/sql/test_sql_upload_stager.py

@@ -1,6 +1,5 @@
 from contextlib import contextmanager
 from dataclasses import dataclass, field
-from pathlib import Path
 from typing import TYPE_CHECKING, Generator, Optional
 
 import numpy as np
@@ -9,6 +8,7 @@
 
 from unstructured_ingest.utils.data_prep import split_dataframe
 from unstructured_ingest.utils.dep_check import requires_dependencies
+from unstructured_ingest.v2.interfaces.file_data import FileData
 from unstructured_ingest.v2.logger import logger
 from unstructured_ingest.v2.processes.connector_registry import (
     DestinationRegistryEntry,
@@ -160,9 +160,17 @@ class SnowflakeUploader(SQLUploader):
     connector_type: str = CONNECTOR_TYPE
     values_delimiter: str = "?"
 
-    def upload_contents(self, path: Path) -> None:
-        df = pd.read_json(path, orient="records", lines=True)
+    def upload_dataframe(self, df: pd.DataFrame, file_data: FileData) -> None:
+        if self.can_delete():
+            self.delete_by_record_id(file_data=file_data)
+        else:
+            logger.warning(
+                f"table doesn't contain expected "
+                f"record id column "
+                f"{self.upload_config.record_id_key}, skipping delete"
+            )
         df.replace({np.nan: None}, inplace=True)
+        self._fit_to_schema(df=df, columns=self.get_table_columns())
 
         columns = list(df.columns)
         stmt = "INSERT INTO {table_name} ({columns}) VALUES({values})".format(

unstructured_ingest/__version__.py

@@ -3,7 +3,7 @@
 from abc import ABC, abstractmethod
 from contextlib import contextmanager
 from dataclasses import dataclass, field
-from datetime import date, datetime
+from datetime import datetime
 from pathlib import Path
 from time import time
 from typing import Any, Generator, Union
@@ -92,7 +92,7 @@ class SqlBatchFileData(BatchFileData):
     additional_metadata: SqlAdditionalMetadata
 
 
-def parse_date_string(date_value: Union[str, int]) -> date:
+def parse_date_string(date_value: Union[str, int]) -> datetime:
     try:
         timestamp = float(date_value) / 1000 if isinstance(date_value, int) else float(date_value)
         return datetime.fromtimestamp(timestamp)
@@ -267,7 +267,7 @@ def conform_dict(self, element_dict: dict, file_data: FileData) -> dict:
 
     def conform_dataframe(self, df: pd.DataFrame) -> pd.DataFrame:
         for column in filter(lambda x: x in df.columns, _DATE_COLUMNS):
-            df[column] = df[column].apply(parse_date_string)
+            df[column] = df[column].apply(parse_date_string).apply(lambda date: date.timestamp())
         for column in filter(
             lambda x: x in df.columns,
             ("permissions_data", "record_locator", "points", "links"),