fix clin issue when sql-query already exists (#117)

* fix sql-query issue when sql-query already exists * make the compaction key optional * clin reformatting * remove the of conditions and use get to return the default value Co-authored-by: Amr Abdelshafi <[email protected]>
zalando-incubator · Jul 19, 2022 · 88d501d · 88d501d
1 parent 2636696
commit 88d501d
Show file tree

Hide file tree

Showing 3 changed files with 12 additions and 5 deletions.
diff --git a/clin/clients/nakadi_sql.py b/clin/clients/nakadi_sql.py
@@ -81,6 +81,7 @@ def convert_repartitioning() -> Optional[Partitioning]:
             policy=Cleanup.Policy(payload["cleanup_policy"]),
             retention_time_days=payload.get("retention_time", 0) // MS_IN_DAY,
         ),
+        partition_compaction_key_field=payload.get("partition_compaction_key_field"),
     )
 
 
@@ -115,7 +116,6 @@ def sql_query_to_payload(sql_query: SqlQuery) -> dict:
             "cleanup_policy": str(sql_query.output_event_type.cleanup.policy),
             "retention_time": sql_query.output_event_type.cleanup.retention_time_days
             * MS_IN_DAY,
-            "partition_compaction_key_field": sql_query.output_event_type.partition_compaction_key_field,
         },
         "authorization": auth_to_payload(sql_query.auth),
     }
@@ -129,4 +129,9 @@ def sql_query_to_payload(sql_query: SqlQuery) -> dict:
             "partition_key_fields": sql_query.output_event_type.repartitioning.keys,
         }
 
+    if sql_query.output_event_type.partition_compaction_key_field:
+        payload["output_event_type"][
+            "partition_compaction_key_field"
+        ] = sql_query.output_event_type.partition_compaction_key_field
+
     return payload
diff --git a/clin/models/sql_query.py b/clin/models/sql_query.py
@@ -16,7 +16,7 @@ class OutputEventType:
     audience: Audience
     repartitioning: Optional[Partitioning]
     cleanup: Cleanup
-    partition_compaction_key_field: str
+    partition_compaction_key_field: Optional[str]
 
     @staticmethod
     def from_spec(spec: dict[str, any]):
@@ -28,7 +28,7 @@ def from_spec(spec: dict[str, any]):
             if "repartitioning" in spec
             else None,
             cleanup=Cleanup.from_spec(spec["cleanup"]),
-            partition_compaction_key_field=spec["partitionCompactionKeyField"],
+            partition_compaction_key_field=spec.get("partitionCompactionKeyField"),
         )
 
     def to_spec(self) -> dict[str, any]:
@@ -37,12 +37,14 @@ def to_spec(self) -> dict[str, any]:
             "owningApplication": self.owning_application,
             "audience": str(self.audience),
             "cleanup": self.cleanup.to_spec(),
-            "partitionCompactionKeyField": self.partition_compaction_key_field,
         }
 
         if self.repartitioning:
             spec["repartitioning"] = self.repartitioning.to_spec()
 
+        if self.partition_compaction_key_field:
+            spec["partitionCompactionKeyField"] = self.partition_compaction_key_field
+
         return spec
 
 

diff --git a/docs/examples/single/sql-query.yaml b/docs/examples/single/sql-query.yaml
@@ -13,7 +13,7 @@ spec:
     cleanup:
       policy: delete
       retentionTimeDays: 2
-    partitionCompactionKeyField": product_logistics.metadata.partition_compaction_key
+    partitionCompactionKeyField: product_logistics.metadata.partition_compaction_key
     repartitioning:
       strategy: hash
       keys: