fix(data-warehouse): A couple of small fixes for management command (#…

…26879)
PostHog · Dec 12, 2024 · 95713b3 · 95713b3
1 parent 0c6535e
commit 95713b3
Show file tree

Hide file tree

Showing 3 changed files with 17 additions and 2 deletions.
diff --git a/mypy-baseline.txt b/mypy-baseline.txt
@@ -149,6 +149,8 @@ posthog/hogql_queries/legacy_compatibility/filter_to_query.py:0: error: Argument
 posthog/hogql_queries/legacy_compatibility/filter_to_query.py:0: error: Dict entry 0 has incompatible type "str": "PathsFilter"; expected "str": "TrendsFilter"  [dict-item]
 posthog/hogql_queries/legacy_compatibility/filter_to_query.py:0: error: Dict entry 0 has incompatible type "str": "LifecycleFilter"; expected "str": "TrendsFilter"  [dict-item]
 posthog/hogql_queries/legacy_compatibility/filter_to_query.py:0: error: Dict entry 0 has incompatible type "str": "StickinessFilter"; expected "str": "TrendsFilter"  [dict-item]
+posthog/warehouse/models/external_data_schema.py:0: error: Incompatible types in assignment (expression has type "str", variable has type "int | float")  [assignment]
+posthog/warehouse/models/external_data_schema.py:0: error: Incompatible types in assignment (expression has type "str", variable has type "int | float")  [assignment]
 posthog/session_recordings/models/session_recording.py:0: error: Argument "distinct_id" to "MissingPerson" has incompatible type "str | None"; expected "str"  [arg-type]
 posthog/session_recordings/models/session_recording.py:0: error: Incompatible type for lookup 'persondistinctid__team_id': (got "Team", expected "str | int")  [misc]
 posthog/models/hog_functions/hog_function.py:0: error: Argument 1 to "get" of "dict" has incompatible type "str | None"; expected "str"  [arg-type]

diff --git a/posthog/management/commands/warehouse_last_incremental_value_sync.py b/posthog/management/commands/warehouse_last_incremental_value_sync.py
@@ -128,8 +128,16 @@ def handle(self, *args, **options):
                 last_incremental_value = incremental.get("last_value")
             except Exception as e:
                 print(f"Cant get last_incremental_value for schema: {schema.pk}. ERROR: {e}")  # noqa: T201
+                pipeline.drop()
                 continue
 
-            schema.update_incremental_field_last_value(last_incremental_value)
+            try:
+                schema.update_incremental_field_last_value(last_incremental_value)
+            except Exception as e:
+                print(  # noqa: T201
+                    f"Cant update_incremental_field_last_value for schema: {schema.pk}. With last_incremental_value={last_incremental_value}. ERROR: {e}"
+                )
+                pipeline.drop()
+                continue
 
             pipeline.drop()
diff --git a/posthog/warehouse/models/external_data_schema.py b/posthog/warehouse/models/external_data_schema.py
@@ -84,7 +84,12 @@ def update_incremental_field_last_value(self, last_value: Any) -> None:
             incremental_field_type == IncrementalFieldType.Integer
             or incremental_field_type == IncrementalFieldType.Numeric
         ):
-            last_value_json = last_value_py
+            if isinstance(last_value_py, int | float):
+                last_value_json = last_value_py
+            elif isinstance(last_value_py, datetime):
+                last_value_json = str(last_value_py)
+            else:
+                last_value_json = int(last_value_py)
         else:
             last_value_json = str(last_value_py)