chore: change custom_labels type

Author: kitagry

gokart/gcs_obj_metadata_client.py

@@ -40,7 +40,7 @@ def _path_to_bucket_and_key(path: str) -> tuple[str, str]:
     def add_task_state_labels(
         path: str,
         task_params: dict[str, str] | None = None,
-        custom_labels: dict[str, Any] | None = None,
+        custom_labels: dict[str, str] | None = None,
         required_task_outputs: FlattenableItems[RequiredTaskOutput] | None = None,
     ) -> None:
         if GCSObjectMetadataClient._is_log_related_path(path):
@@ -90,7 +90,7 @@ def _normalize_labels(labels: dict[str, Any] | None) -> dict[str, str]:
     def _get_patched_obj_metadata(
         metadata: Any,
         task_params: dict[str, str] | None = None,
-        custom_labels: dict[str, Any] | None = None,
+        custom_labels: dict[str, str] | None = None,
         required_task_outputs: FlattenableItems[RequiredTaskOutput] | None = None,
     ) -> dict | Any:
         # If metadata from response when getting bucket and object information is not dictionary,

gokart/target.py

@@ -37,7 +37,7 @@ def dump(
         obj,
         lock_at_dump: bool = True,
         task_params: dict[str, str] | None = None,
-        custom_labels: dict[str, Any] | None = None,
+        custom_labels: dict[str, str] | None = None,
         required_task_outputs: FlattenableItems[RequiredTaskOutput] | None = None,
     ) -> None:
         if lock_at_dump:
@@ -120,7 +120,7 @@ def _dump(
         self,
         obj,
         task_params: dict[str, str] | None = None,
-        custom_labels: dict[str, Any] | None = None,
+        custom_labels: dict[str, str] | None = None,
         required_task_outputs: FlattenableItems[RequiredTaskOutput] | None = None,
     ) -> None:
         with self._target.open('w') as f:
@@ -172,7 +172,7 @@ def _dump(
         self,
         obj,
         task_params: dict[str, str] | None = None,
-        custom_labels: dict[str, Any] | None = None,
+        custom_labels: dict[str, str] | None = None,
         required_task_outputs: FlattenableItems[RequiredTaskOutput] | None = None,
     ) -> None:
         self._make_temporary_directory()