add utils

Author: iamtatsuki05

pyproject.toml

@@ -25,6 +25,7 @@ dependencies = [
     "aiohttp>=3.9.5",
     "tenacity>=8.4.1",
     "playwright>=1.44.0",
+    "toml>=0.10.2",
 ]
 
 [tool.hatch.build.targets.wheel]
@@ -40,6 +41,8 @@ dev-dependencies = [
     "pytest>=8.3.3",
     "ruff>=0.6.9",
     "tox>=4.21.2",
+    "types-pyyaml>=6.0.12.20250402",
+    "types-toml>=0.10.8.20240310",
 ]
 
 [tool.ruff]

src/project/common/utils/async_utils.py

@@ -0,0 +1,42 @@
+import asyncio
+from typing import Any, Callable
+
+
+def sync_to_async_func(sync_func: Callable) -> Callable:
+    """
+    同期関数を非同期関数として使えるように変換する
+    """
+
+    async def wrapper(*args: Any, **kwargs: Any) -> Any:
+        return await asyncio.to_thread(sync_func, *args, **kwargs)
+
+    wrapper.__name__ = sync_func.__name__
+    wrapper.__doc__ = sync_func.__doc__
+    return wrapper
+
+
+def async_to_sync_func(async_func: Callable) -> Callable:
+    """
+    非同期関数を同期関数として使えるように変換する
+    """
+
+    def wrapper(*args: Any, **kwargs: Any) -> Any:
+        return asyncio.run(async_func(*args, **kwargs))
+
+    wrapper.__name__ = async_func.__name__
+    wrapper.__doc__ = async_func.__doc__
+    return wrapper
+
+
+async def run_async_function_with_semaphore(
+    async_func: Callable, concurrency_sema: asyncio.Semaphore | None, *args: Any, **kwargs: Any
+) -> Any:
+    """
+    指定した関数 func を、セマフォで同時実行数を制限して呼び出す関数。
+    concurrency_sema が None の場合は制限しない。
+    """
+    if concurrency_sema is not None:
+        async with concurrency_sema:
+            return await async_func(*args, **kwargs)
+    else:
+        return await async_func(*args, **kwargs)

src/project/common/utils/file/__init__.py

@@ -0,0 +1,21 @@
+from pathlib import Path
+from typing import Any
+
+from project.common.utils.file.json import load_json
+from project.common.utils.file.toml import load_toml
+from project.common.utils.file.yaml import load_yaml
+
+
+def load_config(path: str | Path) -> dict[str, Any]:
+    """
+    Load configuration from a file (JSON, YAML, or TOML).
+    """
+    ext = Path(path).suffix.lower()
+    if ext == '.json':
+        return load_json(path)
+    elif ext in ('.yaml', '.yml'):
+        return load_yaml(path)
+    elif ext == '.toml':
+        return load_toml(path)
+    else:
+        raise ValueError(f'Unsupported config file format: {ext}')

src/project/common/utils/file/config.py

@@ -0,0 +1,21 @@
+import json
+from pathlib import Path
+
+
+def load_json(path: str | Path) -> dict | list:
+    with Path(path).open(mode='r', encoding='utf-8') as fin:
+        data = json.load(fin)
+    return data
+
+
+def save_as_indented_json(
+    data: dict | list,
+    path: str | Path,
+    parents: bool = True,
+    exist_ok: bool = True,
+) -> None:
+    path = Path(path)
+    path.parent.mkdir(parents=parents, exist_ok=exist_ok)
+    with path.open(mode='w', encoding='utf-8') as fout:
+        json.dump(data, fout, ensure_ascii=False, indent=4, separators=(',', ': '))
+    return

src/project/common/utils/file/json.py

@@ -0,0 +1,25 @@
+from pathlib import Path
+
+import jsonlines
+
+
+def load_jsonlines(path: str | Path) -> list[dict]:
+    data_list = []
+    with jsonlines.open(str(path)) as reader:
+        for data in reader:
+            data_list.append(data)
+    return data_list
+
+
+def save_as_jsonlines(
+    data: list[dict],
+    path: str | Path,
+    parents: bool = True,
+    exist_ok: bool = True,
+) -> None:
+    path = Path(path)
+    path.parent.mkdir(parents=parents, exist_ok=exist_ok)
+    with jsonlines.open(str(path), mode='w') as writer:
+        for datum in data:
+            writer.write(datum)
+    return

src/project/common/utils/file/jsonlines.py

@@ -0,0 +1,22 @@
+from pathlib import Path
+
+import toml
+
+
+def load_toml(path: str | Path) -> dict:
+    with Path(path).open(mode='r', encoding='utf-8') as fin:
+        data = toml.load(fin)
+    return data
+
+
+def save_as_toml(
+    data: dict,
+    path: str | Path,
+    parents: bool = True,
+    exist_ok: bool = True,
+) -> None:
+    path = Path(path)
+    path.parent.mkdir(parents=parents, exist_ok=exist_ok)
+    with path.open(mode='w', encoding='utf-8') as fout:
+        toml.dump(data, fout)
+    return

src/project/common/utils/file/toml.py

@@ -0,0 +1,22 @@
+from pathlib import Path
+
+import yaml
+
+
+def load_yaml(path: str | Path) -> dict | list:
+    with Path(path).open(mode='r', encoding='utf-8') as fin:
+        data = yaml.safe_load(fin)
+    return data
+
+
+def save_as_indented_yaml(
+    data: dict | list,
+    path: str | Path,
+    parents: bool = True,
+    exist_ok: bool = True,
+) -> None:
+    path = Path(path)
+    path.parent.mkdir(parents=parents, exist_ok=exist_ok)
+    with path.open(mode='w', encoding='utf-8') as fout:
+        yaml.dump(data, fout, allow_unicode=True, indent=4, default_flow_style=False)
+    return

src/project/common/utils/file/yaml.py

@@ -0,0 +1,22 @@
+import importlib
+import inspect
+import sys
+from pathlib import Path
+from typing import Callable
+
+
+def import_function(function_file_path: str, function_name: str | None = None) -> Callable:
+    function_file_path = Path(function_file_path).resolve()
+    function_name = function_name or function_file_path.stem
+
+    project_root = Path.cwd()
+    if str(project_root) not in sys.path:
+        sys.path.append(str(project_root))
+
+    module_path = '.'.join(function_file_path.relative_to(project_root).with_suffix('').parts)
+    module = importlib.import_module(module_path)
+    return getattr(module, function_name)
+
+
+def get_imported_function_path(obj: Callable) -> str:
+    return inspect.getfile(obj)