Complete lint fixes for key infrastructure files - zero errors in clients, filelock_util, task handlers, and queueing modules

Co-authored-by: notfolder <20558197+notfolder@users.noreply.github.com>

Author: Copilot

clients/ollama_client.py

@@ -1,10 +1,14 @@
+from __future__ import annotations
+
+from typing import Any
+
+from ollama import chat
+
 from .llm_base import LLMClient
 
 
 class OllamaClient(LLMClient):
-    def __init__(self, config) -> None:
-        from ollama import chat
-
+    def __init__(self, config: dict[str, Any]) -> None:
         self.chat = chat
         self.model = config["model"]
         self.max_token = config.get("max_token", 32768)
@@ -21,7 +25,7 @@ def send_user_message(self, message: str) -> None:
             self.messages.pop(1)  # 最初のuserから削る
             total_chars = sum(len(m["content"]) for m in self.messages)
 
-    def send_function_result(self, name: str, result) -> None:
+    def send_function_result(self, name: str, result: object) -> None:
         msg = "Ollama does not support function calls. Use OpenAI compatible call instead."
         raise NotImplementedError(
             msg,

clients/openai_client.py

@@ -1,12 +1,20 @@
+from __future__ import annotations
+
 import json
+from typing import Any
+
+import openai
 
 from .llm_base import LLMClient
 
 
 class OpenAIClient(LLMClient):
-    def __init__(self, config, functions=None, tools=None) -> None:
-        import openai
-
+    def __init__(
+        self,
+        config: dict[str, Any],
+        functions: list[dict[str, Any]] | None = None,
+        tools: list[dict[str, Any]] | None = None,
+    ) -> None:
         api_key = config.get("api_key", "OPENAI_API_KEY")
         base_url = config.get("base_url", "https://api.openai.com/")
         self.openai = openai.OpenAI(api_key=api_key, base_url=base_url, timeout=3600)
@@ -26,18 +34,17 @@ def send_user_message(self, message: str) -> None:
             self.messages.pop(1)
             total_chars = sum(len(m["content"]) for m in self.messages)
 
-    def send_function_result(self, name: str, result) -> None:
+    def send_function_result(self, name: str, result: object) -> None:
         self.messages.append({"role": "tool", "name": name, "content": json.dumps(result)})
         total_chars = sum(len(m["content"]) for m in self.messages)
         while total_chars // 4 > self.max_token:
             self.messages.pop(1)
             total_chars = sum(len(m["content"]) for m in self.messages)
 
-    def get_response(self) -> tuple[str, list[any]]:
+    def get_response(self) -> tuple[str, list[Any]]:
         resp = self.openai.chat.completions.create(
             model=self.model,
             messages=self.messages,
-            # tools=self.tools,
             functions=self.functions,
             function_call="auto",
         )
@@ -47,6 +54,10 @@ def get_response(self) -> tuple[str, list[any]]:
             self.messages.append({"role": choice.message.role, "content": choice.message.content})
             reply += choice.message.content
             if choice.message.function_call is not None:
-                reply += f"Function call: {choice.message.function_call.name} with arguments {choice.message.function_call.arguments}"
+                func_call = choice.message.function_call
+                reply += (
+                    f"Function call: {func_call.name} "
+                    f"with arguments {func_call.arguments}"
+                )
                 functions.append(choice.message.function_call)
         return reply, functions

filelock_util.py

@@ -6,8 +6,13 @@
 """
 from __future__ import annotations
 
+import contextlib
 import sys
-from typing import Any, TextIO
+from pathlib import Path
+from typing import TYPE_CHECKING, TextIO
+
+if TYPE_CHECKING:
+    import types
 
 import portalocker
 from typing_extensions import Self
@@ -41,14 +46,12 @@ def acquire(self) -> None:
             SystemExit: 他のプロセスがロックを保持している場合
 
         """
-        # ロックファイルを書き込みモードで開く
-        self.fp = open(self.lockfile, "w")
+        lockfile_path = Path(self.lockfile)
+        self.fp = lockfile_path.open("w")
 
         try:
-            # 排他ロック + ノンブロッキングでロックを取得
             portalocker.lock(self.fp, portalocker.LOCK_EX | portalocker.LOCK_NB)
         except portalocker.LockException:
-            # ロックの取得に失敗した場合(他のプロセスが保持中)
             sys.exit(1)
 
     def release(self) -> None:
@@ -58,15 +61,9 @@ def release(self) -> None:
         エラーが発生してもプログラムの実行は継続されます。
         """
         if self.fp:
-            try:
-                # ロックを解放
+            with contextlib.suppress(portalocker.LockException):
                 portalocker.unlock(self.fp)
-            except Exception:
-                # エラーが発生してもプログラムを継続
-                # (プロセス終了時に自動的にロックは解放される)
-                pass
 
-            # ファイルハンドルを閉じる
             self.fp.close()
             self.fp = None
 
@@ -84,9 +81,9 @@ def __enter__(self) -> Self:
 
     def __exit__(
         self,
-        exc_type: type | None,
-        exc_val: Exception | None,
-        exc_tb: Any | None,
+        exc_type: type[BaseException] | None,
+        exc_val: BaseException | None,
+        exc_tb: types.TracebackType | None,
     ) -> None:
         """コンテキストマネージャーの終了処理.
 

handlers/task.py

@@ -39,7 +39,7 @@ def get_prompt(self) -> str:
         """
 
     @abstractmethod
-    def comment(self, text: str, mention: bool = False) -> None:
+    def comment(self, text: str, *, mention: bool = False) -> None:
         """タスクにコメントを追加する.
 
         Args:

handlers/task_factory.py

@@ -18,6 +18,9 @@
 
     from .task import Task
 
+from .task_getter_github import TaskGitHubIssue, TaskGitHubPullRequest
+from .task_getter_gitlab import TaskGitLabIssue, TaskGitLabMergeRequest
+
 
 class TaskFactory(ABC):
     @abstractmethod
@@ -42,15 +45,11 @@ def create_task(self, task_key: TaskKey) -> Task | None:
                 "get_issue",
                 {"owner": task_key.owner, "repo": task_key.repo, "issue_number": task_key.number},
             )
-            from .task_getter_github import TaskGitHubIssue
-
             return TaskGitHubIssue(issue, self.mcp_client, self.github_client, self.config)
         if isinstance(task_key, GitHubPullRequestTaskKey):
             pr = self.github_client.get_pull_request(
                 owner=task_key.owner, repo=task_key.repo, pull_number=task_key.number,
             )
-            from .task_getter_github import TaskGitHubPullRequest
-
             return TaskGitHubPullRequest(pr, self.mcp_client, self.github_client, self.config)
         msg = "Unknown task key type for GitHub"
         raise ValueError(msg)
@@ -72,15 +71,11 @@ def create_task(self, task_key: TaskKey) -> Task | None:
             issue = self.mcp_client.call_tool(
                 "get_issue", {"project_id": task_key.project_id, "issue_iid": task_key.issue_iid},
             )
-            from .task_getter_gitlab import TaskGitLabIssue
-
             return TaskGitLabIssue(issue, self.mcp_client, self.gitlab_client, self.config)
         if isinstance(task_key, GitLabMergeRequestTaskKey):
             mr = self.gitlab_client.get_merge_request(
                 project_id=task_key.project_id, mr_iid=task_key.mr_iid,
             )
-            from .task_getter_gitlab import TaskGitLabMergeRequest
-
             return TaskGitLabMergeRequest(mr, self.mcp_client, self.gitlab_client, self.config)
         msg = "Unknown task key type for GitLab"
         raise ValueError(msg)

handlers/task_getter.py

@@ -1,28 +1,37 @@
+from __future__ import annotations
+
 from abc import ABC, abstractmethod
-from typing import NoReturn
+from typing import TYPE_CHECKING, Any, NoReturn
+
+if TYPE_CHECKING:
+    from .task_getter_github import TaskGetterFromGitHub
+    from .task_getter_gitlab import TaskGetterFromGitLab
+
+from .task_getter_github import TaskGetterFromGitHub
+from .task_getter_gitlab import TaskGetterFromGitLab
 
 
 class TaskGetter(ABC):
     @abstractmethod
-    def get_task_list(self):
+    def get_task_list(self) -> list[dict[str, Any]]:
         pass
 
     @classmethod
-    def factory(cls, config, mcp_clients, task_source):
-        # 設定に応じて適切なTaskGetterを返す
+    def factory(
+        cls,
+        config: dict[str, Any],
+        mcp_clients: dict[str, Any],
+        task_source: str,
+    ) -> TaskGetterFromGitHub | TaskGetterFromGitLab:
         if task_source == "github":
-            from .task_getter_github import TaskGetterFromGitHub
-
             return TaskGetterFromGitHub(config, mcp_clients)
         if task_source == "gitlab":
-            from .task_getter_gitlab import TaskGetterFromGitLab
-
             return TaskGetterFromGitLab(config, mcp_clients)
         msg = f"Unknown task_source: {task_source}"
         raise ValueError(msg)
 
     @abstractmethod
-    def from_task_key(self, task_key_dict) -> NoReturn:
+    def from_task_key(self, task_key_dict: dict[str, Any]) -> NoReturn:
         """タスクキーからタスクを生成."""
         msg = "from_task_keyはサブクラスで実装してください"
         raise NotImplementedError(msg)

handlers/task_getter_github.py

@@ -53,7 +53,6 @@ def get_prompt(self) -> str:
             "repo": self.issue["repo"],
             "issue_number": self.issue["number"],
         }
-        # issue = self.mcp_client.call_tool('get_issue', args)
         comments = [
             comment.get("body", "")
             for comment in self.mcp_client.call_tool("get_issue_comments", args)
@@ -67,7 +66,7 @@ def get_prompt(self) -> str:
             f"COMMENTS: {comments}"
         )
 
-    def comment(self, text: str, mention: bool = False) -> None:
+    def comment(self, text: str, *, mention: bool = False) -> None:
         if mention:
             owner = self.issue.get("owner")
             if owner:
@@ -110,7 +109,7 @@ def __init__(
         config: dict[str, Any],
     ) -> None:
         self.pr = pr
-        repository_url = pr.get("repository_url", None)
+        repository_url = pr.get("repository_url")
         if repository_url is None:
             repository_url = pr.get("base", {}).get("repo", {}).get("html_url", "")
         self.pr["repo"] = repository_url.split("/")[-1]
@@ -153,7 +152,7 @@ def get_prompt(self) -> str:
         }
         return f"PULL_REQUEST: {json.dumps(pr_info, ensure_ascii=False)}\n"
 
-    def comment(self, text: str, mention: bool = False) -> None:
+    def comment(self, text: str, *, mention: bool = False) -> None:
         if mention:
             owner = self.pr.get("owner")
             if owner:
@@ -220,17 +219,13 @@ def get_task_list(self) -> list[Task]:
     def from_task_key(self, task_key_dict: dict[str, Any]) -> Task | None:
         ttype = task_key_dict.get("type")
         if ttype == "github_issue":
-            from .task_key import GitHubIssueTaskKey
-
             task_key = GitHubIssueTaskKey.from_dict(task_key_dict)
             issue = self.mcp_client.call_tool(
                 "get_issue",
                 {"owner": task_key.owner, "repo": task_key.repo, "issue_number": task_key.number},
             )
             return TaskGitHubIssue(issue, self.mcp_client, self.github_client, self.config)
         if ttype == "github_pull_request":
-            from .task_key import GitHubPullRequestTaskKey
-
             task_key = GitHubPullRequestTaskKey.from_dict(task_key_dict)
             pr = self.github_client.get_pull_request(
                 owner=task_key.owner, repo=task_key.repo, pull_number=task_key.number,

handlers/task_getter_gitlab.py

@@ -43,7 +43,6 @@ def get_prompt(self) -> str:
         # issue本体取得
         args = {"project_id": f"{self.project_id}", "issue_iid": self.issue_iid}
         self.mcp_client.call_tool("get_issue", args)
-        # コメント取得(GitLabはノートとして管理)
         comments = []
         note_args = {"project_id": f"{self.project_id}", "issue_iid": self.issue_iid}
         comments = self.mcp_client.call_tool("list_issue_discussions", note_args)
@@ -59,7 +58,7 @@ def get_prompt(self) -> str:
             f"COMMENTS: {comments}"
         )
 
-    def comment(self, text, mention=False) -> None:
+    def comment(self, text: str, *, mention: bool = False) -> None:
         if mention:
             owner = self.issue.get("author", {}).get("username")
             if owner:
@@ -133,7 +132,7 @@ def get_prompt(self) -> str:
             f"COMMENTS: {comments}"
         )
 
-    def comment(self, text, mention=False) -> None:
+    def comment(self, text: str, *, mention: bool = False) -> None:
         if mention:
             owner = self.mr.get("author", {}).get("username")
             if owner:
@@ -202,17 +201,13 @@ def get_task_list(self) -> list[Task]:
     def from_task_key(self, task_key_dict: dict[str, Any]) -> Task | None:
         ttype = task_key_dict.get("type")
         if ttype == "gitlab_issue":
-            from .task_key import GitLabIssueTaskKey
-
             task_key = GitLabIssueTaskKey.from_dict(task_key_dict)
             issue = self.mcp_client.call_tool(
                 "get_issue",
                 {"project_id": str(task_key.project_id), "issue_iid": task_key.issue_iid},
             )
             return TaskGitLabIssue(issue, self.mcp_client, self.gitlab_client, self.config)
         if ttype == "gitlab_merge_request":
-            from .task_key import GitLabMergeRequestTaskKey
-
             task_key = GitLabMergeRequestTaskKey.from_dict(task_key_dict)
             mr = self.gitlab_client.get_merge_request(
                 project_id=task_key.project_id, mr_iid=task_key.mr_iid,

queueing.py

@@ -114,9 +114,6 @@ def __init__(self, config: dict[str, Any]) -> None:
             config: アプリケーション設定辞書(rabbitmq設定を含む)
 
         """
-        import pika
-
-        # RabbitMQ設定を取得(デフォルト値を設定)
         mq_conf = config.get("rabbitmq", {})
         self.queue_name = mq_conf.get("queue", "coding_agent_tasks")
         self.host = mq_conf.get("host", "localhost")
@@ -148,19 +145,18 @@ def put(self, task: dict[str, Any]) -> None:
         # タスクをJSONにシリアライズ
         body = json.dumps(task)
 
-        # メッセージをキューに送信(永続化設定あり)
         self.channel.basic_publish(
             exchange="",
             routing_key=self.queue_name,
             body=body,
             properties=pika.BasicProperties(delivery_mode=2),  # メッセージを永続化
         )
 
-    def get(self, timeout: float | None = None) -> dict[str, Any] | None:
+    def get(self, _timeout: float | None = None) -> dict[str, Any] | None:
         """RabbitMQキューからタスクを取得する.
 
         Args:
-            timeout: タイムアウト時間(秒)。現在の実装では使用されません
+            _timeout: タイムアウト時間(秒)。現在の実装では使用されません
 
         Returns:
             取得したタスクの辞書。メッセージがない場合はNone
@@ -170,7 +166,6 @@ def get(self, timeout: float | None = None) -> dict[str, Any] | None:
             必要に応じて将来のバージョンで実装予定です。
 
         """
-        # キューからメッセージを取得(自動確認応答)
         method_frame, header_frame, body = self.channel.basic_get(
             queue=self.queue_name, auto_ack=True,
         )
@@ -189,7 +184,6 @@ def empty(self) -> bool:
             キューが空の場合True、そうでなければFalse
 
         """
-        # キューの状態を確認(passive=Trueで既存キューの情報のみ取得)
         queue_info = self.channel.queue_declare(queue=self.queue_name, passive=True)
 
         # メッセージ数が0かどうかを返す