chore(client): add typing hint for python lambda (#945)

add typing hint for python lambda

client/starwhale/api/_impl/model.py

@@ -31,16 +31,20 @@
 
 _TASK_ROOT_DIR = "/var/starwhale" if in_production() else "/tmp/starwhale"
 
-_p = lambda p, sub: Path(p) if p else Path(_TASK_ROOT_DIR) / sub
 _ptype = t.Union[str, None, Path]
+_p: t.Callable[[_ptype, str], Path] = (
+    lambda p, sub: Path(p) if p else Path(_TASK_ROOT_DIR) / sub
+)
 
 
 class _LogType:
     SW = "starwhale"
     USER = "user"
 
 
-_jl_writer = lambda p: jsonlines.open(str((p).resolve()), mode="w")
+_jl_writer: t.Callable[[Path], jsonlines.Writer] = lambda p: jsonlines.open(
+    str((p).resolve()), mode="w"
+)
 
 
 class _RunConfig:
@@ -52,8 +56,8 @@ def __init__(
         status_dir: _ptype = "",
         log_dir: _ptype = "",
     ) -> None:
-        self.status_dir = _p(status_dir, "status")  # type: ignore
-        self.log_dir = _p(log_dir, "log")  # type: ignore
+        self.status_dir = _p(status_dir, "status")
+        self.log_dir = _p(log_dir, "log")
         # TODO: refactor dataset arguments
         self.dataset_uri = URI(dataset_uri, expected_type=URIType.DATASET)
         self.dataset_row_end = dataset_row_end
@@ -124,7 +128,7 @@ def __init__(
         self._orig_stderr = sys.stderr
 
         # TODO: split status/result files
-        self._timeline_writer = _jl_writer(self.status_dir / "timeline")  # type: ignore
+        self._timeline_writer = _jl_writer(self.status_dir / "timeline")
 
         self._ppl_data_field = "result"
         self._label_field = "label"
@@ -258,7 +262,7 @@ def _wrapper(*args: t.Any, **kwargs: t.Any) -> None:
     def _starwhale_internal_run_cmp(self) -> None:
         self._sw_logger.debug("enter cmp func...")
         self._update_status(self.STATUS.START)
-        now = now_str()  # type: ignore
+        now = now_str()
         try:
             _ppl_results = list(self.evaluation.get_results())
             self._sw_logger.debug("cmp input data size:{}", len(_ppl_results))
@@ -347,7 +351,7 @@ def _do_record(
         *args: t.Any,
     ) -> None:
         _timeline = {
-            "time": now_str(),  # type: ignore
+            "time": now_str(),
             "status": exception is None,
             "exception": str(exception),
             "index": data.idx,

client/starwhale/base/bundle.py

@@ -132,7 +132,7 @@ def _gen_version(self) -> None:
 
         self.uri.object.version = self._version  # type:ignore
         self._manifest["version"] = self._version  # type: ignore
-        self._manifest["created_at"] = now_str()  # type: ignore
+        self._manifest["created_at"] = now_str()
         logger.info(f"[step:version]version: {self._version}")
         console.print(f":new: version {self._version[:SHORT_VERSION_CNT]}")  # type: ignore
 

client/starwhale/base/tag.py

@@ -48,7 +48,7 @@ def _get_manifest(self) -> t.Dict[str, t.Any]:
             return load_yaml(self._manifest_path)  # type: ignore
 
     def _save_manifest(self, _manifest: t.Dict[str, t.Any]) -> None:
-        _manifest["updated_at"] = now_str()  # type: ignore
+        _manifest["updated_at"] = now_str()
         _manifest["name"] = self.uri.object.name
         _manifest["typ"] = self.uri.object.typ
 

client/starwhale/base/uri.py

@@ -241,9 +241,9 @@ def capsulate_uri_str(
         obj_name: str = "",
         obj_ver: str = "",
     ) -> str:
-        _fmt = lambda x: x.strip().strip("/").lower()
+        _fmt: t.Callable[[str], str] = lambda x: x.strip().strip("/").lower()
 
-        _rt = f"{_fmt(instance)}"  # type: ignore
+        _rt = f"{_fmt(instance)}"
         if project:
             _rt = f"{_rt}/project/{project}"
         else:

client/starwhale/base/view.py

@@ -235,7 +235,7 @@ def list_data(
         return sort_obj_list(result, order_keys)
 
     @staticmethod
-    def place_holder_for_empty(place_holder: str = "--") -> t.Callable:
+    def place_holder_for_empty(place_holder: str = "--") -> t.Callable[[str], str]:
         return lambda x: x or place_holder
 
     @staticmethod

client/starwhale/core/eval/executor.py

@@ -291,7 +291,7 @@ def _render_manifest(self) -> None:
                 datasets=[u.full_uri for u in self.dataset_uris],
                 runtime=self.runtime_uri,
                 status=_STATUS.SUCCESS if _status else _STATUS.FAILED,
-                finished_at=now_str(),  # type: ignore
+                finished_at=now_str(),
             )
         )
         _f = self._workdir / DEFAULT_MANIFEST_NAME

client/starwhale/core/project/view.py

@@ -145,7 +145,9 @@ def list(
             "location": cls.place_holder_for_empty(),
             "owner": cls.place_holder_for_empty(),
         }
-        custom_row = lambda row: {"style": "magenta"} if row["in_use"] else None
+        custom_row: t.Callable[[t.Dict[str, t.Any]], t.Optional[t.Dict[str, str]]] = (
+            lambda row: {"style": "magenta"} if row["in_use"] else None
+        )
         cls.print_table(
             title, projects, custom_column=custom_column, custom_row=custom_row
         )

client/starwhale/utils/__init__.py

@@ -20,7 +20,9 @@
 from starwhale.utils.error import NoSupportError
 
 console = Console(soft_wrap=True)
-now_str = lambda: datetime.now().astimezone().strftime(FMT_DATETIME)
+now_str: t.Callable[[], str] = (
+    lambda: datetime.now().astimezone().strftime(FMT_DATETIME)
+)
 
 
 def timestamp_to_datatimestr(timestamp: float) -> str:

client/starwhale/utils/config.py

@@ -71,7 +71,7 @@ def render_default_swcli_config(fpath: str) -> t.Dict[str, t.Any]:
                 user_name=_CURRENT_SHELL_USERNAME,
                 current_project=DEFAULT_PROJECT,
                 type=InstanceType.STANDALONE,
-                updated_at=now_str(),  # type: ignore
+                updated_at=now_str(),
             )
         },
         current_instance=DEFAULT_INSTANCE,
@@ -226,7 +226,7 @@ def update_instance(
             user_role=user_role,
             sw_token=sw_token,
             type=InstanceType.CLOUD,
-            updated_at=now_str(),  # type: ignore
+            updated_at=now_str(),
         )
 
         update_swcli_config(**self._config)

client/starwhale/utils/http.py

@@ -19,12 +19,12 @@ def wrap_sw_error_resp(
 ) -> None:
 
     if silent:
-        _rprint = lambda x: x
+        _rprint: t.Callable = lambda x: x
     else:
         _rprint = rprint
 
     if r.status_code == HTTPStatus.OK:
-        _rprint(f":clap: {header} success")  # type: ignore
+        _rprint(f":clap: {header} success")
         return
 
     msg = f":disappointed_face: url:{r.url}\n:dog: http status code: {r.status_code} \n"