✨♻️ Enhance type hints and others (#41)

* stage1

* stage2

* Simplify _is_special

* Fix _get_params

* refine error message

* fix something

* fix

.github/workflows/codestyle-check.yml

@@ -9,15 +9,19 @@ on:
 jobs:
   code-style-check:
     runs-on: ubuntu-latest
-    name: CodeStyle Check (Python 3.8)
+    strategy:
+      matrix:
+        python-version: [3.8, 3.9, '3.10']
+
+    name: CodeStyle Check ${{ matrix.python-version }}
     steps:
       - name: Checkout
         uses: actions/checkout@v3
 
       - name: Install python
         uses: actions/setup-python@v4
         with:
-          python-version: 3.8
+          python-version: ${{ matrix.python-version }}
 
       - name: Install dependencies
         run: |

.github/workflows/mypy.yaml

@@ -0,0 +1,35 @@
+name: MYPY Check
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+    branches: [main]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: [3.8, 3.9, '3.10']
+
+    steps:
+      - uses: actions/checkout@v2
+        with:
+          fetch-depth: 1
+
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v2
+        with:
+          python-version: ${{ matrix.python-version }}
+
+      - name: Install Poetry
+        uses: snok/install-poetry@v1
+        with:
+          version: 1.6.1
+
+      - name: Install Dependencies
+        run: poetry install --with dev
+
+      - name: Mypy Check
+        run: poetry run mypy excore

.github/workflows/tests.yaml

@@ -31,12 +31,17 @@ jobs:
       - name: Install Dependencies
         run: poetry install --with dev
 
-      - name: Test with pytest
+      - name: Initialize ExCore
         run: |
           cd ./tests
           export EXCORE_DEBUG=1
           poetry run python init.py
+
+      - name: Test with pytest
+        run: |
+          cd ./tests
           poetry run pytest --cov=../excore
+          poetry run python init.py
           poetry run pytest test_config.py
           poetry run pytest test_config.py
           poetry run pytest test_config.py

excore/__init__.py

@@ -3,7 +3,7 @@
 from rich.traceback import install
 
 from . import config, plugins
-from ._constants import __author__, __version__, _load_workspace_config, _workspace_cfg
+from ._constants import __author__, __version__, workspace
 from .config.action import DictAction
 from .config.config import build_all, load
 from .config.parse import set_primary_fields
@@ -43,7 +43,6 @@
 
 install()
 init_logger()
-_load_workspace_config()
-set_primary_fields(_workspace_cfg)
+set_primary_fields(workspace)
 _enable_excore_debug()
-sys.path.append(_workspace_cfg["base_dir"])
+sys.path.append(workspace.base_dir)

excore/_constants.py

@@ -1,60 +1,80 @@
+from __future__ import annotations
+
 import os
 import os.path as osp
+from dataclasses import dataclass, field
+from typing import Any
+
+import toml
 
 from .engine.logging import logger
 
 __author__ = "Asthestarsfalll"
 __version__ = "0.1.1beta"
 
-_cache_base_dir = osp.expanduser("~/.cache/excore/")
 _workspace_config_file = "./.excore.toml"
 _registry_cache_file = "registry_cache.pkl"
 _json_schema_file = "excore_schema.json"
 _class_mapping_file = "class_mapping.json"
 
 
-def _load_workspace_config():
-    if osp.exists(_workspace_config_file):
-        _workspace_cfg.update(toml.load(_workspace_config_file))
-        logger.ex("load `.excore.toml`")
-    else:
-        logger.warning("Please use `excore init` in your command line first")
+@dataclass
+class _WorkspaceConfig:
+    name: str = field(default="")
+    src_dir: str = field(default="")
+    base_dir: str = field(default="")
+    cache_base_dir: str = field(default=osp.expanduser("~/.cache/excore/"))
+    cache_dir: str = field(default="")
+    registry_cache_file: str = field(default="")
+    json_schema_file: str = field(default="")
+    class_mapping_file: str = field(default="")
+    registries: list[str] = field(default_factory=list)
+    primary_fields: list[str] = field(default_factory=list)
+    primary_to_registry: dict[str, str] = field(default_factory=dict)
+    json_schema_fields: dict[str, str | list[str]] = field(default_factory=dict)
+    props: dict[Any, Any] = field(default_factory=dict)
 
+    @property
+    def base_name(self):
+        return osp.split(self.cache_dir)[-1]
 
-def _update_name(base_name):
-    name = base_name
+    def __post_init__(self) -> None:
+        if not osp.exists(_workspace_config_file):
+            self.base_dir = os.getcwd()
+            self.cache_dir = self._get_cache_dir()
+            self.registry_cache_file = osp.join(self.cache_dir, _registry_cache_file)
+            self.json_schema_file = osp.join(self.cache_dir, _json_schema_file)
+            self.class_mapping_file = osp.join(self.cache_dir, _class_mapping_file)
+            logger.warning("Please use `excore init` in your command line first")
+        else:
+            self.update(toml.load(_workspace_config_file))
 
-    suffix = 1
-    while osp.exists(osp.join(_cache_base_dir, name)):
-        name = f"{_base_name}_{suffix}"
-        suffix += 1
+    def _get_cache_dir(self) -> str:
+        base_name = osp.basename(osp.normpath(os.getcwd()))
+        base_name = self._update_name(base_name)
+        return osp.join(self.cache_base_dir, base_name)
 
-    return name
+    def _update_name(self, base_name: str) -> str:
+        name = base_name
 
+        suffix = 1
+        while osp.exists(osp.join(self.cache_base_dir, name)):
+            name = f"{base_name}_{suffix}"
+            suffix += 1
 
-if not osp.exists(_workspace_config_file):
-    _base_name = osp.basename(osp.normpath(os.getcwd()))
-    _base_name = _update_name(_base_name)
-else:
-    import toml  # pylint: disable=import-outside-toplevel
+        return name
 
-    cfg = toml.load(_workspace_config_file)
-    _base_name = cfg["name"]
+    def update(self, _cfg: dict[Any, Any]) -> None:
+        self.__dict__.update(_cfg)
 
-_cache_dir = osp.join(_cache_base_dir, _base_name)
+    def dump(self, path: str) -> None:
+        with open(path, "w") as f:
+            cfg = self.__dict__
+            cfg.pop("base_dir", None)
+            toml.dump(cfg, f)
 
-# TODO: Use a data class to store this
-_workspace_cfg = dict(
-    name="",
-    src_dir="",
-    base_dir=os.getcwd(),
-    registries=[],
-    primary_fields=[],
-    primary_to_registry={},
-    json_schema_fields={},
-    props={},
-)
 
+workspace = _WorkspaceConfig()
 LOGO = r"""
 ▓█████ ▒██   ██▒ ▄████▄   ▒█████   ██▀███  ▓█████
 ▓█   ▀ ▒▒ █ █ ▒░▒██▀ ▀█  ▒██▒  ██▒▓██ ▒ ██▒▓█   ▀

excore/_exceptions.py

@@ -56,3 +56,7 @@ class HookManagerBuildError(BaseException):
 
 class HookBuildError(BaseException):
     pass
+
+
+class AnnotationsFutureError(Exception):
+    pass

excore/_misc.py

@@ -1,14 +1,15 @@
+from __future__ import annotations
+
 import functools
-import threading
-import time
+from typing import Any, Callable, Sequence
 
 from tabulate import tabulate
 
 
 class CacheOut:
-    def __call__(self, func):
+    def __call__(self, func: Callable[..., Any]):
         @functools.wraps(func)
-        def _cache(self):
+        def _cache(self) -> Any:
             if not hasattr(self, "cached_elem"):
                 cached_elem = func(self)
                 if cached_elem != self:
@@ -19,27 +20,14 @@ def _cache(self):
         return _cache
 
 
-class FileLock:
-    def __init__(self, file_path, timeout=15):
-        self.file_path = file_path
-        self.timeout = timeout
-        self.lock = threading.Lock()
-
-    def __enter__(self):
-        start_time = time.time()
-        while not self.lock.acquire(False):
-            if time.time() - start_time >= self.timeout:
-                raise TimeoutError("Failed to acquire lock on file")
-            time.sleep(0.1)
-        return self
-
-    def __exit__(self, exc_type, exc_val, exc_tb):
-        self.lock.release()
-
-
-def _create_table(header, contents, split=True, prefix="\n", **tabel_kwargs):
-    if split:
-        contents = [(i,) for i in contents]
+def _create_table(
+    header: str | list[str] | tuple[str, ...] | None,
+    contents: Sequence[str] | Sequence[Sequence[str]],
+    prefix: str = "\n",
+    **tabel_kwargs: Any,
+) -> str:
+    if len(contents) > 0 and isinstance(contents[0], str):
+        contents = [(i,) for i in contents]  # type: ignore
     if header is None:
         header = ()
     if not isinstance(header, (list, tuple)):

excore/cli/_cache.py

@@ -1,14 +1,17 @@
+from __future__ import annotations
+
 import os
 
 import typer
 
-from .._constants import _base_name, _cache_base_dir, _cache_dir
+from excore import workspace
+
 from .._misc import _create_table
 from ..engine.logging import logger
 from ._app import app
 
 
-def _clear_cache(cache_dir):
+def _clear_cache(cache_dir: str) -> None:
     if os.path.exists(cache_dir):
         import shutil  # pylint: disable=import-outside-toplevel
 
@@ -19,38 +22,41 @@ def _clear_cache(cache_dir):
 
 
 @app.command()
-def clear_cache():
+def clear_cache() -> None:
     """
     Remove the cache folder which belongs to current workspace.
     """
-    if not typer.confirm(f"Are you sure you want to clear cache of {_base_name}?"):
+    if not typer.confirm(
+        f"Are you sure you want to clear cache of {workspace.name}?"
+        f" Cache dir is {workspace.cache_dir}."
+    ):
         return
-
-    target = os.path.join(_cache_dir, _base_name)
-    _clear_cache(target)
+    _clear_cache(workspace.cache_dir)
 
 
 @app.command()
-def clear_all_cache():
+def clear_all_cache() -> None:
     """
     Remove the whole cache folder.
     """
+    if not os.path.exists(workspace.cache_base_dir):
+        logger.warning("Cache dir {} does not exist", workspace.cache_base_dir)
+        return
+    print(_create_table("Cache Names", os.listdir(workspace.cache_base_dir)))
     if not typer.confirm("Are you sure you want to clear all cache?"):
         return
-    _clear_cache(_cache_base_dir)
+    _clear_cache(workspace.cache_base_dir)
 
 
 @app.command()
-def cache_list():
+def cache_list() -> None:
     """
     Show cache folders.
     """
-    tabel = _create_table("NAMES", os.listdir(_cache_base_dir))
+    tabel = _create_table("Cache Names", os.listdir(workspace.cache_base_dir))
     logger.info(tabel)
 
 
 @app.command()
-def cache_dir():
-    # if not os.path.exists(_workspace_config_file):
-    #     raise RuntimeError("Not in ExCore project")
-    print(_cache_dir)
+def cache_dir() -> None:
+    print(workspace.cache_dir)