feat: Enhance project structure and add utility functions

- Refactored main entry point in `main.py` to use logging instead of print statements.
- Introduced regex utilities in `regex.py` for email, URL, and time pattern matching.
- Updated async utilities in `async_utils.py` to include an `AsyncResource` base class for managing concurrency.
- Added CLI configuration loading and parsing functions in `cli_utils.py` using Pydantic for type safety.
- Implemented file handling utilities for JSON, TOML, YAML, and JSON Lines formats.
- Created comprehensive unit tests for JSON, TOML, YAML, and async utilities.
- Updated `tox.ini` to specify directories for linting and type checking.

Author: iamtatsuki05

scripts/main.py

@@ -1,2 +1,16 @@
+import logging
+
+from project.env import PACKAGE_DIR
+
+logger = logging.getLogger(__name__)
+
+
+def main() -> None:
+    """Sample entry point."""
+    logging.basicConfig(level=logging.INFO)
+    logger.info('Hello, World!')
+    logger.info('PACKAGE_DIR=%s', PACKAGE_DIR)
+
+
 if __name__ == '__main__':
-    print('Hello, World!')
+    main()

src/project/common/regex.py

@@ -0,0 +1,46 @@
+import re
+from typing import Final
+
+from project.common.utils.regex_utils import concat, unmatched_group
+
+# time regex
+TIME_PATTRN: Final[re.Pattern] = re.compile(r'\d+:\d+')
+
+
+# email regex
+LOCAL_PART_CHARS = r'[\w\-._]'
+DOMAIN_CHARS = r'[\w\-._]'
+TLD_CHARS = r'[A-Za-z]'
+
+local_part = concat([LOCAL_PART_CHARS], without_grouping=True) + r'+'
+domain = concat([DOMAIN_CHARS], without_grouping=True) + r'+'
+tld = concat([TLD_CHARS], without_grouping=True) + r'+'
+
+EMAIL_REGEX = local_part + r'@' + domain + r'\.' + tld
+EMAIL_PATTERN = re.compile(EMAIL_REGEX)
+
+
+# url regex
+SCHEME = r'https?'
+CHARS = r'[\w!?/+\-_~;.,*&@#$%()\[\]]'
+
+url_chars = concat([CHARS], without_grouping=True) + r'+'
+
+HTTP_URL_REGEX = SCHEME + r'://' + url_chars
+
+DATA_SCHEME = r'data:'
+MEDIATYPE = r'[\w/+.-]+'
+BASE64 = r'base64'
+DATA = r'[\w+/=]+'
+
+mediatype_part = unmatched_group(MEDIATYPE) + r'?'
+base64_part = unmatched_group(BASE64) + r'?'
+data_part = unmatched_group(DATA)
+
+DATA_URL_REGEX = DATA_SCHEME + mediatype_part + r'(?:;' + base64_part + r')?,' + data_part
+
+URL_REGEX = concat([HTTP_URL_REGEX, DATA_URL_REGEX])
+
+HTTP_URL_PATTERN = re.compile(HTTP_URL_REGEX)
+DATA_URL_PATTERN = re.compile(DATA_URL_REGEX)
+URL_PATTERN = re.compile(URL_REGEX)

src/project/common/utils/async_utils.py

@@ -1,42 +1,55 @@
 import asyncio
-from typing import Any, Callable
+from abc import ABC, abstractmethod
+from collections.abc import Awaitable, Callable, Coroutine
+from typing import Any
 
 
-def sync_to_async_func(sync_func: Callable) -> Callable:
-    """
-    同期関数を非同期関数として使えるように変換する
-    """
+def sync_to_async_func[R](sync_func: Callable[..., R]) -> Callable[..., Awaitable[R]]:
+    """Convert a synchronous callable into an asynchronous callable."""
 
-    async def wrapper(*args: Any, **kwargs: Any) -> Any:
+    async def wrapper(*args: object, **kwargs: object) -> R:
         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 async_to_sync_func[R](async_func: Callable[..., Coroutine[Any, Any, R]]) -> Callable[..., R]:
+    """Convert an asynchronous callable into a synchronous callable."""
 
-    def wrapper(*args: Any, **kwargs: Any) -> Any:
+    def wrapper(*args: object, **kwargs: object) -> R:
         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 の場合は制限しない。
-    """
+async def run_async_function_with_semaphore[R](
+    async_func: Callable[..., Awaitable[R]],
+    concurrency_sema: asyncio.Semaphore | None,
+    *args: object,
+    **kwargs: object,
+) -> R:
+    """Execute async_func with an optional semaphore limiting concurrency."""
     if concurrency_sema is not None:
         async with concurrency_sema:
             return await async_func(*args, **kwargs)
-    else:
-        return await async_func(*args, **kwargs)
+    return await async_func(*args, **kwargs)
+
+
+class AsyncResource[R](ABC):
+    """Base class for async resources protected by a semaphore."""
+
+    def __init__(self, concurrency: int = 1) -> None:
+        self.semaphore = asyncio.Semaphore(concurrency)
+
+    async def task(self, *args: object, **kwargs: object) -> R:
+        async with self.semaphore:
+            return await self.call(*args, **kwargs)
+
+    @abstractmethod
+    async def call(self, *args: object, **kwargs: object) -> R:
+        """Execute the concrete asynchronous operation."""
+        raise NotImplementedError

src/project/common/utils/cli_utils.py

@@ -2,22 +2,58 @@
 from pathlib import Path
 from typing import Any
 
+from pydantic import BaseModel
+
 from project.common.utils.file.config import load_config
 
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
 
-def load_cli_config(config_file_path: str | Path | None = None, **kwargs: Any) -> dict[str, Any]:
-    """
-    Load configuration from a file and merge it with runtime arguments.
-    """
+def load_cli_config(config_file_path: str | Path | None = None, **kwargs: object) -> dict[str, Any]:
+    """Load configuration from a file and merge it with runtime arguments."""
     if config_file_path:
-        logger.info(f'Loading configuration from {config_file_path}')
+        logger.info('Loading configuration from %s', config_file_path)
         merged = load_config(config_file_path)
         merged.update(kwargs)
-        logger.info(f'Merged config from file with overrides: {kwargs.keys() if kwargs else {}}')
+        logger.info('Merged config with overrides: %s', list(kwargs) if kwargs else [])
     else:
-        merged = kwargs
+        merged = dict(kwargs)
         logger.info('No config file provided; using runtime arguments only.')
     return merged
+
+
+def load_and_parse_config[T: BaseModel](
+    config_class: type[T],
+    config_file_path: str | Path | None = None,
+    **kwargs: object,
+) -> T:
+    """Load configuration from file, merge with kwargs, and parse into Pydantic model.
+
+    This function provides type-safe configuration loading by:
+    1. Loading config from file (if provided)
+    2. Merging with CLI overrides
+    3. Validating and parsing into the specified Pydantic model
+
+    Args:
+        config_class: Pydantic BaseModel subclass to parse into
+        config_file_path: Path to config file (JSON/YAML/TOML)
+        **kwargs: CLI overrides to merge with file config
+
+    Returns:
+        Validated instance of config_class
+
+    Raises:
+        ValidationError: If configuration is invalid
+
+    Example:
+        >>> from pydantic import BaseModel, Field
+        >>> class MyConfig(BaseModel):
+        ...     name: str = Field(...)
+        ...     value: int = Field(default=0)
+        >>> cfg = load_and_parse_config(MyConfig, 'config.json', value=42)
+        >>> assert isinstance(cfg, MyConfig)
+
+    """
+    raw_config = load_cli_config(config_file_path, **kwargs)
+    return config_class(**raw_config)

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

@@ -1,15 +1,13 @@
 from pathlib import Path
-from typing import Any, cast
+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).
-    """
+    """Load configuration from a file (JSON, YAML, or TOML)."""
     ext = Path(path).suffix.lower()
 
     if ext == '.json':
@@ -24,4 +22,4 @@ def load_config(path: str | Path) -> dict[str, Any]:
     if not isinstance(data, dict):
         raise TypeError(f'Config file {path!r} did not return a dict, got {type(data).__name__}')
 
-    return cast(dict[str, Any], data)
+    return data

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

@@ -1,21 +1,22 @@
 import json
 from pathlib import Path
+from typing import Any
 
+JsonValue = dict[Any, Any] | list[Any] | str | int | float | bool | None
 
-def load_json(path: str | Path) -> dict | list:
+
+def load_json(path: str | Path) -> JsonValue:
     with Path(path).open(mode='r', encoding='utf-8') as fin:
-        data = json.load(fin)
-    return data
+        return json.load(fin)
 
 
 def save_as_indented_json(
-    data: dict | list,
+    data: JsonValue,
     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:
+    target = Path(path)
+    target.parent.mkdir(parents=parents, exist_ok=exist_ok)
+    with target.open(mode='w', encoding='utf-8') as fout:
         json.dump(data, fout, ensure_ascii=False, indent=4, separators=(',', ': '))
-    return

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

@@ -3,23 +3,19 @@
 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 load_jsonlines(path: str | Path) -> list[dict[str, object]]:
+    with jsonlines.open(str(Path(path))) as reader:
+        return [dict(entry) for entry in reader]
 
 
 def save_as_jsonlines(
-    data: list[dict],
+    data: list[dict[str, object]],
     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:
+    target = Path(path)
+    target.parent.mkdir(parents=parents, exist_ok=exist_ok)
+    with jsonlines.open(str(target), mode='w') as writer:
         for datum in data:
             writer.write(datum)
-    return

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

@@ -1,22 +1,21 @@
 from pathlib import Path
+from typing import Any
 
 import toml
 
 
-def load_toml(path: str | Path) -> dict:
+def load_toml(path: str | Path) -> dict[str, Any]:
     with Path(path).open(mode='r', encoding='utf-8') as fin:
-        data = toml.load(fin)
-    return data
+        return toml.load(fin)
 
 
 def save_as_toml(
-    data: dict,
+    data: dict[str, Any],
     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:
+    target = Path(path)
+    target.parent.mkdir(parents=parents, exist_ok=exist_ok)
+    with target.open(mode='w', encoding='utf-8') as fout:
         toml.dump(data, fout)
-    return

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

@@ -1,22 +1,23 @@
 from pathlib import Path
+from typing import Any
 
 import yaml
 
+YamlValue = dict[str, Any] | list[Any] | str | int | float | bool | None
 
-def load_yaml(path: str | Path) -> dict | list:
+
+def load_yaml(path: str | Path) -> YamlValue:
     with Path(path).open(mode='r', encoding='utf-8') as fin:
-        data = yaml.safe_load(fin)
-    return data
+        return yaml.safe_load(fin)
 
 
 def save_as_indented_yaml(
-    data: dict | list,
+    data: YamlValue,
     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:
+    target = Path(path)
+    target.parent.mkdir(parents=parents, exist_ok=exist_ok)
+    with target.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/import_utils.py

@@ -1,19 +1,19 @@
 import importlib
 import inspect
 import sys
+from collections.abc import Callable
 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
+    resolved_path: Path = Path(function_file_path).resolve()
+    function_name = function_name or resolved_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_path = '.'.join(resolved_path.relative_to(project_root).with_suffix('').parts)
     module = importlib.import_module(module_path)
     return getattr(module, function_name)